From 38ef0587f5f6c55086ba330d4a94601bfc8fdea2 Mon Sep 17 00:00:00 2001 From: AdrianoDev Date: Sun, 16 Aug 2026 19:38:56 +0200 Subject: [PATCH] fix(vision): il guasto del worker resta suo, il gate anticipa il worker MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit I2: _call_worker lasciava propagare httpx.ConnectError e ReadTimeout (worker fermo o job oltre i 120s), senza handler in main.py: l'operatore vedeva un 500 nudo. Ora ConnectError->502, ReadTimeout->504, ciascuno con un messaggio che nomina il guasto. La forma della risposta è validata alla frontiera: outputs/failures/engine_version mancanti o un body non- dict non fanno più KeyError, rispondono 502 "malformed". M3: il gate del fuori tolleranza (pending_authorisation) girava dopo _call_worker: un operatore bloccato bruciava un'intera esecuzione di visione prima di ricevere il 409. Spostato prima della chiamata al worker; le quote necessarie erano già caricate prima, nessun'altra query serviva. Copertura: test_vision_worker_transport.py (nuovo) per i quattro casi di guasto/malformazione; test_vision_execute.py aggiunge test_a_pending_fail_blocks_before_the_worker_is_called, rossa prima dello spostamento del gate. Co-Authored-By: Claude Opus 5 (1M context) Claude-Session: https://claude.ai/code/session_014BBnuACZSCJqXrMYC3LUMU --- src/backend/services/vision_service.py | 109 +++++++++++----- src/backend/tests/test_vision_execute.py | 39 ++++++ .../tests/test_vision_worker_transport.py | 117 ++++++++++++++++++ 3 files changed, 235 insertions(+), 30 deletions(-) create mode 100644 src/backend/tests/test_vision_worker_transport.py diff --git a/src/backend/services/vision_service.py b/src/backend/services/vision_service.py index 74fbad8..99a5171 100644 --- a/src/backend/services/vision_service.py +++ b/src/backend/services/vision_service.py @@ -22,19 +22,64 @@ from src.backend.services import measurement_service async def _call_worker(image_bytes: bytes, graph: dict) -> dict: - """The one place that talks to the worker; the tests replace it.""" - async with httpx.AsyncClient(timeout=120.0) as client: - response = await client.post( - f"{settings.vision_worker_url}/run", - files={"image": ("frame.png", image_bytes, "image/png")}, - data={"graph": json.dumps(graph)}, - ) + """The one place that talks to the worker; the tests replace it. + + The worker's fault must not become the API's fault (spec's edge-case + table: "il worker isola il guasto, l'API resta in piedi, il task + riporta l'errore"). A stopped worker raises `httpx.ConnectError`; a job + past the timeout raises `httpx.ReadTimeout` - both would otherwise + propagate past this function and surface to the operator as a naked + 500. The response shape is validated here too: a proxy answering 200 + with something that is not the worker's JSON contract must not turn + into a bare KeyError further down. + """ + try: + async with httpx.AsyncClient(timeout=120.0) as client: + response = await client.post( + f"{settings.vision_worker_url}/run", + files={"image": ("frame.png", image_bytes, "image/png")}, + data={"graph": json.dumps(graph)}, + ) + except httpx.ConnectError as exc: + raise HTTPException( + status_code=status.HTTP_502_BAD_GATEWAY, + detail=f"the vision worker is unreachable: {exc}", + ) from exc + except httpx.ReadTimeout as exc: + raise HTTPException( + status_code=status.HTTP_504_GATEWAY_TIMEOUT, + detail=f"the vision worker did not answer in time: {exc}", + ) from exc + if response.status_code != 200: raise HTTPException( status_code=status.HTTP_502_BAD_GATEWAY, detail=f"the vision worker refused the job: {response.text}", ) - return response.json() + + try: + report = response.json() + except ValueError as exc: + raise HTTPException( + status_code=status.HTTP_502_BAD_GATEWAY, + detail=f"the vision worker's answer was malformed: not valid JSON ({exc})", + ) from exc + + if not isinstance(report, dict): + raise HTTPException( + status_code=status.HTTP_502_BAD_GATEWAY, + detail="the vision worker's answer was malformed: expected a JSON object", + ) + missing = {"outputs", "failures", "engine_version"} - report.keys() + if missing: + raise HTTPException( + status_code=status.HTTP_502_BAD_GATEWAY, + detail=( + "the vision worker's answer was malformed: missing " + f"{', '.join(sorted(missing))}" + ), + ) + return report async def execute_task( @@ -70,6 +115,32 @@ async def execute_task( detail="no quote on this task names a vision output", ) + # Ruling R10: the gate is evaluated once per acquisition, not once per + # quote. `save_measurement`'s own gate models "moving on" between + # measurements taken one after another - a real later moment where the + # operator could leave a bad reading behind. A camera acquisition produces + # every quote of this task at the same instant: there is no "moving on" + # between them, so applying the sequential rule inside the batch would + # block on a fail this very request just saved. Check it once here, + # against the whole set this acquisition is about to produce, and pass + # enforce_tolerance_gate=False to every save_measurement call below. + # + # Checked before the worker runs (M3): a blocked operator finds out + # without burning a whole vision execution first - the worker is the + # expensive part, the gate check is not. + blocking = await measurement_service.pending_authorisation( + db, version_id, measured_by, production_run_id, + ) + quote_ids = {q.id for q in quotes} + if blocking is not None and blocking.subtask_id not in quote_ids: + raise HTTPException( + status_code=status.HTTP_409_CONFLICT, + detail=( + "A measurement is out of tolerance and awaiting the supervisor: " + "authorise it or measure that quote again before going on" + ), + ) + report = await _call_worker(image_bytes, task.vision_json) outputs = report["outputs"] @@ -86,28 +157,6 @@ async def execute_task( ), ) - # Ruling R10: the gate is evaluated once per acquisition, not once per - # quote. `save_measurement`'s own gate models "moving on" between - # measurements taken one after another - a real later moment where the - # operator could leave a bad reading behind. A camera acquisition produces - # every quote of this task at the same instant: there is no "moving on" - # between them, so applying the sequential rule inside the batch would - # block on a fail this very request just saved. Check it once here, - # against the whole set this acquisition is about to produce, and pass - # enforce_tolerance_gate=False to every save_measurement call below. - blocking = await measurement_service.pending_authorisation( - db, version_id, measured_by, production_run_id, - ) - quote_ids = {q.id for q in quotes} - if blocking is not None and blocking.subtask_id not in quote_ids: - raise HTTPException( - status_code=status.HTTP_409_CONFLICT, - detail=( - "A measurement is out of tolerance and awaiting the supervisor: " - "authorise it or measure that quote again before going on" - ), - ) - result = VisionResult( task_id=task.id, engine_version=report["engine_version"], diff --git a/src/backend/tests/test_vision_execute.py b/src/backend/tests/test_vision_execute.py index 695da9d..5eb1547 100644 --- a/src/backend/tests/test_vision_execute.py +++ b/src/backend/tests/test_vision_execute.py @@ -250,6 +250,45 @@ async def test_a_pending_fail_outside_the_batch_still_blocks( )).scalars().all() == [] +@pytest.mark.asyncio +async def test_a_pending_fail_blocks_before_the_worker_is_called( + client, db_session, measurement_tec_user, monkeypatch, +): + """M3: a blocked operator must not burn a whole vision execution to learn + they are blocked. The gate is checked before the worker runs, not after.""" + recipe = await create_test_recipe(db_session, measurement_tec_user.id) + version, task, first, second = await two_vision_quotes(db_session, recipe.id) + + third = RecipeSubtask( + task_id=task.id, marker_number=3, description="Not a vision quote", + nominal=10.0, utl=10.5, uwl=10.3, lwl=9.7, ltl=9.5, unit="mm", + ) + db_session.add(third) + await db_session.commit() + await db_session.refresh(third) + await measurement_service.save_measurement( + db_session, subtask_id=third.id, version_id=version.id, + measured_by=measurement_tec_user.id, value=OUT, + ) + await db_session.commit() + + async def worker_must_not_be_called(image_bytes: bytes, graph: dict) -> dict: + raise AssertionError("the worker must not run while the operator is blocked") + monkeypatch.setattr( + "src.backend.services.vision_service._call_worker", worker_must_not_be_called, + ) + + response = await client.post( + "/api/vision/execute", + headers=auth_headers(measurement_tec_user), + files={"image": ("p.png", an_image(), "image/png")}, + data={"task_id": str(task.id)}, + ) + + assert response.status_code == 409 + assert "supervisor" in response.json()["detail"].lower() + + @pytest.mark.asyncio async def test_a_new_acquisition_can_remeasure_its_own_pending_quote( client, db_session, measurement_tec_user, worker_says, diff --git a/src/backend/tests/test_vision_worker_transport.py b/src/backend/tests/test_vision_worker_transport.py new file mode 100644 index 0000000..0365c5b --- /dev/null +++ b/src/backend/tests/test_vision_worker_transport.py @@ -0,0 +1,117 @@ +"""I2: a broken worker must answer with a named status, never a naked 500. + +`_call_worker` mapped a non-200 HTTP response to 502, but let +`httpx.ConnectError` (worker stopped) and `httpx.ReadTimeout` (job past the +120s budget) propagate unhandled - with no exception handler in main.py that +turns those into anything but FastAPI's generic 500. It also read +`report["outputs"]`, `report["failures"]` and `report["engine_version"]` +without checking the response had that shape at all, so a proxy answering 200 +with something else (HTML, an empty body) would raise a bare KeyError instead +of a diagnosable error. + +The spec's edge-case table: "il worker isola il guasto, l'API resta in +piedi, il task riporta l'errore" - the worker's fault must not become the +API's fault. +""" +import httpx +import pytest +from fastapi import HTTPException + +from src.backend.services.vision_service import _call_worker + + +class _FakeResponse: + """Stands in for httpx.Response: only what `_call_worker` touches.""" + + def __init__(self, status_code: int, body=..., text: str = "", raise_on_json: bool = False): + self.status_code = status_code + self._body = body + self.text = text + self._raise_on_json = raise_on_json + + def json(self): + if self._raise_on_json: + raise ValueError("not JSON") + return self._body + + +class _FakeAsyncClient: + """Stands in for httpx.AsyncClient: post() does whatever the test wants.""" + + def __init__(self, *, response=None, raises=None): + self._response = response + self._raises = raises + + async def __aenter__(self): + return self + + async def __aexit__(self, *exc_info): + return False + + async def post(self, *args, **kwargs): + if self._raises is not None: + raise self._raises + return self._response + + +def _install(monkeypatch, *, response=None, raises=None): + monkeypatch.setattr( + "src.backend.services.vision_service.httpx.AsyncClient", + lambda *a, **k: _FakeAsyncClient(response=response, raises=raises), + ) + + +@pytest.mark.asyncio +async def test_a_stopped_worker_answers_502_not_a_naked_500(monkeypatch): + _install(monkeypatch, raises=httpx.ConnectError("Connection refused")) + + with pytest.raises(HTTPException) as excinfo: + await _call_worker(b"img", {"schema_version": 1, "tools": []}) + + assert excinfo.value.status_code == 502 + assert "unreachable" in excinfo.value.detail.lower() + + +@pytest.mark.asyncio +async def test_a_job_past_the_time_budget_answers_504(monkeypatch): + _install(monkeypatch, raises=httpx.ReadTimeout("timed out")) + + with pytest.raises(HTTPException) as excinfo: + await _call_worker(b"img", {"schema_version": 1, "tools": []}) + + assert excinfo.value.status_code == 504 + assert "time" in excinfo.value.detail.lower() + + +@pytest.mark.asyncio +async def test_a_response_missing_the_expected_keys_is_reported_not_raised_raw(monkeypatch): + """A proxy answering 200 with an unexpected shape must not KeyError.""" + _install(monkeypatch, response=_FakeResponse(200, body={"outputs": {}})) + + with pytest.raises(HTTPException) as excinfo: + await _call_worker(b"img", {"schema_version": 1, "tools": []}) + + assert excinfo.value.status_code == 502 + assert "malformed" in excinfo.value.detail.lower() + + +@pytest.mark.asyncio +async def test_a_response_that_is_not_a_json_object_is_reported(monkeypatch): + """200 with HTML or a bare list, say - not a dict at all.""" + _install(monkeypatch, response=_FakeResponse(200, body=["not", "a", "dict"])) + + with pytest.raises(HTTPException) as excinfo: + await _call_worker(b"img", {"schema_version": 1, "tools": []}) + + assert excinfo.value.status_code == 502 + assert "malformed" in excinfo.value.detail.lower() + + +@pytest.mark.asyncio +async def test_a_well_formed_response_still_passes_through(monkeypatch): + body = {"outputs": {"t1.diameter": 10.0}, "failures": [], "engine_version": "a" * 40} + _install(monkeypatch, response=_FakeResponse(200, body=body)) + + report = await _call_worker(b"img", {"schema_version": 1, "tools": []}) + + assert report == body