fix(vision): /run del worker non blocca più l'event loop, timeout e semaforo (#34)

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 <noreply@anthropic.com>
This commit is contained in:
2026-09-21 22:39:28 +02:00
parent b024724c7e
commit b15037851b
3 changed files with 166 additions and 30 deletions
+75
View File
@@ -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