ocr: a pod that vanished must not take the queue with it (v0.3.18)

RunPod reclaimed an A40 mid-document today. The pool went on reporting it
healthy, so every page chose the same corpse: the in-flight count climbed, the
pod was never dropped, and nothing drained. From the client that reads as
"requests neither complete nor time out", and it is what wedged the test.

RunpodBackend has always handled this — _unreachable() drops the pod and the
request goes again. The fan-out gateway never did: it returned 502 and left the
dead pod in the pool as healthy. So a lease can now report its upstream dead,
which drops the pod, clears the cached candidate set and releases the slot; the
page is retried ONCE against whatever replaces it, and a third attempt would
just be a retry loop against a real outage.

Only connect-level failures count as dead. A read timeout is a model taking its
time over a dense page, and discarding a healthy pod for that would be an
outage of its own.

Also: the first version of that handler could not work. It used a metaclass
with __instancecheck__ so `except _DEAD_UPSTREAM` would match httpx's errors —
but `except` matches on the class hierarchy at C level and never consults
__instancecheck__, so it caught nothing. A test caught it; the check is now an
explicit isinstance() where it cannot lie.

And the registry path again: 0.3.15 keyed the fallback off XDG_CONFIG_HOME,
which the container launcher exports only into the processes IT starts. A
diagnostic script entering the container another way printed a THIRD registry
path and created a second pod TLS CA. The fallback now reads
CODERAI_CONFIG_DIR, the variable the launcher actually sets, so every process
agrees however it was started — and since pod_tls.tls_dir() derives from the
registry path, the CA follows.
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.17"
+__version__ = "0.3.18"
 __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

@@ -3533,8 +3533,12 @@ def _pod_registry_path() -> str:
         base = ""
     if not base:
         try:
-            from codai.platform_paths import user_config_dir
-            base = str(user_config_dir())
+            from codai.platform_paths import (deployment_config_dir,
+                                              user_config_dir)
+            # The deployment's own answer first: XDG_CONFIG_HOME is exported by
+            # the launcher and so exists only in the processes it started.
+            dep = deployment_config_dir()
+            base = str(dep) if dep else str(user_config_dir())
         except Exception:
             base = os.path.expanduser("~/.coderai")
     path = os.path.join(base, "runpod-pods.json")

codai/models/placement.py

@@ -91,12 +91,14 @@ class Lease:
     """One chosen upstream for one request. Always released."""
 
     def __init__(self, base: str, api_key: str, key: str, kind: str,
-                 release: Optional[Callable] = None):
+                 release: Optional[Callable] = None,
+                 dead: Optional[Callable] = None):
         self.base = base
         self.api_key = api_key
         self.key = key
         self.kind = kind
         self._release = release
+        self._dead = dead
         self._done = False
 
     def release(self) -> None:
@@ -110,6 +112,25 @@ class Lease:
             except Exception:
                 pass
 
+    def dead(self) -> None:
+        """Nothing answered at this address: tell the pool, then release.
+
+        A rented pod can vanish under us — RunPod reclaimed one mid-document
+        today, and the pool went on reporting it healthy. Every page then chose
+        the same corpse, the in-flight count climbed and nothing drained, which
+        is what a client sees as "requests neither complete nor time out". The
+        proxy backend has always reported this (RunpodBackend._unreachable);
+        a caller holding a lease has to do the same.
+        """
+        if self._dead is not None:
+            try:
+                self._dead()
+            except Exception:
+                pass
+            self._dead = None
+        forget()          # the candidate set named a pod that is gone
+        self.release()
+
     def __enter__(self):
         return self
 
@@ -277,7 +298,9 @@ def _pod_candidate(model_id: str) -> Optional[Candidate]:
         base, release = backend._acquire_endpoint()
         auth = (backend._headers or {}).get("Authorization", "")
         key = auth[7:].strip() if auth[:7].lower() == "bearer " else ""
-        return base, key, release
+        # _unreachable() drops the pod the backend just handed out, which is
+        # what stops the next page choosing the same dead one.
+        return base, key, release, backend._unreachable
 
     # `backend: runpod` on the entry means the pool IS this model's home, so it
     # comes before a node that happens to list the same id.
@@ -321,7 +344,8 @@ def _spill_pod_candidate(model_id: str) -> Optional[Candidate]:
     def _acquire():
         pod, url = pool.acquire()
         key = getattr(pool, "api_key", "") or ""
-        return (url.rstrip("/") + "/v1", key, lambda: pool.release(pod))
+        return (url.rstrip("/") + "/v1", key, lambda: pool.release(pod),
+                lambda: pool.discard(pod, "connection refused"))
 
     return Candidate(key=f"spill:{model_id}", kind="pod", base="", priority=3,
                      acquire=_acquire)
@@ -457,12 +481,16 @@ def choose(model_id: str, spill_after: int = 0) -> Lease:
 def _take(cand: Candidate) -> Optional[Lease]:
     """Acquire a pool candidate, or None when it cannot provide a server."""
     try:
-        base, key, release = cand.acquire()
+        got = cand.acquire()
     except Exception as exc:
         print(f"[ocr] {cand.key} cannot serve this page: {exc}", flush=True)
         return None
+    # A pool may also hand back a way to say "this one is dead"; older
+    # three-value forms simply have none.
+    base, key, release = got[0], got[1], got[2]
+    dead = got[3] if len(got) > 3 else None
     _inc(cand.key)
-    return Lease(base, key, cand.key, cand.kind, release)
+    return Lease(base, key, cand.key, cand.kind, release, dead)
 
 
 def spill_threshold(model_id: str) -> int:

codai/ocr/gateway.py

@@ -48,6 +48,30 @@ router = APIRouter()
 #: and it must not be one keystroke from the public ``/v1/ocr/engines``.
 PREFIX = "/v1/ocr/attach"
 
+#: Failures that mean "there is nothing at this address", as opposed to a
+#: server that answered slowly or badly. Only these drop the upstream: a read
+#: timeout is a model taking its time over a dense page, and discarding a
+#: perfectly good pod for that would be its own outage.
+def _dead_upstream_errors():
+    import httpx
+    return (httpx.ConnectError, httpx.ConnectTimeout, httpx.RemoteProtocolError,
+            ConnectionRefusedError, ConnectionResetError)
+
+
+def _is_dead_upstream(exc: BaseException) -> bool:
+    """Whether this failure means "nothing is at that address".
+
+    Written as a check rather than an exception class on purpose: ``except``
+    matches on the class hierarchy at C level and never consults
+    ``__instancecheck__``, so a clever proxy class silently catches nothing. A
+    test caught exactly that here.
+    """
+    try:
+        return isinstance(exc, _dead_upstream_errors())
+    except Exception:
+        return False
+
+
 #: How long one page may take upstream. A dense scan on a cold pod legitimately
 #: takes minutes; a timeout here would look to the engine like a failed page.
 UPSTREAM_TIMEOUT_S = 3600.0
@@ -255,34 +279,58 @@ async def chat_completions(token: str, request: Request):
     from codai.models import placement
     body = await request.body()
 
-    try:
-        # How busy is too busy is the MODEL's own setting, the same one the pool
-        # scales on — not a second, OCR-shaped copy of it.
-        chosen = placement.choose(
-            got["model_id"],
-            spill_after=placement.spill_threshold(got["model_id"]))
-    except Exception as exc:
-        status = int(getattr(exc, "status", 503) or 503)
-        return JSONResponse(status_code=status,
-                            content={"error": {"message": str(exc),
-                                               "type": "server_error"}})
-
-    try:
-        headers = {"Content-Type": request.headers.get("content-type",
-                                                       "application/json")}
-        if chosen.api_key:
-            headers["Authorization"] = f"Bearer {chosen.api_key}"
-        url = chosen.base.rstrip("/") + "/chat/completions"
-        reply = await _upstream_client(url).post(url, content=body, headers=headers)
-        # The reply goes back as it came: a re-serialised response is how
-        # logprobs got lost in the first place.
-        media = reply.headers.get("content-type", "application/json")
-        return Response(status_code=reply.status_code, content=reply.content,
-                        media_type=media)
-    except Exception as exc:
-        return JSONResponse(status_code=502,
-                            content={"error": {
-                                "message": f"OCR upstream {chosen.kind} failed: {exc}",
-                                "type": "server_error"}})
-    finally:
-        chosen.release()
+    content_type = request.headers.get("content-type", "application/json")
+    # Twice at most: a rented pod can vanish between being chosen and being
+    # asked (RunPod reclaimed one mid-document today). The first attempt reports
+    # the corpse so the pool drops it; the second runs against whatever replaces
+    # it. A third would just be a retry loop against a real outage.
+    last_dead = ""
+    for attempt in (1, 2):
+        try:
+            # How busy is too busy is the MODEL's own setting, the same one the
+            # pool scales on — not a second, OCR-shaped copy of it.
+            chosen = placement.choose(
+                got["model_id"],
+                spill_after=placement.spill_threshold(got["model_id"]))
+        except Exception as exc:
+            status = int(getattr(exc, "status", 503) or 503)
+            return JSONResponse(status_code=status,
+                                content={"error": {"message": str(exc),
+                                                   "type": "server_error"}})
+        try:
+            headers = {"Content-Type": content_type}
+            if chosen.api_key:
+                headers["Authorization"] = f"Bearer {chosen.api_key}"
+            url = chosen.base.rstrip("/") + "/chat/completions"
+            reply = await _upstream_client(url).post(url, content=body,
+                                                     headers=headers)
+            # The reply goes back as it came: a re-serialised response is how
+            # logprobs got lost in the first place.
+            media = reply.headers.get("content-type", "application/json")
+            return Response(status_code=reply.status_code, content=reply.content,
+                            media_type=media)
+        except Exception as exc:
+            if not _is_dead_upstream(exc):
+                return JSONResponse(status_code=502,
+                                    content={"error": {
+                                        "message": f"OCR upstream {chosen.kind} "
+                                                   f"failed: {exc}",
+                                        "type": "server_error"}})
+            # Nothing answered at the address at all. Report it so the pool
+            # stops handing out a pod that no longer exists — left unsaid, every
+            # following page picks the same one and the in-flight count climbs
+            # with nothing draining.
+            last_dead = f"{type(exc).__name__}: {exc}"
+            print(f"[ocr] upstream {chosen.base} is not answering "
+                  f"({last_dead}) — dropping it", flush=True)
+            chosen.dead()
+            chosen = None
+            continue
+        finally:
+            if chosen is not None:
+                chosen.release()
+    return JSONResponse(status_code=503,
+                        content={"error": {
+                            "message": f"no OCR server answered for "
+                                       f"{got['model_id']!r} ({last_dead})",
+                            "type": "server_error"}})

codai/platform_paths.py

@@ -56,6 +56,26 @@ def ensure_dir(path: Path) -> Path:
     return path
 
 
+def deployment_config_dir():
+    """The config dir this DEPLOYMENT uses, whatever started the process.
+
+    The container launcher sets ``CODERAI_CONFIG_DIR`` and keeps its state in
+    ``$CODERAI_CONFIG_DIR/coderai``. It also exports ``XDG_CONFIG_HOME`` to
+    match — but only into the processes IT starts, so anything entering the
+    container another way (``podman exec``, a cron job, a diagnostic script)
+    resolves ``~/.config/coderai`` instead and quietly gets its own pod
+    registry and its own pod TLS CA. Proven by accident: a one-off script in
+    the container printed a third registry path and created a second CA.
+
+    Returns None when the variable is unset, so a host install keeps whatever
+    it was already using.
+    """
+    env = os.environ.get("CODERAI_CONFIG_DIR")
+    if not env:
+        return None
+    return Path(env).expanduser() / APP_NAME
+
+
 def default_config_dir() -> Path:
     return ensure_dir(user_config_dir())
 

tests/test_ocr_fanout.py

@@ -1220,3 +1220,113 @@ def test_a_plain_http_upstream_is_not_pinned(monkeypatch):
     assert is_pinned_url("http://127.0.0.1:8000/v1") is False
     assert is_pinned_url("https://194.68.245.79:22198/v1") is True
     assert is_pinned_url("https://node.lan:8443/v1") is False
+
+
+# --------------------------------------- a pod can vanish under a document #
+# RunPod reclaimed an A40 mid-document today. The pool went on reporting it
+# healthy, so every page chose the same corpse: the in-flight count climbed and
+# nothing drained, which a client sees as "requests neither complete nor time
+# out". RunpodBackend has always reported this (_unreachable); a caller holding a
+# lease has to do the same.
+
+
+def _dead_then_alive(monkeypatch, failures=1):
+    """An upstream that refuses the first N connections, then answers."""
+    import httpx
+    state = {"n": 0}
+
+    class _Flaky(_CapturingClient):
+        @property
+        def is_closed(self):
+            return False
+
+        async def post(self, url, content=None, headers=None):
+            state["n"] += 1
+            if state["n"] <= failures:
+                raise httpx.ConnectError("connection refused")
+            return await super().post(url, content=content, headers=headers)
+
+    monkeypatch.setattr(httpx, "AsyncClient", _Flaky)
+    return state
+
+
+def _pod_cands(monkeypatch, discarded):
+    """Two successive pod leases, each able to report itself dead."""
+    def _acquire():
+        return ("https://10.0.0.9:8000/v1", "k", lambda: None,
+                lambda: discarded.append(1))
+    cand = dispatch.Candidate(key="pod:m", kind="pod", base="", priority=2,
+                              acquire=_acquire, primary=True)
+    monkeypatch.setattr(dispatch, "candidates", lambda m: [cand])
+    monkeypatch.setattr(dispatch, "_node_reported_load", lambda c: 0)
+
+
+def test_a_vanished_pod_is_reported_and_the_page_retried(monkeypatch):
+    _reset()
+    import asyncio
+    discarded = []
+    _dead_then_alive(monkeypatch, failures=1)
+    _pod_cands(monkeypatch, discarded)
+    token = gateway.issue("m", "m")
+    reply = asyncio.run(gateway.chat_completions(token, _Req(SURYA_BODY)))
+    assert discarded == [1]                  # the pool was told
+    assert reply.status_code == 200          # and the page still got read
+
+
+def test_it_gives_up_after_one_retry_rather_than_looping(monkeypatch):
+    """A real outage must not become a retry loop against every pod in turn."""
+    _reset()
+    import asyncio
+    discarded = []
+    _dead_then_alive(monkeypatch, failures=5)
+    _pod_cands(monkeypatch, discarded)
+    token = gateway.issue("m", "m")
+    reply = asyncio.run(gateway.chat_completions(token, _Req(SURYA_BODY)))
+    assert reply.status_code == 503
+    assert len(discarded) == 2               # both attempts reported, then stop
+
+
+def test_the_inflight_count_does_not_outlive_a_dropped_pod(monkeypatch):
+    """The symptom the client actually saw: a count that only grew."""
+    _reset()
+    import asyncio
+    discarded = []
+    _dead_then_alive(monkeypatch, failures=5)
+    _pod_cands(monkeypatch, discarded)
+    token = gateway.issue("m", "m")
+    asyncio.run(gateway.chat_completions(token, _Req(SURYA_BODY)))
+    assert dispatch.inflight() == 0
+
+
+def test_a_slow_server_is_not_mistaken_for_a_dead_one(monkeypatch):
+    """A read timeout is a model taking its time over a dense page. Discarding a
+    healthy pod for that would be an outage of its own."""
+    _reset()
+    import asyncio
+    import httpx
+    discarded = []
+
+    class _Slow(_CapturingClient):
+        @property
+        def is_closed(self):
+            return False
+
+        async def post(self, url, content=None, headers=None):
+            raise httpx.ReadTimeout("too slow")
+
+    monkeypatch.setattr(httpx, "AsyncClient", _Slow)
+    _pod_cands(monkeypatch, discarded)
+    token = gateway.issue("m", "m")
+    reply = asyncio.run(gateway.chat_completions(token, _Req(SURYA_BODY)))
+    assert reply.status_code == 502          # reported, not retried
+    assert discarded == []                   # and the pod kept
+    assert dispatch.inflight() == 0
+
+
+def test_the_candidate_set_is_rebuilt_after_a_death():
+    """It named a pod that is gone; keeping it cached would hand it out again."""
+    _reset()
+    dispatch._candidates["m"] = (9e9, [_cand("node:a")])
+    lease = dispatch.Lease("http://a/v1", "", "node:a", "node")
+    lease.dead()
+    assert "m" not in dispatch._candidates