pont: fermeture amont a la deconnexion client, pool httpx 512 + timeout de pool (fuite de flux -> pont muet)

anthropic_proxy.py

@@ -57,7 +57,16 @@ ALIASES = [a for pair in ((x, f"{x}[1m]") for x in ALIASES) for a in pair]
 ALIASES = list(dict.fromkeys(ALIASES))   # dedoublonne en gardant l'ordre
 
 app = FastAPI(title="anthropic-bridge")
-_client = httpx.AsyncClient(base_url=UPSTREAM, timeout=TIMEOUT)
+# Pool EXPLICITE. Mesure 22/08 nuit (5 agents Claude Code, ~150 k de contexte) :
+# 189 flux ouverts pour 100 termines, 45 connexions amont pour 2 clients -- les
+# flux abandonnes cote client (retry, coupure proxy) gardaient leur connexion
+# amont, vLLM generait pour personne, et a 100 connexions (defaut httpx) le pool
+# bloquait TOUT, /health compris : pont muet, processus vivant a 3 % CPU.
+# Le timeout de pool transforme une saturation en 503 rapide au lieu d'un blocage.
+_client = httpx.AsyncClient(
+    base_url=UPSTREAM,
+    timeout=httpx.Timeout(TIMEOUT, connect=10.0, pool=10.0),
+    limits=httpx.Limits(max_connections=512, max_keepalive_connections=64))
 
 
 # ------------------------------------------------------------------ garde-fou
@@ -467,7 +476,7 @@ def _sse(event: str, data: dict) -> bytes:
     return f"event: {event}\ndata: {json.dumps(data, separators=(',', ':'))}\n\n".encode()
 
 
-async def stream_anthropic(payload: dict, req_model: str):
+async def stream_anthropic(payload: dict, req_model: str, request: Request | None = None):
     """Traduit le flux OpenAI en evenements Anthropic.
 
     Le point delicat est l'indexation des blocs : Anthropic numerote chaque bloc
@@ -567,11 +576,21 @@ async def stream_anthropic(payload: dict, req_model: str):
             try:
                 line = await asyncio.wait_for(file.get(), timeout=KEEPALIVE_S)
             except asyncio.TimeoutError:
+                # Client parti pendant le prefill ? On ferme l'amont (vLLM
+                # annule alors la generation) au lieu de pomper dans le vide.
+                if request is not None and await request.is_disconnected():
+                    print("[deconnexion] client parti pendant l'attente, amont ferme", flush=True)
+                    break
                 yield _sse("ping", {"type": "ping"})
                 last_out = time.monotonic()
                 continue
             if line is None:
                 break
+            # Le `yield` ne leve pas toujours a la deconnexion (le proxy Runpod
+            # garde parfois sa connexion ouverte) : on verifie explicitement.
+            if request is not None and (out_tokens % 32 == 0) and await request.is_disconnected():
+                print(f"[deconnexion] client parti apres {out_tokens} jetons, amont ferme", flush=True)
+                break
             if not line.startswith("data: "):
                 continue
             chunk = line[6:].strip()
@@ -739,7 +758,7 @@ async def messages(request: Request):
 
     if payload.get("stream"):
         payload["stream_options"] = {"include_usage": True}
-        return StreamingResponse(stream_anthropic(payload, req_model),
+        return StreamingResponse(stream_anthropic(payload, req_model, request),
                                  media_type="text/event-stream")
 
     r = await _client.post("/v1/chat/completions", json=payload)