renting a fleet must not starve the process serving it (v0.4.20)

0.4.19 fixed the lease -- pages now reach the pool, where in-flight had been
stuck at zero -- and the run still produced nothing: 56 pods, 27 healthy,
1,040 pages in flight, ZERO `POST /v1/ocr` in any pod's log, and
/v1/runpod/status unanswerable while /healthz answered in 2.9 ms.

The ramp was eating the process. Measured on that run: coderai-cpu at 75%,
event-loop stalls of 57.4 s and 44.9 s, and 406 candidate rejections in 120
seconds with every GPU type re-probed 33 to 72 times. Two duplicated costs,
both paid once per BOOT and so multiplied by 32 parallel boots:

  * Each boot thread swept the whole ~70-candidate list itself, so each
    independently rediscovered that A100 SXM has no capacity in any of seven
    regions. The verdict is now shared and short-lived (`_unavail`, 90 s):
    one thread learns it, the other 31 skip it. On this run's numbers that is
    2,240 probes reduced to 70. A successful deploy clears the entry, and the
    TTL is short because capacity returns without telling us.

  * Each boot fetched and parsed the FULL GPU price catalogue
    (`list_gpu_types` was uncached), so 32 boots meant 32 identical large JSON
    bodies parsed within seconds. Cached 60 s process-wide, each caller
    getting its own copy so a mutating caller cannot corrupt it, with
    `fresh=True` for a caller showing prices to a person rather than picking a
    card in a storm.

Both costs are Python, so they hold the GIL, and CPython gives it to whoever
is running: the event loop carrying the page forwards simply did not get
scheduled. That is why in-flight climbed while no socket was ever opened to a
pod -- and the undelivered queue then read as demand, so the scaler rented
more pods, which starved the loop harder. The fleet grew because it could not
serve and could not serve because it was growing.

No blocking call is involved, which is why the 0.4.17 static check passes and
the loop watchdog was the thing that found it.
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.19"
+__version__ = "0.4.20"
 __tagline__ = "Complete orchestration, distribution and escalation of remotizable advanced inference."
 
 # Configure the CUDA caching allocator BEFORE torch is imported anywhere.

codai/api/runpod_client.py

@@ -59,6 +59,20 @@ class RunpodTransient(RunpodError):
 #: every boot.
 _GLOBAL_VOLUME_WARNED = False
 
+#: The GPU catalogue, shared and briefly cached: (expiry, rows).
+#:
+#: Every `_rank_gpus` asked for it, and `_rank_gpus` runs once per pod being
+#: provisioned. With 32 pods booting in parallel that is 32 identical catalogue
+#: queries and 32 large JSON bodies parsed, all within a few seconds, and the
+#: parsing is Python -- so it holds the GIL and starves whatever else the
+#: process is trying to do. On a live run that was the event loop carrying the
+#: page forwards: 1,754 pages in flight and not one socket opened to a pod.
+#:
+#: The catalogue is a price list. It does not change inside a boot storm.
+_GPU_CATALOG: tuple = (0.0, None)
+_GPU_CATALOG_LOCK = __import__("threading").Lock()
+GPU_CATALOG_TTL_S = 60.0
+
 
 def _global_field_refused(text: str) -> bool:
     """Whether a create failed *because* of the global-volume field.
@@ -223,12 +237,21 @@ class RunpodClient:
     # ------------------------------------------------------------------ #
     # GPU catalog + selection
     # ------------------------------------------------------------------ #
-    def list_gpu_types(self) -> list:
+    def list_gpu_types(self, fresh: bool = False) -> list:
         """Return the RunPod GPU catalog with pricing, normalized:
 
         [{id, display_name, memory_gb, secure_price, community_price,
           on_demand_price, spot_price}]  (prices are USD/hr for 1 GPU; None if N/A)
+
+        Cached for `GPU_CATALOG_TTL_S` across every caller in the process — see
+        `_GPU_CATALOG`. `fresh=True` forces a fetch, for a caller that is about
+        to show prices to a person rather than pick a card in a boot storm.
         """
+        global _GPU_CATALOG
+        if not fresh:
+            until, rows = _GPU_CATALOG
+            if rows is not None and until > time.time():
+                return list(rows)
         q = """
         query GpuTypes {
           gpuTypes {
@@ -259,6 +282,8 @@ class RunpodClient:
                 "on_demand_price": on_demand,
                 "spot_price": spot,
             })
+        with _GPU_CATALOG_LOCK:
+            _GPU_CATALOG = (time.time() + GPU_CATALOG_TTL_S, list(out))
         return out
 
     def pick_gpu(self, *, min_vram_gb: float = 0.0, max_hourly_usd: float = 0.0,

codai/api/runpod_worker.py

@@ -3135,6 +3135,7 @@ class RunpodPodPool:
                       f"HTTP — that image cannot take a certificate; the bearer "
                       f"token travels unencrypted", flush=True)
         last = None
+        _skipped_cached = 0
         for i in range(start, len(ranked)):
             sel = ranked[i]
             # THE POD'S OWN GATE, for the card it is about to run on. The plan
@@ -3160,6 +3161,13 @@ class RunpodPodPool:
                     + _key[:20].replace("/", "_").replace(" ", "_")
                     + "-" + uuid.uuid4().hex[:6])
             _dc = sel.get("_dc") or self._data_center()
+            # ALREADY REFUSED, recently, for somebody else? Then do not spend a
+            # round trip and a GIL slice learning it again. This is the whole
+            # saving: 32 boot threads each sweeping ~70 candidates produced 406
+            # rejections in 120 s and saturated the core the event loop needed.
+            if _is_unavailable(sel.get("gpu_type_id"), _dc):
+                _skipped_cached += 1
+                continue
             # Say so when this is not the card that was asked for. Renting a
             # different GPU than the configuration names is correct here — it is
             # what allow_other_gpus is for — but it must never be silent.
@@ -3181,6 +3189,11 @@ class RunpodPodPool:
                   f"({sel['cloud_type']}{'/spot' if sel['is_spot'] else ''}) ${sel['price']}/hr"
                   + (f" in {_dc}" if _dc else " (region: RunPod's choice)") + _alt,
                   flush=True)
+            if _skipped_cached:
+                print(f"[runpod] {self.model_key!r}: skipped {_skipped_cached} "
+                      f"candidate(s) already known to have no capacity "
+                      f"(cached {UNAVAIL_TTL_S:.0f}s)", flush=True)
+                _skipped_cached = 0
             try:
                 _disk_gb, _disk_note = disk_for(
                     _model_entry(self.served or self.model_key), self.mcfg, self.account)
@@ -3201,10 +3214,14 @@ class RunpodPodPool:
                                       or getattr(self.account, "registry_auth_id", "") or ""),
                     entrypoint=plan.get("entrypoint"), start_cmd=plan.get("start_cmd"),
                     direct_tcp=bool(getattr(self, "_direct", False)))
+                _forget_unavailable(sel.get("gpu_type_id"), _dc)
                 return pod_id, sel, i + 1
             except RunpodError as exc:
                 msg = str(exc).lower()
                 if "no longer any instances" in msg or "no instances" in msg or "capacity" in msg:
+                    # Shared, so the other boot threads skip it instead of each
+                    # discovering the same shortage. See `_unavail`.
+                    _mark_unavailable(sel.get("gpu_type_id"), _dc)
                     print(f"[runpod] {sel['display_name']} ({sel['cloud_type']}) unavailable — "
                           "trying next candidate", flush=True)
                     last = exc
@@ -4964,6 +4981,62 @@ class RunpodPodPool:
 _DENY_CAP = 4096
 
 
+#: (gpu_type_id, data_center) -> when its "no capacity" verdict expires.
+#:
+#: RunPod's answer to "can I have this card here" is the same for every caller
+#: for as long as the shortage lasts, but each boot thread re-asked
+#: independently: 32 pods booting in parallel, each sweeping ~70 candidates,
+#: produced 406 rejections in 120 seconds on a live run -- every GPU type
+#: re-probed 33 to 72 times. That saturated a core (coderai-cpu at 60%), and
+#: because CPython hands the GIL to whoever is running, it starved the event
+#: loop the page forwards live on: 1,754 pages "in flight", not one socket
+#: opened to a pod, and the undelivered queue made the scaler rent MORE pods.
+#:
+#: So the verdict is shared and short-lived. Short because capacity returns
+#: without telling us, shared because asking again is what cost the run.
+_unavail: dict = {}
+_unavail_lock = threading.Lock()
+
+#: How long a "no capacity" answer is trusted. Long enough to collapse a boot
+#: storm's duplicate probes, short enough that a freed card is picked up on the
+#: next rent rather than the next release.
+UNAVAIL_TTL_S = 90.0
+
+
+def _mark_unavailable(gpu_type_id: str, dc: str = "") -> None:
+    if not gpu_type_id:
+        return
+    with _unavail_lock:
+        _unavail[(str(gpu_type_id), str(dc or ""))] = time.time() + UNAVAIL_TTL_S
+        if len(_unavail) > 4096:
+            now = time.time()
+            for k, v in list(_unavail.items()):
+                if v <= now:
+                    _unavail.pop(k, None)
+
+
+def _is_unavailable(gpu_type_id: str, dc: str = "") -> bool:
+    """Has this card already been refused here, recently, for anyone?"""
+    if not gpu_type_id:
+        return False
+    now = time.time()
+    with _unavail_lock:
+        for key in ((str(gpu_type_id), str(dc or "")), (str(gpu_type_id), "")):
+            until = _unavail.get(key)
+            if until and until > now:
+                return True
+            if until:
+                _unavail.pop(key, None)
+    return False
+
+
+def _forget_unavailable(gpu_type_id: str, dc: str = "") -> None:
+    """A card that just deployed is available, whatever we thought."""
+    with _unavail_lock:
+        _unavail.pop((str(gpu_type_id), str(dc or "")), None)
+        _unavail.pop((str(gpu_type_id), ""), None)
+
+
 class NoPodRightNow(RuntimeError):
     """No pod is free THIS INSTANT and the caller asked not to wait here.
 

tests/test_runpod_capacity_cache.py

+"""A card refused for no capacity must not be re-asked by every boot thread.
+
+RunPod's answer to "can I have this card in this region" is the same for every
+caller while the shortage lasts, but each boot thread asked independently. On a
+live run 32 pods were booting in parallel, each sweeping ~70 candidates: 406
+rejections in 120 seconds, every GPU type re-probed 33 to 72 times, one core
+saturated (coderai-cpu at 60%).
+
+That mattered because of WHERE the cost landed. CPython gives the GIL to
+whoever is running, so the provisioning threads starved the event loop that the
+page forwards live on: 1,754 pages counted as in flight, not one socket opened
+to a pod, zero documents completed -- and the undelivered queue read as demand,
+so the scaler rented more pods, which blocked the loop harder. The fleet grew
+because it could not serve, and could not serve because it was growing.
+
+The verdict is now shared and short-lived: shared because asking again is what
+cost the run, short because capacity comes back without telling us.
+"""
+
+import time
+
+import pytest
+
+from codai.api import runpod_worker as rw
+
+
+@pytest.fixture(autouse=True)
+def _clean():
+    rw._unavail.clear()
+    yield
+    rw._unavail.clear()
+
+
+def test_an_unmarked_card_is_available():
+    assert rw._is_unavailable("NVIDIA A100-SXM4-80GB", "EU-SE-1") is False
+
+
+def test_a_refusal_is_remembered():
+    rw._mark_unavailable("NVIDIA A100-SXM4-80GB", "EU-SE-1")
+    assert rw._is_unavailable("NVIDIA A100-SXM4-80GB", "EU-SE-1") is True
+
+
+def test_it_is_remembered_per_region():
+    """A shortage in one region says nothing about another."""
+    rw._mark_unavailable("NVIDIA A40", "EU-SE-1")
+    assert rw._is_unavailable("NVIDIA A40", "EU-SE-1") is True
+    assert rw._is_unavailable("NVIDIA A40", "EU-FR-1") is False
+
+
+def test_a_regionless_refusal_covers_every_region():
+    """RunPod's choice of region: the refusal was not about a place."""
+    rw._mark_unavailable("NVIDIA A40", "")
+    assert rw._is_unavailable("NVIDIA A40", "EU-FR-1") is True
+    assert rw._is_unavailable("NVIDIA A40", "") is True
+
+
+def test_the_verdict_expires(monkeypatch):
+    """Short on purpose: a freed card must be picked up on the next rent."""
+    monkeypatch.setattr(rw, "UNAVAIL_TTL_S", 0.05)
+    rw._mark_unavailable("NVIDIA L40S", "EU-NL-1")
+    assert rw._is_unavailable("NVIDIA L40S", "EU-NL-1") is True
+    time.sleep(0.1)
+    assert rw._is_unavailable("NVIDIA L40S", "EU-NL-1") is False
+
+
+def test_a_successful_deploy_clears_the_verdict():
+    rw._mark_unavailable("NVIDIA A40", "EU-SE-1")
+    rw._forget_unavailable("NVIDIA A40", "EU-SE-1")
+    assert rw._is_unavailable("NVIDIA A40", "EU-SE-1") is False
+
+
+def test_a_deploy_clears_the_regionless_verdict_too():
+    rw._mark_unavailable("NVIDIA A40", "")
+    rw._forget_unavailable("NVIDIA A40", "EU-SE-1")
+    assert rw._is_unavailable("NVIDIA A40", "EU-SE-1") is False
+
+
+def test_an_empty_card_id_is_never_cached():
+    rw._mark_unavailable("", "EU-SE-1")
+    assert rw._unavail == {}
+    assert rw._is_unavailable("", "EU-SE-1") is False
+
+
+def test_the_record_does_not_grow_without_bound(monkeypatch):
+    monkeypatch.setattr(rw, "UNAVAIL_TTL_S", -1.0)      # expire immediately
+    for i in range(5000):
+        rw._mark_unavailable(f"card{i}", "r")
+    assert len(rw._unavail) <= 4096
+
+
+def test_the_ttl_is_short_enough_to_not_strand_a_fleet():
+    """Too long and a shortage that cleared keeps the pool off a good card."""
+    assert 10.0 <= rw.UNAVAIL_TTL_S <= 300.0
+
+
+def test_one_thread_learning_spares_the_others():
+    """The saving, stated as the arithmetic that caused the incident."""
+    types = [f"gpu{i}" for i in range(10)]
+    regions = [f"r{i}" for i in range(7)]
+    probes = []
+
+    def probe(t, r):
+        if rw._is_unavailable(t, r):
+            return False
+        probes.append((t, r))
+        rw._mark_unavailable(t, r)      # what the refusal does
+        return False
+
+    for _thread in range(32):           # 32 parallel boots
+        for t in types:
+            for r in regions:
+                probe(t, r)
+    # 32 x 70 = 2240 without the cache; one pass with it.
+    assert len(probes) == 70, len(probes)
+
+
+# ------------------------------------------------ the catalogue behind it #
+# `_rank_gpus` asks for the GPU catalogue, and it runs once per pod being
+# provisioned. 32 parallel boots fetched and parsed 32 identical price lists
+# within seconds, in Python, holding the GIL the event loop needed.
+
+from codai.api import runpod_client as rc
+
+
+@pytest.fixture(autouse=True)
+def _clean_catalog():
+    rc._GPU_CATALOG = (0.0, None)
+    yield
+    rc._GPU_CATALOG = (0.0, None)
+
+
+class _FakeClient(rc.RunpodClient):
+    def __init__(self):
+        self.fetches = 0
+
+    def _gql(self, q, *a, **k):
+        self.fetches += 1
+        return {"gpuTypes": [
+            {"id": "g1", "displayName": "NVIDIA A40", "memoryInGb": 48,
+             "securePrice": 1.0, "communityPrice": 0.9,
+             "lowestPrice": {"minimumBidPrice": 0.5, "uninterruptablePrice": 1.0}},
+        ]}
+
+
+def test_the_catalogue_is_fetched_once_for_many_callers():
+    c = _FakeClient()
+    for _ in range(32):                 # 32 boot threads
+        rows = c.list_gpu_types()
+        assert rows and rows[0]["display_name"] == "NVIDIA A40"
+    assert c.fetches == 1, f"{c.fetches} catalogue fetches for 32 boots"
+
+
+def test_each_caller_gets_its_own_copy():
+    """A caller that mutates its rows must not corrupt the cache."""
+    c = _FakeClient()
+    first = c.list_gpu_types()
+    first.append({"id": "junk"})
+    assert len(c.list_gpu_types()) == 1
+
+
+def test_fresh_forces_a_fetch():
+    """For a caller showing prices to a person, not picking in a storm."""
+    c = _FakeClient()
+    c.list_gpu_types()
+    c.list_gpu_types(fresh=True)
+    assert c.fetches == 2
+
+
+def test_the_catalogue_cache_expires(monkeypatch):
+    c = _FakeClient()
+    monkeypatch.setattr(rc, "GPU_CATALOG_TTL_S", 0.05)
+    c.list_gpu_types()
+    time.sleep(0.1)
+    c.list_gpu_types()
+    assert c.fetches == 2
+
+
+def test_the_catalogue_ttl_is_not_so_long_that_prices_go_stale():
+    assert 10.0 <= rc.GPU_CATALOG_TTL_S <= 600.0