From b15037851b47136d7f0ca1b64290f91164440bbc Mon Sep 17 00:00:00 2001 From: AdrianoDev Date: Mon, 21 Sep 2026 22:39:28 +0200 Subject: [PATCH] =?UTF-8?q?fix(vision):=20/run=20del=20worker=20non=20bloc?= =?UTF-8?q?ca=20pi=C3=B9=20l'event=20loop,=20timeout=20e=20semaforo=20(#34?= =?UTF-8?q?)?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Decodifica + run_graph girano in asyncio.to_thread sotto asyncio.wait_for (VISION_RUN_TIMEOUT_S, default 100s, sotto i 120s del backend) -> 504 del worker allo scadere. /health resta leggero e risponde durante un calcolo in corso. asyncio.Semaphore(VISION_MAX_CONCURRENCY, default 2): oltre -> 503 "worker busy". vision_service._call_worker inoltra 422/503/504 del worker con il loro status e motivo invece di appiattirli su 502. Co-Authored-By: Claude Sonnet 5 --- src/backend/services/vision_service.py | 23 ++++-- src/vision_worker/main.py | 98 ++++++++++++++++++++------ src/vision_worker/tests/test_worker.py | 75 ++++++++++++++++++++ 3 files changed, 166 insertions(+), 30 deletions(-) diff --git a/src/backend/services/vision_service.py b/src/backend/services/vision_service.py index 1836015..20b7c72 100644 --- a/src/backend/services/vision_service.py +++ b/src/backend/services/vision_service.py @@ -21,6 +21,14 @@ from src.backend.models.orm.task import RecipeSubtask, RecipeTask from src.backend.models.orm.vision import VisionReferenceImage, VisionResult from src.backend.services import measurement_service +# Worker responses whose status and reason are forwarded as-is rather than +# flattened into a generic 502 - see `_call_worker`. +_FORWARDED_WORKER_STATUSES = frozenset({ + status.HTTP_422_UNPROCESSABLE_CONTENT, + status.HTTP_503_SERVICE_UNAVAILABLE, + status.HTTP_504_GATEWAY_TIMEOUT, +}) + def _is_measurable(value: object) -> bool: """A worker output counts as a quote only if it is a finite, non-bool number. @@ -69,17 +77,18 @@ async def _call_worker(image_bytes: bytes, graph: dict) -> dict: detail=f"the vision worker is unreachable: {exc}", ) from exc - if response.status_code == status.HTTP_422_UNPROCESSABLE_CONTENT: - # The worker already diagnosed a bad request (undecodable image, - # malformed graph) - forward its reason instead of flattening it - # into a generic 502. + # The worker already gave this fault a name - forward its status and + # reason instead of flattening every non-200 into a generic 502: 422 is a + # bad request (undecodable image, malformed graph), 503 is "busy, try + # again" (VISION_MAX_CONCURRENCY exceeded), 504 is its own run timeout + # (VISION_RUN_TIMEOUT_S, always tighter than this client's own 120s so + # the worker answers first). + if response.status_code in _FORWARDED_WORKER_STATUSES: try: detail = response.json().get("detail", response.text) except ValueError: detail = response.text - raise HTTPException( - status_code=status.HTTP_422_UNPROCESSABLE_CONTENT, detail=detail, - ) + raise HTTPException(status_code=response.status_code, detail=detail) if response.status_code != 200: raise HTTPException( diff --git a/src/vision_worker/main.py b/src/vision_worker/main.py index 45b7265..d62aa6f 100644 --- a/src/vision_worker/main.py +++ b/src/vision_worker/main.py @@ -6,22 +6,59 @@ crashes does not take the other tablets' requests down with it. """ from __future__ import annotations +import asyncio +import io import json +import os import numpy as np from fastapi import FastAPI, File, Form, HTTPException, UploadFile, status from PIL import Image, UnidentifiedImageError -from src.vision.runner import GraphBuildError, engine_version, run_graph +from src.vision.runner import GraphBuildError, RunOutcome, engine_version, run_graph app = FastAPI(title="TieMeasureFlow Vision Worker", version="0.1.0") +# Below the backend's own 120s budget on purpose (#34): the worker must give +# up and answer 504 before the backend's httpx client times out on its own, +# so the operator sees a reason instead of a bare timeout. +_RUN_TIMEOUT_S = float(os.environ.get("VISION_RUN_TIMEOUT_S", "100")) +# Two slow graphs at once must not also stall a third tablet's request past +# any reasonable wait - reject it instead, with a status the backend can turn +# into a 503 the operator understands ("try again"), not a queued request +# that quietly outlasts anyone's patience. +_MAX_CONCURRENCY = int(os.environ.get("VISION_MAX_CONCURRENCY", "2")) +_run_slots = asyncio.Semaphore(_MAX_CONCURRENCY) + + +class _ImageDecodeError(Exception): + """The upload is not a decodable image - raised inside the worker thread + so it survives the trip through `asyncio.to_thread`.""" + @app.get("/health") async def health() -> dict: + """Deliberately just this: reading `engine_version()` (cached after the + first call) is the only work here, so this answers even while `/run` is + busy decoding an image or running a graph in its own thread.""" return {"status": "ok", "engine_version": engine_version()} +def _decode_and_run(content: bytes, graph: dict) -> RunOutcome: + """The synchronous, CPU/IO-bound half of a request: PIL decode plus + `run_graph`. Runs in a worker thread (see `run` below) so the event loop - + and therefore `/health` - stays responsive while it executes. Once this + thread is past its budget it is not interruptible: it keeps computing to + completion in the background even after `run` has already answered 504 + for it (see the timeout handling there). + """ + try: + frame = np.array(Image.open(io.BytesIO(content)).convert("L")) + except (UnidentifiedImageError, OSError) as exc: + raise _ImageDecodeError(str(exc)) from exc + return run_graph(frame, graph) + + @app.post("/run") async def run( image: UploadFile = File(...), @@ -41,30 +78,45 @@ async def run( detail=f"graph must be a JSON object, got {type(parsed_graph).__name__}", ) - try: - frame = np.array(Image.open(image.file).convert("L")) - except (UnidentifiedImageError, OSError) as exc: - raise HTTPException( - status_code=status.HTTP_422_UNPROCESSABLE_CONTENT, - detail=f"image is not decodable: {exc}", - ) from exc + content = await image.read() - try: - outcome = run_graph(frame, parsed_graph) - except GraphBuildError as exc: - # The graph itself is bad (schema version, a tool missing "kind", ...) - - # a bad request, not a server fault. + if _run_slots.locked(): raise HTTPException( - status_code=status.HTTP_422_UNPROCESSABLE_CONTENT, - detail=str(exc), - ) from exc - except Exception as exc: - # The graph built fine; something failed while actually running it - - # an engine fault, not the caller's. Still named, not a naked 500. - raise HTTPException( - status_code=status.HTTP_500_INTERNAL_SERVER_ERROR, - detail=f"vision run failed: {type(exc).__name__}: {exc}", - ) from exc + status_code=status.HTTP_503_SERVICE_UNAVAILABLE, + detail=f"worker busy: already running {_MAX_CONCURRENCY} job(s)", + ) + + async with _run_slots: + try: + outcome = await asyncio.wait_for( + asyncio.to_thread(_decode_and_run, content, parsed_graph), + timeout=_RUN_TIMEOUT_S, + ) + except asyncio.TimeoutError as exc: + raise HTTPException( + status_code=status.HTTP_504_GATEWAY_TIMEOUT, + detail=f"vision run exceeded {_RUN_TIMEOUT_S}s", + ) from exc + except _ImageDecodeError as exc: + raise HTTPException( + status_code=status.HTTP_422_UNPROCESSABLE_CONTENT, + detail=f"image is not decodable: {exc}", + ) from exc + except GraphBuildError as exc: + # The graph itself is bad (schema version, a tool missing + # "kind", ...) - a bad request, not a server fault. + raise HTTPException( + status_code=status.HTTP_422_UNPROCESSABLE_CONTENT, + detail=str(exc), + ) from exc + except Exception as exc: + # The graph built fine; something failed while actually running + # it - an engine fault, not the caller's. Still named, not a + # naked 500. + raise HTTPException( + status_code=status.HTTP_500_INTERNAL_SERVER_ERROR, + detail=f"vision run failed: {type(exc).__name__}: {exc}", + ) from exc return { # `outcome.outputs` never carries NaN/inf/bool (see runner.run_graph): diff --git a/src/vision_worker/tests/test_worker.py b/src/vision_worker/tests/test_worker.py index f63c5cb..3b812c2 100644 --- a/src/vision_worker/tests/test_worker.py +++ b/src/vision_worker/tests/test_worker.py @@ -21,10 +21,15 @@ pytest.importorskip( ), ) +import asyncio +import time + import numpy as np from httpx import ASGITransport, AsyncClient from PIL import Image +from src.vision import runner +from src.vision_worker import main as worker_main from src.vision_worker.main import app @@ -147,3 +152,73 @@ async def test_a_tool_without_a_kind_is_refused_not_crashed(): assert response.status_code == 422 assert "malformed graph" in response.json()["detail"].lower() assert "kind" in response.json()["detail"].lower() + + +# --------------------------------------------------------------------------- +# #34: /run used to execute PIL decode + run_graph synchronously inside the +# event loop, with no timeout - a slow graph blocked every other request, +# /health included, and there was no cap on how many ran at once. +# --------------------------------------------------------------------------- + + +@pytest.mark.asyncio +async def test_a_run_past_its_budget_answers_504_and_health_still_answers( + monkeypatch, +): + monkeypatch.setattr(worker_main, "_RUN_TIMEOUT_S", 0.1) + + def slow_run_graph(frame, graph): + time.sleep(0.5) # off the event loop - it runs inside asyncio.to_thread + return runner.RunOutcome(engine_version="a" * 40) + + monkeypatch.setattr(worker_main, "run_graph", slow_run_graph) + + transport = ASGITransport(app=app) + async with AsyncClient(transport=transport, base_url="http://worker") as client: + run_task = asyncio.create_task(client.post( + "/run", + files={"image": ("piece.png", a_disc_png(), "image/png")}, + data={"graph": json.dumps(a_circle_graph())}, + )) + await asyncio.sleep(0.05) # let /run start and enter the thread + + health_response = await client.get("/health") + assert health_response.status_code == 200 + + run_response = await run_task + + assert run_response.status_code == 504 + assert "0.1" in run_response.json()["detail"] + + +@pytest.mark.asyncio +async def test_a_third_concurrent_run_is_refused_not_queued(monkeypatch): + monkeypatch.setattr(worker_main, "_MAX_CONCURRENCY", 1) + monkeypatch.setattr(worker_main, "_run_slots", asyncio.Semaphore(1)) + monkeypatch.setattr(worker_main, "_RUN_TIMEOUT_S", 2.0) + + def slow_run_graph(frame, graph): + time.sleep(0.3) + return runner.RunOutcome(engine_version="a" * 40) + + monkeypatch.setattr(worker_main, "run_graph", slow_run_graph) + + transport = ASGITransport(app=app) + async with AsyncClient(transport=transport, base_url="http://worker") as client: + first = asyncio.create_task(client.post( + "/run", + files={"image": ("piece.png", a_disc_png(), "image/png")}, + data={"graph": json.dumps(a_circle_graph())}, + )) + await asyncio.sleep(0.05) # let the first request take the only slot + + second_response = await client.post( + "/run", + files={"image": ("piece.png", a_disc_png(), "image/png")}, + data={"graph": json.dumps(a_circle_graph())}, + ) + + first_response = await first + + assert second_response.status_code == 503 + assert first_response.status_code == 200