front: an in-flight request that leaks freezes the install, so stop leaking one (v0.3.27)

coderai_engine_inflight / coderai_requests_active are load-bearing: the hourly
autoupgrade's idle guard reads them to decide whether a restart is safe, and with
MAX_DEFERRALS=0 a non-zero count makes it SKIP for ever. So one request counted
in and never out silently stops a host taking upgrades, and the only symptom is
a version that stops moving.

Observed on the Digesta orchestrator today: the gauge read 13 against two real
requests, the 09:01 tick logged "SKIP: 7 piece(s) of work in progress" and left
the image alone, and only a manual restart cleared it.

The structural leak is in the streaming proxy path. Starlette awaits a
response's background task only AFTER the body is sent --
`await self.stream_response(send)` then `await self.background()` -- so when the
client disconnects mid-stream the send raises, the exception propagates, and
`BackgroundTask(_release)` never runs. The relay generator's `finally` DOES run
in that case (Starlette closes the iterator) but not when the body is returned
without streaming, so both now call `_release`, which is idempotent via a flag.

`exit_request` is idempotent per rid as well, and that is required rather than
decorative: two release paths firing for one request would otherwise decrement
an UNRELATED in-flight request, and the count would under-report real work --
worse than over-reporting, because then a restart lands during a request.

Every request is now timestamped, not just the inference ones that carry `meta`.
Without that a leaked non-inference request was invisible as well as
unreleasable: the count said 13 and the Tasks page showed nothing, which is how
this went unnoticed.

And a reaper underneath all of it (`server.inflight_max_age_s`, default 3600,
0 disables), called from the two places that ASK whether anything is in flight:

  * the /metrics scrape, because that is the exact moment a stale count does its
    damage -- the scrape that would report a lie is the right place to correct
    it, not to hope somebody notices;
  * the supervisor's drain wait, which would otherwise sit out the whole grace
    period for a request that finished an hour ago.

An hour, not a minute: a cold RunPod pod can make a single OCR page take 14
minutes, and reaping a live request would be a worse bug than the one being
fixed. Training is unaffected -- it is tracked as a JOB and the request that
starts it returns at once.

NOT fixed here, noted for later: `_proxy_passthrough` has the same
BackgroundTask-only release for the system worker. It never calls
enter_request, so it cannot leak the gauge -- it leaks an httpx stream, which is
a smaller and separate problem.

tests/test_front_inflight_leak.py: 14 tests, including double-release, late
release after a reap, thread safety, and the source invariants for the two
release paths. 299 passed across the front/metrics/registry/config/admin
selection; the 9 failures are pre-existing (broker_config x5, broker_streaming,
studio_composed_surfaces, whisper_server x2) and touch no file changed here.
Co-Authored-By: Claude Opus 5 <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_01PxYHWbCAjjjD7A1rhFKtaW

codai/__init__.py

@@ -16,7 +16,7 @@
 
 # Canonical product version for CoderAI — single source of truth. Both the API
 # metadata and the admin web UI read from here.
-__version__ = "0.3.26"
+__version__ = "0.3.27"
 __tagline__ = "Complete orchestration, distribution and escalation of remotizable advanced inference."
 
 # Configure the CUDA caching allocator BEFORE torch is imported anywhere.

codai/config.py

@@ -73,6 +73,15 @@ class ServerConfig:
                                         # generous enough that a GIL-busy engine's
                                         # health poll doesn't time out mid-generation
     proxy_max_inflight: int = 64        # max concurrent proxied requests through the front
+    # Release an in-flight request the front has held for longer than this, so a
+    # leaked one cannot wedge the install. The count is what the hourly
+    # autoupgrade's idle guard reads, and with MAX_DEFERRALS=0 a non-zero count
+    # stops upgrading ALTOGETHER -- a leak is therefore not a cosmetic metric
+    # problem, it silently freezes the host on old code. Well above the slowest
+    # legitimate request: a cold RunPod pod can make one OCR page take 14 minutes,
+    # so this is an hour, not a minute. Training is unaffected (it is tracked as a
+    # JOB; the request that starts it returns at once). 0 disables the reaper.
+    inflight_max_age_s: float = 3600.0
     gpu_swap_batch: int = 10            # on a shared GPU (GGUF-isolation split), serve up to
                                         # this many requests for the model that currently owns
                                         # the card before swapping to a queued different model
@@ -1440,6 +1449,7 @@ class ConfigManager:
                 "engine_gpus": self.config.server.engine_gpus,
                 "proxy_status_timeout": self.config.server.proxy_status_timeout,
                 "proxy_max_inflight": self.config.server.proxy_max_inflight,
+                "inflight_max_age_s": self.config.server.inflight_max_age_s,
                 "gpu_swap_batch": self.config.server.gpu_swap_batch,
                 "engine_restart_drain_grace": self.config.server.engine_restart_drain_grace,
                 "isolate_gguf_engine": self.config.server.isolate_gguf_engine,

codai/frontproxy/app.py

@@ -2571,7 +2571,27 @@ class FrontProxy:
             return JSONResponse(
                 {"error": f"Engine#{engine.id} unreachable: {exc}"}, status_code=502)
 
+        # Released from TWO places, and neither alone is enough.
+        #
+        # Starlette awaits a response's background task only AFTER the body has
+        # been sent: `await self.stream_response(send)` then `await
+        # self.background()`. When the client disconnects mid-stream the send
+        # raises, the exception propagates, and the background task never runs --
+        # so the in-flight count keeps the request for ever. That count is what
+        # the hourly autoupgrade's idle guard reads, and with MAX_DEFERRALS=0 a
+        # non-zero count stops the install upgrading at all: one leaked entry and
+        # the host silently sits on old code until someone restarts it by hand.
+        #
+        # The relay generator's `finally` DOES run on a disconnect (Starlette
+        # closes the iterator), but not when the body is returned without
+        # streaming. So both call this, and exit_request is idempotent per rid.
+        _released = False
+
         async def _release():
+            nonlocal _released
+            if _released:
+                return
+            _released = True
             try:
                 await rp_resp.aclose()
             finally:
@@ -2633,6 +2653,11 @@ class FrontProxy:
                 print(f"[front] upstream engine#{engine.id} ({engine.name}) closed the "
                       f"stream early on {path}: {exc!r}", flush=True)
                 return
+            finally:
+                # Runs on normal completion AND when Starlette closes the iterator
+                # because the client went away -- which is the case the background
+                # task below never reaches.
+                await _release()
 
         resp_headers = self._filter_headers(rp_resp.headers, _DROP_RESP)
         return StreamingResponse(

codai/frontproxy/engine_supervisor.py

@@ -1411,6 +1411,17 @@ class EngineSupervisor:
                 self.registry.update_state(engine_id, healthy=False)
                 deadline = time.time() + drain_grace
                 waited = False
+                # A leaked in-flight entry would make this wait the whole grace
+                # period for a request that finished long ago, so clear the stale
+                # ones before believing the count. The ceiling comes from config
+                # (0 = never reap) and is far above any real request.
+                try:
+                    _ttl = float(getattr(self.config.server,
+                                         "inflight_max_age_s", 3600.0) or 0.0)
+                    if _ttl > 0:
+                        engine.reap_stale_requests(_ttl)
+                except Exception:
+                    pass
                 while engine.inflight > 0 and time.time() < deadline \
                         and not self._stopped.is_set():
                     if not waited:

codai/frontproxy/metrics.py

@@ -239,6 +239,22 @@ def render(front) -> str:
         pass
 
     # Queue / active requests as the overview counts them.
+    #
+    # Reaped FIRST, deliberately. This gauge is not merely watched: the hourly
+    # autoupgrade's idle guard reads it to decide whether a restart is safe, and
+    # with MAX_DEFERRALS=0 a stuck non-zero value stops the install upgrading for
+    # ever. So the scrape that would report a lie is the right moment to correct
+    # it, rather than reporting it and hoping somebody notices.
+    try:
+        _max_age = float(getattr(getattr(front, "config", None), "server", None)
+                         and front.config.server.inflight_max_age_s or 3600.0)
+    except Exception:
+        _max_age = 3600.0
+    for _e in front.registry.all():
+        try:
+            _e.reap_stale_requests(_max_age)
+        except Exception:
+            pass
     try:
         active = sum(int(getattr(e, "inflight", 0) or 0) for e in front.registry.all())
         lines += ["# HELP coderai_requests_active Requests in flight through the front.",

codai/frontproxy/registry.py

@@ -99,6 +99,11 @@ class Engine:
     # front can synthesize Tasks-page entries for work it dispatched — visible even
     # when the engine is too GIL-busy generating to answer its own /admin/api/tasks.
     active: dict = field(default_factory=dict, repr=False, compare=False)
+    #: rid -> time the request was admitted, for EVERY in-flight request (``active``
+    #: only holds the inference ones, which is why a leaked non-inference request
+    #: used to be uncountable as well as unreleasable). Read by
+    #: :meth:`reap_stale_requests`.
+    inflight_since: dict = field(default_factory=dict, repr=False, compare=False)
     # A cluster NODE: another coderai install used as an engine of this front.
     # Nothing is spawned; the node's own front is polled on ``state_path`` and
     # proxied to through ``http_short``/``http_long`` (clients carrying the
@@ -165,6 +170,13 @@ class Engine:
         rid = _u.uuid4().hex[:12]
         with self._inflight_lock:
             self.inflight += 1
+            # EVERY request gets a timestamp, not just the inference ones that get
+            # ``meta``. Without this a leaked non-inference request is invisible:
+            # the count says "13 in flight" and the Tasks page shows nothing, which
+            # is exactly how one went unnoticed long enough to wedge the hourly
+            # autoupgrade (its idle guard reads this count, and MAX_DEFERRALS=0
+            # means a non-zero count stops upgrading for ever).
+            self.inflight_since[rid] = _t.time()
             if meta is not None:
                 m = dict(meta)
                 m["started_at"] = _t.time()
@@ -172,12 +184,54 @@ class Engine:
         return rid
 
     def exit_request(self, rid: Optional[str] = None) -> None:
+        """Release a request. Safe to call twice for the same ``rid``.
+
+        Idempotence is required, not a nicety: the streaming path releases from
+        both the relay generator's ``finally`` and the response's background task,
+        because neither alone is reliable (see FrontApp._proxy). Without the rid
+        bookkeeping the second call would decrement someone else's request and the
+        count would drift NEGATIVE-wards, hiding real work.
+        """
         with self._inflight_lock:
+            if rid is not None:
+                if rid not in self.inflight_since:
+                    return                      # already released
+                self.inflight_since.pop(rid, None)
             if self.inflight > 0:
                 self.inflight -= 1
             if rid:
                 self.active.pop(rid, None)
 
+    def reap_stale_requests(self, max_age_s: float) -> int:
+        """Release in-flight entries older than ``max_age_s``. Returns how many.
+
+        The safety net under the two release paths. A request that leaks is worse
+        than a request that is wrongly reaped: the count is read to decide whether
+        it is safe to restart, so a stuck entry does not merely misreport, it stops
+        the install taking upgrades at all. ``max_age_s`` therefore has to be well
+        above the slowest legitimate request — a cold RunPod pod can make one OCR
+        page take 14 minutes — which is why the default is an hour and not a
+        minute. Training does not count here: it is tracked as a JOB, and the
+        request that started it returns immediately.
+        """
+        if max_age_s <= 0:
+            return 0
+        import time as _t
+        cutoff = _t.time() - float(max_age_s)
+        with self._inflight_lock:
+            stale = [rid for rid, at in self.inflight_since.items() if at < cutoff]
+            for rid in stale:
+                age = _t.time() - self.inflight_since.pop(rid, _t.time())
+                meta = self.active.pop(rid, None)
+                if self.inflight > 0:
+                    self.inflight -= 1
+                print(f"[front] engine#{self.id}: releasing in-flight request {rid} "
+                      f"stuck for {age:.0f}s "
+                      f"({(meta or {}).get('path') or 'non-inference'}) — the count "
+                      f"is what decides whether a restart is safe, so a leaked "
+                      f"entry cannot be left in it", flush=True)
+            return len(stale)
+
 
 class EngineRegistry:
     def __init__(self):

tests/test_front_inflight_leak.py

+"""The in-flight count is load-bearing, so a leak in it freezes the install.
+
+`coderai_engine_inflight` / `coderai_requests_active` are not decoration. The hourly
+autoupgrade's idle guard reads them to decide whether a restart is safe, and with
+`MAX_DEFERRALS=0` a non-zero count means it SKIPS — for ever. So a single request that
+is counted in and never out stops the host taking upgrades at all, silently, and the
+only symptom is a version that stops moving.
+
+That happened on the Digesta orchestrator on 2026-10-12: the gauge read 13 with two
+real requests in the log, the 09:01 tick logged "SKIP: 7 piece(s) of work in progress"
+and left the image alone, and nothing short of a manual restart would have cleared it.
+
+Three defences, one test each:
+
+* the streaming path releases from BOTH the relay generator and the background task,
+  because Starlette runs the latter only after a successful send;
+* `exit_request` is idempotent per rid, so releasing twice cannot decrement someone
+  else's request;
+* a reaper releases anything held implausibly long, so an unknown fourth leak cannot
+  wedge the install either.
+"""
+
+import pathlib
+import sys
+import threading
+import time
+
+ROOT = pathlib.Path(__file__).resolve().parents[1]
+sys.path.insert(0, str(ROOT))
+
+from codai.frontproxy.registry import Engine                   # noqa: E402
+
+
+def _engine():
+    return Engine(id=1, name="e1", backend="gguf", gpu="", port=9001)
+
+
+# ------------------------------------------------------------- the accounting #
+def test_a_request_is_counted_in_and_out():
+    e = _engine()
+    rid = e.enter_request()
+    assert e.inflight == 1
+    e.exit_request(rid)
+    assert e.inflight == 0
+
+
+def test_releasing_twice_does_not_double_decrement():
+    """The streaming path deliberately releases from two places. If the second
+    call decremented again it would cancel an unrelated in-flight request, and the
+    count would under-report real work — which is worse than over-reporting it,
+    because then a restart happens DURING a request."""
+    e = _engine()
+    a = e.enter_request()
+    b = e.enter_request()
+    assert e.inflight == 2
+    e.exit_request(a)
+    e.exit_request(a)          # the second path firing for the same request
+    e.exit_request(a)
+    assert e.inflight == 1     # b is still genuinely in flight
+    e.exit_request(b)
+    assert e.inflight == 0
+
+
+def test_the_count_never_goes_negative():
+    e = _engine()
+    e.exit_request("never-existed")
+    e.exit_request(None)
+    assert e.inflight == 0
+
+
+def test_every_request_is_timestamped_not_just_inference_ones():
+    """`active` only holds requests that carried metadata (the inference ones), so
+    a leaked non-inference request used to be invisible: the count said 13 and the
+    Tasks page showed nothing. Timestamping all of them is what makes a leak both
+    visible and reapable."""
+    e = _engine()
+    plain = e.enter_request()                       # no meta
+    infer = e.enter_request({"model": "m", "kind": "chat", "path": "/v1/chat"})
+    assert set(e.inflight_since) == {plain, infer}
+    assert set(e.active) == {infer}
+
+
+# ----------------------------------------------------------------- the reaper #
+def test_a_stuck_request_is_released():
+    e = _engine()
+    rid = e.enter_request({"path": "/v1/chat/completions"})
+    e.inflight_since[rid] = time.time() - 7200       # two hours ago
+    assert e.reap_stale_requests(3600) == 1
+    assert e.inflight == 0
+    assert rid not in e.inflight_since
+    assert rid not in e.active
+
+
+def test_a_young_request_is_left_alone():
+    """A cold RunPod pod can make one OCR page take 14 minutes. Reaping a live
+    request would be a far worse bug than the one being fixed."""
+    e = _engine()
+    rid = e.enter_request()
+    assert e.reap_stale_requests(3600) == 0
+    assert e.inflight == 1
+    e.exit_request(rid)
+
+
+def test_the_reaper_can_be_switched_off():
+    e = _engine()
+    rid = e.enter_request()
+    e.inflight_since[rid] = time.time() - 99999
+    assert e.reap_stale_requests(0) == 0
+    assert e.inflight == 1
+
+
+def test_reaping_is_selective():
+    e = _engine()
+    old = e.enter_request()
+    new = e.enter_request()
+    e.inflight_since[old] = time.time() - 7200
+    assert e.reap_stale_requests(3600) == 1
+    assert e.inflight == 1
+    assert new in e.inflight_since
+
+
+def test_a_reaped_request_releasing_late_is_harmless():
+    """The request may finish after being reaped; its release must be a no-op
+    rather than a decrement of whatever is in flight by then."""
+    e = _engine()
+    rid = e.enter_request()
+    e.inflight_since[rid] = time.time() - 7200
+    e.reap_stale_requests(3600)
+    other = e.enter_request()
+    e.exit_request(rid)                  # the late release
+    assert e.inflight == 1               # `other` survived
+    e.exit_request(other)
+    assert e.inflight == 0
+
+
+def test_the_accounting_is_thread_safe():
+    """The front releases from a background task and an async generator; the
+    supervisor reaps from its own thread."""
+    e = _engine()
+    rids = []
+    lock = threading.Lock()
+
+    def _work():
+        for _ in range(200):
+            rid = e.enter_request()
+            with lock:
+                rids.append(rid)
+
+    threads = [threading.Thread(target=_work) for _ in range(4)]
+    for t in threads:
+        t.start()
+    for t in threads:
+        t.join()
+    assert e.inflight == 800
+    for rid in rids:
+        e.exit_request(rid)
+    assert e.inflight == 0
+
+
+# --------------------------------------------------- the structural fix itself #
+def test_the_streaming_path_releases_from_both_places():
+    """Starlette awaits a background task only AFTER the body is sent, so a client
+    that disconnects mid-stream never triggers it. The relay generator's finally is
+    the half that covers that case; the background task covers the non-streaming
+    return. Asserted on the source because the alternative is standing up a real
+    ASGI server and a disconnecting client to prove a two-line invariant."""
+    import inspect
+    from codai.frontproxy.app import FrontProxy
+    src = inspect.getsource(FrontProxy.proxy)
+    assert "_released = False" in src, "the release must be idempotent"
+    assert "BackgroundTask(_release)" in src, "the non-streaming half"
+    # ...and the generator half: a finally that awaits the release.
+    assert src.count("await _release()") >= 1, "the disconnect half"
+
+
+def test_the_metrics_scrape_reaps_before_reporting():
+    """The scrape the autoupgrade reads is exactly the moment a stale count does
+    damage, so it is the right moment to correct it."""
+    import inspect
+    from codai.frontproxy import metrics
+    src = inspect.getsource(metrics)
+    assert "reap_stale_requests" in src
+    assert src.index("reap_stale_requests") < src.index("coderai_requests_active")
+
+
+def test_the_drain_wait_reaps_first():
+    """Otherwise a restart waits out the whole grace period for a request that
+    finished an hour ago."""
+    import inspect
+    from codai.frontproxy import engine_supervisor
+    src = inspect.getsource(engine_supervisor)
+    assert "reap_stale_requests" in src
+
+
+def test_the_ceiling_is_configurable_and_generous():
+    from codai.config import ServerConfig
+    assert ServerConfig().inflight_max_age_s == 3600.0