ocr: a closed keep-alive socket is not a dead pod (v0.3.41)

Five A40s were terminated between 19:14 and 20:11 today. None of them was
broken: each pod's own log ended in `"POST /v1/chat/completions HTTP/1.1"
200 OK` at 4664 tokens/s, seconds before it was reaped, and every replacement
cost a ~6-minute image pull before it could read a page.

`httpx.RemoteProtocolError("Server disconnected without sending a response")`
was in the dead-upstream set. It does not mean the server is gone — it is what
httpx raises when a POOLED connection was closed by the far end between two
requests. uvicorn closes an idle connection after 5s and nothing here raised
that, while this gateway held pooled connections for 60s. So the first page
after any quiet gap hit a socket the pod had already closed, the gateway read
it as an empty address, and the pool terminated a rented GPU for it.

  - a closed pooled connection is retried on a fresh connection, against the
    SAME upstream: re-choosing would send the page to a different pod, or rent
    one;
  - a pod is asked before it is dropped — /healthz, then /health, on a new
    connection. Any answer under 500 counts as alive, a 401 included: that is
    a server, authenticating. Only silence drops it, because dropping is
    irreversible;
  - idle pooled connections are released after 4s
    (CODERAI_OCR_KEEPALIVE_EXPIRY_S), under the server's own keep-alive, so the
    race is rare rather than routine. Reuse within a burst is unaffected.

A pod RunPod really has reclaimed mid-document is still dropped at once, which
is what the original logic was for: left in the pool, every following page picks
the corpse and the in-flight count climbs with nothing draining. The tests pin
both halves, and a read timeout still neither probes nor drops.
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.40"
+__version__ = "0.3.41"
 __tagline__ = "Complete orchestration, distribution and escalation of remotizable advanced inference."
 
 # Configure the CUDA caching allocator BEFORE torch is imported anywhere.

codai/ocr/gateway.py

@@ -54,8 +54,25 @@ PREFIX = "/v1/ocr/attach"
 #: 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)
+    return (httpx.ConnectError, httpx.ConnectTimeout, ConnectionRefusedError)
+
+
+#: Failures that mean "the socket we reused was already closed" — which says
+#: nothing at all about the server.
+#:
+#: ``RemoteProtocolError("Server disconnected without sending a response")`` is
+#: what httpx raises when a POOLED keep-alive connection was closed by the far
+#: end between two requests. uvicorn closes an idle connection after 5s; this
+#: module used to hold them for 60 (see ``_KEEPALIVE_EXPIRY_S``), so the first
+#: page sent after a quiet gap raised it as a matter of course. It was in the
+#: dead-upstream set, so the pool dropped the pod and TERMINATED it: five A40s
+#: on 2026-10-12 between 19:14 and 20:11, every one of them mid-document and
+#: logging 200s at 4664 tokens/s when it was killed. A closed socket is retried
+#: on a fresh connection; only if that also fails does the address come into
+#: question.
+def _stale_socket_errors():
+    import httpx
+    return (httpx.RemoteProtocolError, ConnectionResetError)
 
 
 def _is_dead_upstream(exc: BaseException) -> bool:
@@ -72,6 +89,14 @@ def _is_dead_upstream(exc: BaseException) -> bool:
         return False
 
 
+def _is_stale_socket(exc: BaseException) -> bool:
+    """Whether this failure is a reused connection the server had closed."""
+    try:
+        return isinstance(exc, _stale_socket_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
@@ -90,7 +115,14 @@ UPSTREAM_TIMEOUT_S = 3600.0
 #: client bound to the closed one that used to live there.
 _MAX_CONNECTIONS = 256
 _MAX_KEEPALIVE = 64
-_KEEPALIVE_EXPIRY_S = 60.0
+#: How long an IDLE pooled connection may be kept. This must stay BELOW the
+#: upstream's own keep-alive timeout, or the pool hands out sockets the server
+#: has already closed: uvicorn's default is 5s and nothing in coderai raises it,
+#: so 60s here guaranteed a stale-socket error on the first page after any quiet
+#: gap. Retries now cover the race (``_is_stale_socket``); this keeps it rare.
+#: Reuse within a burst — the case this pooling exists for — is unaffected,
+#: since only connections idle longer than this are dropped.
+_KEEPALIVE_EXPIRY_S = float(os.environ.get("CODERAI_OCR_KEEPALIVE_EXPIRY_S") or 4.0)
 
 _lock = threading.Lock()
 _leases: dict = {}              # lease token -> {"model_id", "served_name"}
@@ -206,6 +238,66 @@ def _upstream_client(url: str = ""):
         return got
 
 
+async def _one_shot_client(url: str = ""):
+    """A client of its very own: no pooling, nothing reused.
+
+    Used for the retry after a stale socket and for the health probe. Both exist
+    precisely because the pooled connection was bad, so neither may be allowed
+    to draw from the same pool.
+    """
+    import httpx
+
+    try:
+        from codai.api.pod_http import is_pinned_url
+        pinned = bool(is_pinned_url(url))
+    except Exception:
+        pinned = False
+    verify = _pinned_ssl_context() if pinned else True
+    return httpx.AsyncClient(
+        timeout=UPSTREAM_TIMEOUT_S, verify=verify,
+        limits=httpx.Limits(max_connections=4, max_keepalive_connections=0))
+
+
+#: How long the confirming health probe may take. Short: it only has to
+#: establish that something is listening and speaking HTTP, and a pod busy with
+#: a dense page still answers /healthz promptly.
+_PROBE_TIMEOUT_S = 10.0
+
+
+async def _still_alive(base: str, api_key: str = "") -> bool:
+    """Whether anything is still serving at ``base``, asked on a new connection.
+
+    A pod is a rented GPU, and dropping it from the pool terminates it. One
+    failed request is not evidence enough for that: on 2026-10-12 five A40s were
+    killed mid-document, each still logging 200s, because a closed keep-alive
+    socket was read as a dead address. So the address gets asked directly before
+    anything irreversible happens.
+
+    Any HTTP answer counts, 401 included — that is a server, authenticating.
+    Only silence means nothing is there.
+    """
+    root = base.rstrip("/")
+    if root.endswith("/v1"):
+        root = root[: -len("/v1")]
+    headers = {"Authorization": f"Bearer {api_key}"} if api_key else {}
+    client = await _one_shot_client(base)
+    try:
+        for path in ("/healthz", "/health"):
+            try:
+                reply = await client.get(root + path, headers=headers,
+                                         timeout=_PROBE_TIMEOUT_S)
+            except Exception:
+                continue            # try the other path, then give up
+            if reply.status_code < 500:
+                return True
+        return False
+    finally:
+        try:
+            await client.aclose()
+        except Exception:
+            pass
+
+
 async def close_upstream_clients() -> None:
     """Close the pooled clients (process shutdown, or a test tidying up)."""
     with _lock:
@@ -297,20 +389,54 @@ async def chat_completions(token: str, request: Request):
             return JSONResponse(status_code=status,
                                 content={"error": {"message": str(exc),
                                                    "type": "server_error"}})
+        headers = {"Content-Type": content_type}
+        if chosen.api_key:
+            headers["Authorization"] = f"Bearer {chosen.api_key}"
+        url = chosen.base.rstrip("/") + "/chat/completions"
         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)
+            try:
+                reply = await _upstream_client(url).post(url, content=body,
+                                                         headers=headers)
+            except Exception as exc:
+                if not _is_stale_socket(exc):
+                    raise
+                # The connection we reused had already been closed. That is a
+                # race, not an outage: send the page again on a socket of its
+                # own. Retried here rather than by picking a new candidate,
+                # because the upstream we have is almost certainly fine and
+                # re-choosing would send the page to a different pod — or rent
+                # one.
+                print(f"[ocr] upstream {chosen.base} closed a pooled connection "
+                      f"({type(exc).__name__}) — resending on a fresh one",
+                      flush=True)
+                client = await _one_shot_client(url)
+                try:
+                    reply = await client.post(url, content=body, headers=headers)
+                finally:
+                    try:
+                        await client.aclose()
+                    except Exception:
+                        pass
             # 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):
+            if not (_is_dead_upstream(exc) or _is_stale_socket(exc)):
+                return JSONResponse(status_code=502,
+                                    content={"error": {
+                                        "message": f"OCR upstream {chosen.kind} "
+                                                   f"failed: {exc}",
+                                        "type": "server_error"}})
+            # Looks like nothing is there — but dropping a pod TERMINATES a
+            # rented GPU, so ask the address itself first, on a new connection.
+            # Skipping this cost five A40s in an hour, each killed while it was
+            # still answering 200s.
+            if await _still_alive(chosen.base, chosen.api_key):
+                print(f"[ocr] upstream {chosen.base} failed this page "
+                      f"({type(exc).__name__}: {exc}) but answers its health "
+                      f"check — keeping it", flush=True)
                 return JSONResponse(status_code=502,
                                     content={"error": {
                                         "message": f"OCR upstream {chosen.kind} "

docs/ocr.md

@@ -385,6 +385,40 @@ failed pages are re-read by the next engine instead of failing the request. Set
 explicit order (`"doctr, paddle"`) to override it, or `off` to get the original error
 back.
 
+### A closed keep-alive socket is not a dead pod
+
+A pod dropped from the pool is **terminated** — it is a rented GPU, and the bill stops
+when it does. So what counts as "nothing is at that address" has to be narrow, and for a
+while it was not.
+
+`httpx.RemoteProtocolError("Server disconnected without sending a response")` was in the
+dead-upstream set. That error does not mean the server is gone: it is what httpx raises
+when a **pooled** connection was closed by the far end between two requests. uvicorn
+closes an idle connection after 5s by default and nothing in coderai raised that, while
+the OCR gateway held pooled connections for 60s — so the first page sent after any quiet
+gap hit a socket the pod had already closed, and the pool killed the pod for it.
+
+On 2026-10-12 that cost five A40s between 19:14 and 20:11. None of them was broken: each
+pod's own log ended in `"POST /v1/chat/completions HTTP/1.1" 200 OK` at 4664 tokens/s,
+seconds before it was reaped, and every replacement needed a ~6-minute image pull before
+it could read a page. From outside it looked like OCR flapping; it was coderai killing
+its own workers.
+
+Three things changed, all on the orchestrator (no pod image is involved):
+
+* a closed pooled connection is **retried on a fresh connection**, against the *same*
+  upstream — re-choosing would send the page to a different pod, or rent one;
+* a pod is **asked before it is dropped**: `/healthz` (then `/health`) on a new
+  connection, with any answer below 500 counting as alive — a 401 is a server,
+  authenticating. Only silence drops it;
+* idle pooled connections are now released after 4s (`CODERAI_OCR_KEEPALIVE_EXPIRY_S`),
+  under the server's own keep-alive, so the race is rare rather than routine. Reuse
+  *within* a burst of pages — the reason the pool exists — is unaffected.
+
+A pod that RunPod really has reclaimed mid-document is still dropped immediately, which
+is what this logic was added for: left in the pool, every following page picks the corpse
+and the in-flight count climbs with nothing draining.
+
 ## Enable & configure
 
 Settings → **OCR** card, or `config.json` `"ocr"`. Key fields: `enabled`,

tests/test_ocr_fanout.py

@@ -1334,3 +1334,222 @@ def test_the_candidate_set_is_rebuilt_after_a_death():
     lease = dispatch.Lease("http://a/v1", "", "node:a", "node")
     lease.dead()
     assert "m" not in dispatch._candidates
+
+
+# ------------------------------- a closed keep-alive socket is not a dead pod #
+# On 2026-10-12 the OCR pool burned through five A40s between 19:14 and 20:11.
+# Not one of them was broken: each was killed mid-document while its own log
+# still read `"POST /v1/chat/completions HTTP/1.1" 200 OK` at 4664 tokens/s.
+# What killed them was `httpx.RemoteProtocolError("Server disconnected without
+# sending a response")` — the error httpx raises when a POOLED connection was
+# closed by the far end between two requests. uvicorn closes an idle connection
+# after 5s and nothing in coderai raises that; this module held them for 60. So
+# the first page after any quiet gap raised it, the gateway read it as "nothing
+# is at that address", and the pool TERMINATED a rented GPU. Each pod then cost
+# a 6-minute image pull to replace, which is what the operator saw as OCR
+# flapping.
+def _stale_socket(monkeypatch, failures=1, alive=True, fresh_ok=True):
+    """An upstream whose pooled connection is closed, but which is perfectly fine.
+
+    ``alive`` is whether its health check answers — the difference between a
+    closed socket and a pod that really has gone. ``fresh_ok`` is whether the
+    resend on a brand-new connection succeeds, which is how a stale socket is
+    told apart from a server that is dropping every connection it is given.
+    """
+    import httpx
+    state = {"posts": 0, "fresh_posts": 0, "probes": 0}
+
+    class _Stale(_CapturingClient):
+        def __init__(self, *a, **kw):
+            super().__init__(*a, **kw)
+            # A one-shot client is the one built with keep-alive disabled.
+            limits = kw.get("limits")
+            self.pooled = not (limits is not None
+                               and getattr(limits, "max_keepalive_connections", 1) == 0)
+
+        @property
+        def is_closed(self):
+            return False
+
+        async def aclose(self):
+            return None
+
+        async def get(self, url, headers=None, timeout=None):
+            state["probes"] += 1
+            if not alive:
+                raise httpx.ConnectError("connection refused")
+
+            class _R:
+                status_code = 200
+            return _R()
+
+        async def post(self, url, content=None, headers=None):
+            if self.pooled:
+                state["posts"] += 1
+                if state["posts"] <= failures:
+                    raise httpx.RemoteProtocolError(
+                        "Server disconnected without sending a response.")
+            else:
+                state["fresh_posts"] += 1
+                if not fresh_ok:
+                    raise httpx.RemoteProtocolError(
+                        "Server disconnected without sending a response.")
+            return await super().post(url, content=content, headers=headers)
+
+    monkeypatch.setattr(httpx, "AsyncClient", _Stale)
+    return state
+
+
+def test_a_closed_pooled_connection_is_not_a_dead_address():
+    """The classification itself: this is the line that cost five GPUs."""
+    import httpx
+    exc = httpx.RemoteProtocolError("Server disconnected without sending a response.")
+    assert gateway._is_stale_socket(exc)
+    assert not gateway._is_dead_upstream(exc)
+    # What a genuinely empty address looks like, by contrast.
+    assert gateway._is_dead_upstream(httpx.ConnectError("connection refused"))
+    assert not gateway._is_stale_socket(httpx.ConnectError("connection refused"))
+
+
+def test_a_stale_socket_resends_the_page_and_keeps_the_pod(monkeypatch):
+    """The page is read, on a connection of its own, and nothing is terminated."""
+    _reset()
+    import asyncio
+    discarded = []
+    state = _stale_socket(monkeypatch, failures=1)
+    _pod_cands(monkeypatch, discarded)
+    token = gateway.issue("m", "m")
+    reply = asyncio.run(gateway.chat_completions(token, _Req(SURYA_BODY)))
+    assert reply.status_code == 200          # the page was read
+    assert reply.body == UPSTREAM_REPLY      # and verbatim, as always
+    assert state["fresh_posts"] == 1         # resent off the pool
+    assert discarded == []                   # the $0.59/hr A40 is still rented
+    assert dispatch.inflight() == 0
+
+
+def test_the_resend_goes_to_the_same_pod_not_a_new_one(monkeypatch):
+    """Re-choosing would send the page to a different pod, or rent one. The
+    upstream we have is fine; only its socket was not."""
+    _reset()
+    import asyncio
+    seen = []
+    _stale_socket(monkeypatch, failures=1)
+
+    def _acquire():
+        seen.append(1)
+        return ("https://10.0.0.9:8000/v1", "k", lambda: None, lambda: None)
+    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)
+    token = gateway.issue("m", "m")
+    reply = asyncio.run(gateway.chat_completions(token, _Req(SURYA_BODY)))
+    assert reply.status_code == 200
+    assert len(seen) == 1                    # one lease, not two
+
+
+def test_a_pod_that_answers_its_health_check_is_never_terminated(monkeypatch):
+    """Even when the resend fails too: dropping a pod is irreversible, so the
+    address is asked directly before anything is billed away."""
+    _reset()
+    import asyncio
+    discarded = []
+    state = _stale_socket(monkeypatch, failures=99, alive=True, fresh_ok=False)
+    _pod_cands(monkeypatch, discarded)
+    token = gateway.issue("m", "m")
+    reply = asyncio.run(gateway.chat_completions(token, _Req(SURYA_BODY)))
+    assert reply.status_code == 502          # this page failed, honestly reported
+    assert discarded == []                   # but the pod kept
+    assert state["probes"] >= 1              # because it was asked
+    assert dispatch.inflight() == 0
+
+
+def test_a_pod_that_answers_nothing_at_all_is_still_dropped(monkeypatch):
+    """The original bug must stay fixed: a pod RunPod reclaimed mid-document
+    still has to leave the pool, or every following page picks the corpse."""
+    _reset()
+    import asyncio
+    discarded = []
+    _stale_socket(monkeypatch, failures=99, alive=False, fresh_ok=False)
+    _pod_cands(monkeypatch, discarded)
+    token = gateway.issue("m", "m")
+    reply = asyncio.run(gateway.chat_completions(token, _Req(SURYA_BODY)))
+    assert reply.status_code == 503          # both attempts, then give up
+    assert len(discarded) == 2               # and the pool was told, both times
+
+
+def test_the_health_probe_counts_a_401_as_alive(monkeypatch):
+    """A 401 is a server, authenticating. Only silence means nothing is there."""
+    import asyncio
+    import httpx
+
+    class _Guarded(_CapturingClient):
+        async def aclose(self):
+            return None
+
+        async def get(self, url, headers=None, timeout=None):
+            class _R:
+                status_code = 401
+            return _R()
+
+    monkeypatch.setattr(httpx, "AsyncClient", _Guarded)
+    assert asyncio.run(gateway._still_alive("https://10.0.0.9:8000/v1", "k"))
+
+
+def test_the_health_probe_asks_the_root_not_the_v1_path(monkeypatch):
+    """/healthz lives at the root; asking /v1/healthz would 404 and read as dead."""
+    import asyncio
+    import httpx
+    asked = []
+
+    class _Noting(_CapturingClient):
+        async def aclose(self):
+            return None
+
+        async def get(self, url, headers=None, timeout=None):
+            asked.append(url)
+            raise httpx.ConnectError("nothing here")
+
+    monkeypatch.setattr(httpx, "AsyncClient", _Noting)
+    assert not asyncio.run(gateway._still_alive("https://10.0.0.9:8000/v1"))
+    assert asked == ["https://10.0.0.9:8000/healthz",
+                     "https://10.0.0.9:8000/health"]
+
+
+def test_idle_connections_are_dropped_before_the_server_drops_them():
+    """uvicorn's default keep-alive is 5s. Holding a pooled connection longer
+    than the server does guarantees the stale-socket race on every quiet gap,
+    which is how this outage started."""
+    assert gateway._KEEPALIVE_EXPIRY_S < 5.0
+    # Still long enough to reuse a connection within a burst of pages, which is
+    # the whole reason the pool exists.
+    assert gateway._KEEPALIVE_EXPIRY_S >= 1.0
+
+
+def test_a_read_timeout_still_never_probes_or_drops(monkeypatch):
+    """Unchanged: a dense page is slow, not dead — and must not even cost a probe."""
+    _reset()
+    import asyncio
+    import httpx
+    discarded = []
+    probes = []
+
+    class _Slow(_CapturingClient):
+        @property
+        def is_closed(self):
+            return False
+
+        async def get(self, url, headers=None, timeout=None):
+            probes.append(url)
+            raise httpx.ConnectError("x")
+
+        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
+    assert discarded == []
+    assert probes == []