From d6941d4b189bbfb6397fab5bbc2c85b16bad9876 Mon Sep 17 00:00:00 2001 From: inesaranab Date: Tue, 11 Aug 2026 20:41:08 +0200 Subject: [PATCH] fix: read the etag where the SDK keeps it, and say what the probe is doing The first production run of the readiness change failed with "IfNotModified must be specified with etag". The etag is metadata on a table entity, not one of its properties, so copying the entity into a plain dict dropped it and the conditional write was sent without one. The test double had put the etag among the properties, so the tests passed against a client that behaves differently from the real one. It now mirrors the SDK, and the test that covers this fails without the fix. The readiness loop logged nothing, so fifteen minutes of polling and a hang were indistinguishable in the logs. Each attempt is now recorded with how long it has waited and why the endpoint was not usable. Raises the probe timeout from 30 to 180 seconds. The platform holds a request open while it starts a replica for an app scaled to zero, so a probe that gives up in thirty seconds can abandon that start before a replica exists -- the endpoint is then never reached however often it is retried. 180 stays below the 240-second limit at which ingress severs any request. The per-job budget is now asserted by a test rather than described in a comment. --- app/adapters/detector_readiness.py | 18 ++++++++++++-- app/adapters/job_store_table.py | 11 ++++++--- app/worker.py | 13 +++++++++- tests/unit/test_detector_readiness.py | 34 +++++++++++++++++++++++++++ tests/unit/test_job_store_table.py | 18 +++++++++++--- tests/unit/test_worker.py | 24 +++++++++++++++++++ 6 files changed, 109 insertions(+), 9 deletions(-) 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