a pod's in-flight count must be able to come down on its own (v0.4.28)

`pod.inflight` is raised in `_pick` and lowered only in `release`, which makes
it a ratchet. Any dispatch path that fails to release raises it for the life of
the pod, and at max_inflight_per_pod the pod is refused all work while its card
sits idle -- reported as "at capacity", which is the most misleading thing this
file can say.

Observed on a deliberately single-pod fleet: the pool reported inflight=64 of
64 with 63 pages queued, while the pod itself reported `Running: 31,
Waiting: 0` at 10.6% KV cache. Half the slots did not exist. A single test
document then waited 900s for a pod that was half idle, and timed out.

Two such leaks have now been found and fixed -- a lease dropped during
acquisition (0.4.23) and a page whose client disconnected. The reason this
change exists is the third one, which has not been found: a counter that can
only rise is one bug away from taking a pod out of service permanently, and
every leak so far has been invisible precisely because the pool reported
confidence about a number nothing could correct.

So each slot now records when it was taken, and the LIST is the authority:

* slots held longer than INFLIGHT_TTL_S (900s) cannot belong to a live page and
  are returned. Deliberately generous -- a safety net, not a timeout. Returning
  a slot that IS in use would hand a pod more work than its cap allows, while
  holding a phantom a few minutes longer costs only those minutes.
* a counter standing above its own recorded slots is counting slots with no
  evidence of ever being taken, and is pulled down to the evidence. This is not
  conditional on anything having expired -- the first version of the sweep only
  examined pods with a stale entry, which would have skipped a pod whose
  counter had drifted upward, and skipped every pod already in the pool when
  this code arrives carrying a counter and no timestamps: exactly the
  locked-out pod the sweep exists to free.

The list is created on demand rather than required at construction, so
duck-typed pod stand-ins are not forced to carry a bookkeeping field, as with
the per-card memo.
Co-Authored-By: Claude Opus 5 <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_01KnfkxBpcZ1v6ypRFFyhrVM

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.4.27"
+__version__ = "0.4.28"
 __tagline__ = "Complete orchestration, distribution and escalation of remotizable advanced inference."
 
 # Configure the CUDA caching allocator BEFORE torch is imported anywhere.

codai/api/runpod_worker.py

@@ -2626,6 +2626,22 @@ class PodHandle:
     is_spot: bool = False
     healthy: bool = True
     inflight: int = 0
+    #: When each in-flight slot was taken, oldest first.
+    #:
+    #: `inflight` on its own is a ratchet: it is raised in `_pick` and lowered
+    #: only in `release`, so any dispatch path that fails to release -- a client
+    #: that disconnects, an exception on a branch nobody tested, a leak not yet
+    #: found -- raises it permanently. Once it reaches max_inflight_per_pod the
+    #: pod is locked out for the rest of its life while its card sits idle, and
+    #: the pool reports it as "at capacity", which is the single most misleading
+    #: thing this file can say.
+    #:
+    #: Observed 2026-10-18: pool said inflight=64 of 64 with 63 pages queued;
+    #: the pod itself reported `Running: 31, Waiting: 0` at 10.6% KV cache. Half
+    #: the slots did not exist. Reconciling against these timestamps lets the
+    #: pool recover from leaks it does not know about, which is the only kind
+    #: that matters.
+    inflight_since: list = field(default_factory=list)
     last_used: float = 0.0
     #: Chosen for retirement and no longer given new work. It stays healthy and
     #: keeps serving what it already has: killing a pod with requests on it to
@@ -4320,6 +4336,7 @@ class RunpodPodPool:
                     grow_now = True
                 if chosen is not None:
                     chosen.inflight += 1
+                    _slots_of(chosen).append(time.time())
                     chosen.last_used = time.time()
                     self._remember_affinity(affinity, chosen)
                     url = chosen.url
@@ -4415,6 +4432,13 @@ class RunpodPodPool:
     def release(self, pod: PodHandle):
         with self._cv:
             pod.inflight = max(0, pod.inflight - 1)
+            # Oldest first: the slots are interchangeable (nothing identifies
+            # which page holds which), and dropping the oldest keeps the list a
+            # true picture of how long the REMAINING ones have been held, which
+            # is what the staleness sweep reads.
+            slots = _slots_of(pod)
+            if slots:
+                slots.pop(0)
             pod.last_used = time.time()
             self._cv.notify_all()
 
@@ -4756,6 +4780,70 @@ class RunpodPodPool:
                 "needed": need, "floor": floor,
                 "retire_when_idle": self._release_pending}
 
+    def _reclaim_stranded_inflight(self, now: float) -> None:
+        """Give back in-flight slots that no page can still be holding.
+
+        `inflight` is a ratchet: raised in `_pick`, lowered only in `release`.
+        Any dispatch path that fails to release raises it for the life of the
+        pod, and at `max_inflight_per_pod` the pod is locked out while its card
+        idles -- reported, misleadingly, as "at capacity". Two such leaks have
+        been found and fixed (a lease dropped during acquisition, a page whose
+        client went away); the reason this sweep exists is the third, which has
+        not been found yet.
+
+        So the pool stops trusting the counter indefinitely. A slot held longer
+        than any page can legitimately take is not in flight, whatever the
+        counter says, and is returned.
+
+        Deliberately generous: this is a safety net, not a timeout. Cancelling
+        real work would be far worse than holding a phantom slot a few minutes
+        longer, so the bound is well past the slowest plausible page (a full
+        token budget, its continuations and a retry, on a contended card).
+        """
+        ttl = INFLIGHT_TTL_S
+        if ttl <= 0:
+            return
+        cutoff = now - ttl
+        freed_total = 0
+        with self._cv:
+            for pod in self.pods:
+                slots = _slots_of(pod)
+                stale = sum(1 for t in slots if t < cutoff)
+                if stale:
+                    slots = [t for t in slots if t >= cutoff]
+                    pod.inflight_since = slots
+                # The LIST is the authority, always, not only when something
+                # expired. `_pick` records a timestamp for every slot it takes,
+                # so a counter standing above the recorded slots is counting
+                # slots with no evidence that they were ever taken -- nothing
+                # can attribute them to a live page, and they can only have
+                # come from a path that raised the counter and lost the page.
+                #
+                # This also covers the upgrade itself: pods already in the pool
+                # when this code arrives carry a counter and no timestamps, and
+                # skipping them would leave exactly the locked-out pod this
+                # sweep exists to free.
+                unaccounted = max(0, pod.inflight - len(slots))
+                if not stale and not unaccounted:
+                    continue
+                pod.inflight = len(slots)
+                freed_total += stale + unaccounted
+                detail = []
+                if stale:
+                    detail.append(f"{stale} held longer than {ttl:.0f}s")
+                if unaccounted:
+                    detail.append(f"{unaccounted} with no record of being taken")
+                print(f"[runpod] pod {pod.pod_id}: returned "
+                      f"{stale + unaccounted} in-flight slot(s) that cannot "
+                      f"belong to a live page ("
+                      + ", ".join(detail)
+                      + f"). inflight is now {pod.inflight}. A slot leaks when "
+                      f"a page is lost between being granted a pod and "
+                      f"releasing it; until now the pod was being refused work "
+                      f"it could serve.", flush=True)
+            if freed_total:
+                self._cv.notify_all()
+
     def _scale_down(self, now: float) -> None:
         """Retire one surplus pod per pass, once the surplus has persisted.
 
@@ -4835,6 +4923,7 @@ class RunpodPodPool:
                 self._drain_pod_log(p)
             except Exception:
                 pass
+        self._reclaim_stranded_inflight(now)
         self._scale_down(now)
         with self._cv:
             pods = list(self.pods)
@@ -5070,6 +5159,34 @@ def _forget_unavailable(gpu_type_id: str, dc: str = "") -> None:
 #: WHY the server is busy, so it must not be the first thing a busy server
 #: stops serving. Short, because a partial answer now beats a complete one
 #: after the incident.
+#: How long an in-flight slot may be held before the pool stops believing it.
+#:
+#: A safety net against leaks in the dispatch path, not a request timeout: see
+#: `_reclaim_stranded_inflight`. Set well past the slowest page a pod can
+#: legitimately be working on -- a full token budget plus its continuations and
+#: a retry, on a card serving dozens of other pages -- because returning a slot
+#: that IS in use would hand the same pod more work than its cap allows, while
+#: holding a phantom a few minutes longer costs only those minutes.
+INFLIGHT_TTL_S = 900.0
+
+
+def _slots_of(pod) -> list:
+    """A pod's in-flight timestamp list, created on demand.
+
+    Not every object the pool is handed is a PodHandle: tests and duck-typed
+    stand-ins carry only what they need, and a field added for bookkeeping
+    should not force any of them to be rewritten. Initialised here rather than
+    required at construction, as the per-card memo is.
+    """
+    slots = getattr(pod, "inflight_since", None)
+    if slots is None:
+        slots = []
+        try:
+            pod.inflight_since = slots
+        except Exception:
+            pass                # read-only stand-in: bookkeeping degrades, nothing breaks
+    return slots
+
 SNAPSHOT_LOCK_TIMEOUT_S = 2.0
 
 

tests/test_runpod_inflight_reclaim.py

+"""A pod's in-flight counter must be able to come down on its own.
+
+`pod.inflight` is raised in `_pick` and lowered only in `release`. That makes
+it a ratchet: any dispatch path that fails to release raises it for the life of
+the pod, and once it reaches max_inflight_per_pod the pod is refused all work
+while its card sits idle -- reported as "at capacity", which is the most
+misleading thing the pool can say.
+
+Observed 2026-10-18 on a single-pod fleet: the pool reported inflight=64 of 64
+with 63 pages queued, while the pod itself reported `Running: 31, Waiting: 0`
+at 10.6% KV cache. Half the slots did not exist, and nothing in the codebase
+could ever have given them back.
+
+Two leaks have been found and fixed (a lease dropped during acquisition in
+0.4.23; a page whose client disconnected). This sweep exists for the third one,
+which has not been found. It is a safety net, not a timeout: returning a slot
+that IS in use would hand a pod more work than its cap allows, so the bound is
+deliberately far past the slowest legitimate page.
+"""
+import time
+import pytest
+
+import codai.api.runpod_worker as rw
+
+
+def _pod(inflight, ages=(), pod_id="p1"):
+    """A pod holding `inflight` slots, taken `ages` seconds ago."""
+    p = rw.PodHandle(pod_id=pod_id, url="https://x", hourly_usd=0.5,
+                     started_at=time.time())
+    p.inflight = inflight
+    now = time.time()
+    p.inflight_since = [now - a for a in ages]
+    return p
+
+
+class _Pool:
+    """Just enough pool to run the sweep against."""
+    def __init__(self, pods):
+        import threading
+        self.pods = list(pods)
+        self._cv = threading.Condition()
+
+    _reclaim_stranded_inflight = rw.RunpodPodPool._reclaim_stranded_inflight
+
+
+def sweep(pods, now=None):
+    pool = _Pool(pods)
+    pool._reclaim_stranded_inflight(now if now is not None else time.time())
+    return pool
+
+
+# --- the observed incident ---
+
+def test_a_fully_stranded_pod_is_freed():
+    """inflight=64 with every slot older than the TTL: the pod is usable again."""
+    pod = _pod(64, ages=[rw.INFLIGHT_TTL_S + 60] * 64)
+    sweep([pod])
+    assert pod.inflight == 0
+    assert pod.inflight_since == []
+
+
+def test_live_slots_survive_and_phantoms_do_not():
+    """31 real pages running, 33 stranded -- the real ones must not be touched."""
+    ages = [5.0] * 31 + [rw.INFLIGHT_TTL_S + 120] * 33
+    pod = _pod(64, ages=ages)
+    sweep([pod])
+    assert pod.inflight == 31
+    assert len(pod.inflight_since) == 31
+
+
+# --- it must never cancel real work ---
+
+def test_nothing_is_freed_before_the_ttl():
+    pod = _pod(64, ages=[rw.INFLIGHT_TTL_S - 1] * 64)
+    sweep([pod])
+    assert pod.inflight == 64
+
+
+def test_a_busy_pod_is_untouched():
+    pod = _pod(8, ages=[1.0, 2.0, 3.0, 4.0, 5.0, 6.0, 7.0, 8.0])
+    sweep([pod])
+    assert pod.inflight == 8
+
+
+def test_the_ttl_is_generous_enough_for_a_slow_page():
+    """A full budget plus continuations and a retry on a contended card.
+    12288 tokens at the ~300 tok/s a single stream gets is ~41s; with three
+    continuations and a retry that is minutes, not seconds."""
+    assert rw.INFLIGHT_TTL_S >= 600
+
+
+# --- the counter and the list cannot disagree ---
+
+def test_release_drops_one_timestamp():
+    pod = _pod(3, ages=[30.0, 20.0, 10.0])
+    pool = _Pool([pod])
+    rw.RunpodPodPool.release(pool, pod)
+    assert pod.inflight == 2
+    assert len(pod.inflight_since) == 2
+
+
+def test_release_drops_the_oldest_timestamp():
+    """The remaining list must still say how long the rest have been held."""
+    pod = _pod(3, ages=[300.0, 20.0, 10.0])
+    pool = _Pool([pod])
+    rw.RunpodPodPool.release(pool, pod)
+    ages = sorted(time.time() - t for t in pod.inflight_since)
+    assert max(ages) < 100, "the 300s-old slot should have been the one dropped"
+
+
+def test_release_below_zero_is_impossible():
+    pod = _pod(0, ages=[])
+    pool = _Pool([pod])
+    rw.RunpodPodPool.release(pool, pod)
+    assert pod.inflight == 0
+    assert pod.inflight_since == []
+
+
+def test_the_counter_is_never_left_above_the_surviving_slots():
+    """A counter that drifted above its own timestamps is corrected down."""
+    pod = _pod(64, ages=[1.0, 2.0])      # counter says 64, list says 2
+    sweep([pod])
+    assert pod.inflight <= len(pod.inflight_since)
+
+
+# --- housekeeping ---
+
+def test_a_pod_with_no_slots_is_skipped():
+    pod = _pod(0, ages=[])
+    sweep([pod])
+    assert pod.inflight == 0
+
+
+def test_every_pod_in_the_pool_is_swept():
+    a = _pod(4, ages=[rw.INFLIGHT_TTL_S + 10] * 4, pod_id="a")
+    b = _pod(4, ages=[rw.INFLIGHT_TTL_S + 10] * 4, pod_id="b")
+    sweep([a, b])
+    assert (a.inflight, b.inflight) == (0, 0)
+
+
+def test_a_zero_ttl_disables_the_sweep(monkeypatch):
+    """An operator must be able to turn a safety net off."""
+    monkeypatch.setattr(rw, "INFLIGHT_TTL_S", 0.0)
+    pod = _pod(64, ages=[99999.0] * 64)
+    sweep([pod])
+    assert pod.inflight == 64
+
+
+def test_the_sweep_runs_in_maintain():
+    src = open("codai/api/runpod_worker.py").read()
+    assert "self._reclaim_stranded_inflight(now)" in src
+    blk = src.split("def maintain")[1].split("def close")[0]
+    assert "_reclaim_stranded_inflight" in blk, "must be wired into the scaler pass"
+
+
+def test_a_counter_with_no_timestamps_at_all_is_reconciled():
+    """The upgrade case, and the live incident.
+
+    A pod already in the pool when this code arrives carries a counter and no
+    timestamp list. It is also precisely the pod that is locked out at its cap.
+    Skipping it would leave the one pod this sweep exists to free.
+    """
+    pod = _pod(64, ages=[])
+    sweep([pod])
+    assert pod.inflight == 0
+
+
+def test_a_drifted_counter_is_pulled_down_to_its_evidence():
+    """Counter says 64, two slots recorded and both fresh: 62 are fiction."""
+    pod = _pod(64, ages=[1.0, 2.0])
+    sweep([pod])
+    assert pod.inflight == 2
+    assert len(pod.inflight_since) == 2
+
+
+def test_reconciliation_does_not_need_anything_to_be_stale():
+    """The first version only looked at pods with an expired slot, so a pod
+    whose counter had drifted upward with no stale entry stayed locked out."""
+    pod = _pod(10, ages=[1.0])
+    sweep([pod])
+    assert pod.inflight == 1