diff --git a/app/adapters/detector_readiness.py b/app/adapters/detector_readiness.py index 4b1b2e3..d63433a 100644 --- a/app/adapters/detector_readiness.py +++ b/app/adapters/detector_readiness.py @@ -5,10 +5,13 @@ by repeating a short request rather than by holding one open. """ +import logging import time from collections.abc import Awaitable, Callable from typing import Any, Protocol +logger = logging.getLogger("screen") + class HttpClientLike(Protocol): """The subset of an async HTTP client this module uses.""" @@ -62,17 +65,28 @@ async def wait_until_ready( True if the detector became ready, False if the deadline passed first. """ started = now() + attempt = 0 while True: + attempt += 1 + reason = "not_serving" try: ready = await probe() - except Exception: # noqa: BLE001 - any failure means "not ready yet" + except Exception as exc: # noqa: BLE001 - any failure means "not ready yet" # The probe is supplied by the caller, so the ways it can fail are # not knowable here. A detector that has not started refuses the # connection, which is the expected state while it loads rather # than a failure to report. ready = False + reason = type(exc).__name__ + waited = round(now() - started) + context = {"attempt": attempt, "waited_s": waited} if ready: + logger.info("detector_ready", extra={"context": context}) return True - if now() - started + interval_s > deadline_s: + logger.info( + "detector_not_ready_yet", extra={"context": {**context, "reason": reason}} + ) + if waited + interval_s > deadline_s: + logger.error("detector_never_ready", extra={"context": context}) return False await sleep(interval_s) diff --git a/app/adapters/job_store_table.py b/app/adapters/job_store_table.py index 049c2e2..840ac5e 100644 --- a/app/adapters/job_store_table.py +++ b/app/adapters/job_store_table.py @@ -171,10 +171,15 @@ async def fail_if_pending(self, job_id: str, error: str) -> bool: refused rather than to replace the result. """ try: - entity = dict(await self._table.get_entity(job_id, job_id)) + entity = await self._table.get_entity(job_id, job_id) except ResourceNotFoundError: return False - if entity.get("status") != JobStatus.PENDING.value: + if dict(entity).get("status") != JobStatus.PENDING.value: + return False + # The etag is metadata on the entity, not one of its properties, so it + # does not survive being copied into a plain dict. + etag = getattr(entity, "metadata", {}).get("etag") + if etag is None: return False try: await self._table.update_entity( @@ -186,7 +191,7 @@ async def fail_if_pending(self, job_id: str, error: str) -> bool: "error": error, }, mode=UpdateMode.MERGE, - etag=entity.get("etag"), + etag=etag, match_condition=MatchConditions.IfNotModified, ) except ResourceModifiedError: diff --git a/app/worker.py b/app/worker.py index 569cf62..6d924d9 100644 --- a/app/worker.py +++ b/app/worker.py @@ -41,7 +41,18 @@ # happen between requests, never inside one. DETECTOR_READY_DEADLINE_S = 900.0 DETECTOR_PROBE_INTERVAL_S = 15.0 -DETECTOR_PROBE_TIMEOUT_S = 30.0 + +# A probe is an ordinary request, so ingress severs it at 240 seconds like any +# other. It is set well above a healthy endpoint's response time because the +# platform holds the request open while it starts a replica for an app scaled to +# zero: a probe that gives up in seconds can abandon that start before a replica +# exists, and the endpoint is then never reached however often it is retried. +DETECTOR_PROBE_TIMEOUT_S = 180.0 + +# How long the queue hides a message it has handed out. Everything one job can +# wait for has to fit inside it, or the message is redelivered while the +# execution holding it is still working. +VISIBILITY_TIMEOUT_S = 1800.0 async def drain( diff --git a/tests/unit/test_detector_readiness.py b/tests/unit/test_detector_readiness.py index b681ca1..efa9afc 100644 --- a/tests/unit/test_detector_readiness.py +++ b/tests/unit/test_detector_readiness.py @@ -124,3 +124,37 @@ async def sleep(seconds: float) -> None: assert not ready assert clock["t"] <= 100 + 30 assert probes <= 3 + + +@pytest.mark.asyncio +async def test_each_attempt_is_logged(): + """A silent wait cannot be told apart from a hung one. Each attempt is + recorded so the logs show whether the endpoint is being reached at all.""" + import logging + + events: list[str] = [] + + class _Collect(logging.Handler): + def emit(self, record: logging.LogRecord) -> None: + events.append(record.getMessage()) + + handler = _Collect() + logger = logging.getLogger("screen") + logger.addHandler(handler) + previous = logger.level + logger.setLevel(logging.INFO) + answers = iter([False, True]) + + async def probe() -> bool: + return next(answers) + + try: + await wait_until_ready( + probe, deadline_s=60, interval_s=5, sleep=_no_sleep, now=lambda: 0.0 + ) + finally: + logger.removeHandler(handler) + logger.setLevel(previous) + + assert "detector_not_ready_yet" in events + assert "detector_ready" in events diff --git a/tests/unit/test_job_store_table.py b/tests/unit/test_job_store_table.py index 00e63d4..24c9382 100644 --- a/tests/unit/test_job_store_table.py +++ b/tests/unit/test_job_store_table.py @@ -73,6 +73,16 @@ def test_the_result_is_stored_as_one_json_column(): assert json.loads(entity["result"])["assessment"]["fit_score"] == 4 +class _Entity(dict): + """Stands in for ``azure.data.tables.TableEntity``. + + The etag lives in ``metadata``, not among the properties, so a plain dict + would let the adapter appear to read one that is not there. + """ + + metadata: dict + + class _FakeTableClient: """Records what the adapter sends, and answers reads from that record. @@ -82,13 +92,14 @@ class _FakeTableClient: def __init__(self) -> None: self.entities: dict[str, dict] = {} + self.etags: dict[str, str] = {} self._version = 0 self.on_read: Callable[[], Awaitable[None]] | None = None def _stamp(self, row_key: str) -> None: """Give the row a new etag, as any write to it does.""" self._version += 1 - self.entities[row_key]["etag"] = f"W/\"{self._version}\"" + self.etags[row_key] = f'W/"{self._version}"' async def upsert_entity(self, entity: dict, **kwargs) -> None: """Merge into the stored row, as UpdateMode.MERGE does. @@ -113,7 +124,7 @@ async def update_entity(self, entity: dict, **kwargs) -> None: from azure.core.exceptions import ResourceNotFoundError raise ResourceNotFoundError("no such entity") - if kwargs.get("etag") is not None and kwargs["etag"] != stored.get("etag"): + if kwargs.get("etag") != self.etags.get(row_key): raise ResourceModifiedError("etag mismatch") stored.update(entity) self._stamp(row_key) @@ -123,7 +134,8 @@ async def get_entity(self, partition_key: str, row_key: str) -> dict: if row_key not in self.entities: raise ResourceNotFoundError("no such entity") - entity = dict(self.entities[row_key]) + entity = _Entity(self.entities[row_key]) + entity.metadata = {"etag": self.etags.get(row_key)} if self.on_read is not None: # Lets a test interleave another writer between a read and the # write that depends on it. diff --git a/tests/unit/test_worker.py b/tests/unit/test_worker.py index 3a20f93..69fd9cc 100644 --- a/tests/unit/test_worker.py +++ b/tests/unit/test_worker.py @@ -154,3 +154,27 @@ async def never_ready() -> bool: job = await store.get(job_id) assert job is not None assert job.status is JobStatus.FAILED + + +def test_the_per_job_budget_fits_inside_one_delivery(): + """Everything one job can wait for has to fit inside the window the queue + hides its message for. Beyond that the message reappears while the + execution holding it is still running, and the job is done twice.""" + from app.config import settings + from app.worker import ( + DETECTOR_PROBE_TIMEOUT_S, + DETECTOR_READY_DEADLINE_S, + VISIBILITY_TIMEOUT_S, + ) + + budget = ( + DETECTOR_READY_DEADLINE_S + + settings.llm_guardrail_timeout_s + + settings.llm_timeout_s + ) + assert budget < VISIBILITY_TIMEOUT_S + + # A probe is one request, so it is subject to the same ingress limit as any + # other. Long enough for the platform to begin starting a replica, short + # enough that the platform does not sever it first. + assert 60 < DETECTOR_PROBE_TIMEOUT_S < 240