runpod: drop a pod that refuses connections and retry once on a fresh one

A pod terminated behind the pool's back — the harness reaping by API, a
reclaimed spot machine — stayed 'healthy' until the next timed probe,
and the next request failed on connection refused for something the
pool could have known. The backend now discards it immediately and
retries the request once on a fresh acquire; a second refusal is a
real outage and raises.
Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_01FKm7Z56iKjHgNkHc6VhWKz

codai/api/runpod_worker.py

@@ -2164,6 +2164,23 @@ class RunpodPodPool:
             pod.last_used = time.time()
             self._cv.notify_all()
 
+    def discard(self, pod: PodHandle, reason: str = "unreachable") -> None:
+        """Drop a pod that refused a connection, now, not at the next sweep.
+
+        The maintenance loop probes on a timer; between two probes a pod that
+        was terminated behind the pool's back (the test harness reaping by
+        API, RunPod reclaiming a spot machine) is still handed out, and the
+        request fails with "connection refused" for something the pool could
+        have known. The caller retries on a fresh acquire.
+        """
+        with self._cv:
+            if pod not in self.pods:
+                return
+            pod.healthy = False
+            self.pods.remove(pod)
+        print(f"[runpod] pod {pod.pod_id} dropped from the pool: {reason}", flush=True)
+        threading.Thread(target=self._terminate, args=(pod, reason), daemon=True).start()
+
     def _terminate(self, pod: PodHandle, reason: str):
         if getattr(pod, "gpu", "") == "(shared)":
             # Borrowed from another engine: it owns the lifecycle and the bill.

codai/backends/runpod.py

@@ -154,8 +154,20 @@ class RunpodBackend(ModelBackend):
         pod, url = self._pool.acquire()
         # The pod runs a vLLM OpenAI server at /v1; its proxy URL has no path, so
         # add /v1 to match the serverless base (which already ends in /openai/v1).
+        self._last_pod = pod
         return url.rstrip("/") + "/v1", (lambda: self._pool.release(pod))
 
+    def _unreachable(self) -> None:
+        """The pod refused the connection: drop it and let the caller retry.
+
+        A pod terminated behind the pool's back (the test harness reaping by
+        API, a reclaimed spot machine) stays "healthy" until the next timed
+        probe; without this, one request in that window fails for nothing.
+        """
+        pod = getattr(self, "_last_pod", None)
+        if pod is not None and self._pool is not None:
+            self._pool.discard(pod, "connection refused")
+
     def _store_usage(self, usage: dict) -> None:
         if usage:
             self._last_usage = {
@@ -193,38 +205,62 @@ class RunpodBackend(ModelBackend):
                       top_p=1.0, stop=None, tools=None, response_format=None):
         import requests
         self._enter_request()
-        base, release = self._acquire_endpoint()
         try:
-            payload = self._chat_payload(messages, max_tokens, temperature, top_p, stop, False)
-            if response_format and response_format.get("type") == "json_object":
-                payload["response_format"] = {"type": "json_object"}
-            if tools:
-                payload["tools"] = tools
-            r = pod_http.post(base + "/chat/completions", json=payload,
-                              headers=self._headers, timeout=3600)
-            r.raise_for_status()
-            data = r.json()
-            self._store_usage(data.get("usage", {}))
-            return data["choices"][0]["message"].get("content") or ""
+            for attempt in (1, 2):
+                base, release = self._acquire_endpoint()
+                try:
+                    payload = self._chat_payload(messages, max_tokens, temperature, top_p, stop, False)
+                    if response_format and response_format.get("type") == "json_object":
+                        payload["response_format"] = {"type": "json_object"}
+                    if tools:
+                        payload["tools"] = tools
+                    r = pod_http.post(base + "/chat/completions", json=payload,
+                                      headers=self._headers, timeout=3600)
+                except requests.exceptions.ConnectionError:
+                    # Nothing answered at all: the pod is gone. Once, on a
+                    # fresh pod — a second refusal is a real outage.
+                    release()
+                    self._unreachable()
+                    if attempt == 2:
+                        raise
+                    continue
+                try:
+                    r.raise_for_status()
+                    data = r.json()
+                    self._store_usage(data.get("usage", {}))
+                    return data["choices"][0]["message"].get("content") or ""
+                finally:
+                    release()
         finally:
-            release()
             self._exit_request()
 
     async def generate_chat_stream(self, messages: List[Dict], max_tokens=None,
                                    temperature=0.7, top_p=1.0, stop=None, tools=None,
                                    response_format=None) -> AsyncGenerator[str, None]:
+        import requests
         self._enter_request()
-        # Pod acquire may provision (blocking) — do it off the event loop.
-        base, release = await asyncio.to_thread(self._acquire_endpoint)
         try:
-            payload = self._chat_payload(messages, max_tokens, temperature, top_p, stop, True)
-            if tools:
-                payload["tools"] = tools
-            async for chunk in self._stream(base + "/chat/completions", payload,
-                                            delta_key="delta"):
-                yield chunk
+            for attempt in (1, 2):
+                # Pod acquire may provision (blocking) — do it off the event loop.
+                base, release = await asyncio.to_thread(self._acquire_endpoint)
+                payload = self._chat_payload(messages, max_tokens, temperature, top_p, stop, True)
+                if tools:
+                    payload["tools"] = tools
+                try:
+                    async for chunk in self._stream(base + "/chat/completions", payload,
+                                                    delta_key="delta"):
+                        yield chunk
+                except requests.exceptions.ConnectionError:
+                    # Refused before a byte arrived: the pod is gone. Drop it
+                    # and go again on a fresh one, once.
+                    self._unreachable()
+                    if attempt == 2:
+                        raise
+                    continue
+                finally:
+                    release()
+                return
         finally:
-            release()
             self._exit_request()
 
     # ------------------------------------------------------------------ #

tests/test_burst_placement.py

@@ -1222,3 +1222,45 @@ def test_the_gguf_backend_can_size_the_card_without_torch(monkeypatch):
     prof = Path("packaging/runpod/profiles")
     assert (prof / "llama.light").exists()
     assert "nvidia-ml-py" in (prof / "llama.txt").read_text()
+
+
+def test_a_pod_that_refuses_connections_is_dropped_and_the_request_retried(monkeypatch):
+    """The harness reaped a pod by API; the engine's pool still held it as
+    healthy until the next timed probe, and the next request died on
+    'connection refused' for something the pool could have known."""
+    import requests
+    from codai.backends import runpod as rb
+    from codai.api.runpod_worker import PodHandle
+
+    dead = PodHandle(pod_id="dead", url="https://203.0.113.9:1", hourly_usd=0.3, started_at=0)
+    live = PodHandle(pod_id="live", url="https://203.0.113.10:1", hourly_usd=0.3, started_at=0)
+    handed = iter([dead, live])
+    class _Pool:
+        pods = [dead, live]; discarded = []
+        def acquire(self, timeout=1200.0, affinity=""):
+            p = next(handed); return p, p.url
+        def release(self, p): pass
+        def discard(self, p, reason=""): self.discarded.append((p.pod_id, reason)); self.pods.remove(p)
+    pool = _Pool()
+
+    b = rb.RunpodBackend.__new__(rb.RunpodBackend)
+    b._mode = "pods"; b._pool = pool; b._headers = {}; b._url = ""
+    b._enter_request = lambda: None; b._exit_request = lambda: None
+    b._chat_payload = lambda *a, **k: {"messages": []}
+    b._store_usage = lambda u: None
+
+    calls = []
+    class _R:
+        status_code = 200
+        def raise_for_status(self): pass
+        def json(self): return {"choices": [{"message": {"content": "hi"}}], "usage": {}}
+    def _post(url, **kw):
+        calls.append(url)
+        if "203.0.113.9" in url:
+            raise requests.exceptions.ConnectionError("refused")
+        return _R()
+    monkeypatch.setattr(rb.pod_http, "post", _post)
+
+    assert b.generate_chat([{"role": "user", "content": "x"}]) == "hi"
+    assert [u.split("/")[2] for u in calls] == ["203.0.113.9:1", "203.0.113.10:1"]
+    assert pool.discarded == [("dead", "connection refused")] and dead not in pool.pods