Skip to content

C
coderai
Commit 83e40d03 authored by Stefy Lanza (nextime / spora )'s avatar Stefy Lanza (nextime / spora )
Browse files
Options

a refused connection is backpressure, not a dead pod (v0.4.37) 

`discard()` terminated the pod, on the reasoning that a connection can
only be refused by a pod that no longer exists. A pod whose listen
backlog is full refuses connections while being perfectly alive and busy
-- and refuses them on /healthz too, so the "is anything still there?"
probe added to protect against exactly this agrees it has vanished.

Observed 2026-10-19, mid-campaign, ~58 concurrent connections per pod:

    08:36:13  pod 2n69g0vt47h0hl terminated (connection refused)
    08:36:30  pod c79lv31ayh28l1 terminated (connection refused)
    ... eight healthy A40s in 69 seconds ...
    08:37:22  pod n9fpbskd7owl2h terminated (connection refused)

522 pages were in flight on them. The warm floor then re-rented eight
more, which booted ~12 minutes, began refusing, and were killed again:
$3.72 of pods billed -- more than the campaign's whole spend -- on boots
that never produced a page, while throughput read 0.577 pages/s against
a measured 0.557 PER POD.

A pod is now QUARANTINED instead: out of rotation, still ours, still
counted against max_pods because it is still billing. `maintain`
re-probes it when the quarantine expires (45s) and

  * it answers -> straight back into rotation, which is the common case
    and the whole reason this stopped terminating;
  * it has work in flight -> left alone whatever the probe says, and it
    accrues no strikes while excused, so it does not die the moment the
    last page drains. Killing a pod with pages on it loses them AND pays
    for them;
  * it fails 3 probes with nothing in flight -> terminated. ~2 minutes of
    a pod that exists, bills and will not answer while carrying no load
    is dead in every sense that matters, so this does not trade a
    terminating bug for a leaking one.

A quarantined pod also counts toward the warm-floor deficit. Without
that, eight set-aside pods read as a deficit of eight and the pool rents
~700s of boot to cover a 45s gap, ending with two cards where the floor
asked for one -- the re-rent loop in one line.

The snapshot reports `quarantined` as its own state with the seconds
remaining, and `pods_quarantined` in the summary. It is NOT counted as
booting, which told an operator a settled fleet was still starting and
told a client sizing from pods_ready to wait for pods already present.

`discard` has exactly one caller (RunpodBackend._unreachable, reason
"connection refused"), so nothing else loses a termination it needed.

19 tests. Full suite 3777 passed, 15 pre-existing failures unchanged.
Co-Authored-By: 's avatarClaude Opus 5 <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_01KnfkxBpcZ1v6ypRFFyhrVM
parent 9ecc4122
Hide whitespace changes
Inline Side-by-side
Showing with 509 additions and 18 deletions
codai/__init__.py
View file @ 83e40d03
......@@ -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.36"
__version__ = "0.4.37"
__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
View file @ 83e40d03
......@@ -2647,6 +2647,21 @@ class PodHandle:
#: keeps serving what it already has: killing a pod with requests on it to
#: save money would be paying for the work twice.
draining: bool = False
#: Set aside after a connection-level failure, until this moment, and NOT
#: terminated. See `discard`.
#:
#: A pod whose listen backlog is full refuses connections while being
#: perfectly alive and busy, and refuses them on /healthz too -- so neither
#: the failed request nor the probe that follows it can tell "overloaded"
#: from "gone". Terminating on that evidence destroyed eight healthy A40s in
#: 69 seconds on 2026-10-19, with 522 pages in flight on them.
quarantined_until: float = 0.0
#: Consecutive post-quarantine probes that found nothing. Only a pod that
#: fails several of these, with no work on it, is treated as dead.
probe_failures: int = 0
#: Last time this pod answered anything at all. A pod that completed a
#: request a moment ago is not a candidate for "it must have vanished".
last_ok: float = 0.0
console_url: str = ""
#: How far its /logs have been drained into our own file.
log_cursor: int = 0
......@@ -4487,6 +4502,11 @@ class RunpodPodPool:
f"for one card)", flush=True)
p.healthy = True
p.last_used = time.time()
# Re-discovering a pod is positive evidence it is there, which
# is what the quarantine was waiting for. Clearing it keeps the
# two pieces of state from disagreeing about one pod.
p.quarantined_until = 0.0
p.probe_failures = 0
self._cv.notify_all()
return False
self.pods.append(handle)
......@@ -4494,9 +4514,92 @@ class RunpodPodPool:
self._cv.notify_all()
return True
def _recheck_quarantined(self, now: float) -> None:
"""Bring back, or finally give up on, the pods `discard` set aside.
The decision that must not be taken on one failed connection is taken
here instead, on evidence that separates the two cases:
* it answers again -> back into rotation. This is the common case, and
the whole reason `discard` stopped terminating.
* it has work in flight -> left alone whatever the probe says. Killing a
pod with pages on it loses those pages AND pays for them.
* it fails `QUARANTINE_STRIKES` probes in a row with nothing in flight
-> treated as dead and terminated. A pod that cannot be reached over
several minutes while carrying no load is not under backpressure.
The account listing remains the authority on a pod that no longer
exists; the reconcile sweep handles that case and closes the bill
without an API call. This is only about a pod that still exists and
will not talk to us.
"""
with self._cv:
due = [p for p in self.pods
if p.quarantined_until and now >= p.quarantined_until]
for pod in due:
if pod.inflight > 0:
# Still carrying work -- so something IS getting through, or the
# slots are stale and the reclaim sweep will lower them. Either
# way this is not a pod to destroy.
with self._cv:
pod.quarantined_until = now + QUARANTINE_S
continue
alive = self._pod_reachable(pod)
with self._cv:
if pod not in self.pods:
continue
if alive:
pod.healthy = True
pod.quarantined_until = 0.0
pod.probe_failures = 0
pod.last_ok = now
self._bump_cards()
self._cv.notify_all()
print(f"[runpod] pod {pod.pod_id} answers again — back in "
f"rotation for {self.model_key!r} (it was overloaded, "
f"not gone)", flush=True)
continue
pod.probe_failures += 1
strikes = pod.probe_failures
if strikes < QUARANTINE_STRIKES:
pod.quarantined_until = now + QUARANTINE_S
if not alive and strikes >= QUARANTINE_STRIKES:
with self._cv:
if pod in self.pods:
self.pods.remove(pod)
self._bump_cards()
print(f"[runpod] pod {pod.pod_id} failed {strikes} probes over "
f"~{QUARANTINE_S * strikes:.0f}s with nothing in flight — "
f"terminating it", flush=True)
threading.Thread(
target=self._terminate,
args=(pod, f"unreachable for {strikes} probes"),
daemon=True).start()
def _pod_reachable(self, pod: PodHandle) -> bool:
"""Does anything answer at this pod's address?
Unauthenticated /healthz on purpose: the question here is "is there a
server there", not "may we use it". A pod that answers anything at all,
including an error, exists -- and a pod that exists must not be
destroyed by this path.
"""
base = str(getattr(pod, "url", "") or "").rstrip("/")
if not base:
return False
if base.endswith("/v1"):
base = base[:-3].rstrip("/")
try:
from codai.api import pod_http
reply = pod_http.get(base + "/healthz", timeout=15)
except Exception:
return False
return int(getattr(reply, "status_code", 0) or 0) > 0
def release(self, pod: PodHandle):
with self._cv:
pod.inflight = max(0, pod.inflight - 1)
pod.last_ok = time.time()
# 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
......@@ -4508,22 +4611,54 @@ class RunpodPodPool:
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.
"""Set aside a pod that refused a connection. DOES NOT TERMINATE IT.
Between two timed probes a pod 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 for something the pool could have
known. Taking it out of rotation at once is right, and that part has not
changed: the caller retries on a fresh acquire.
KILLING IT IS NOT. This used to terminate, on the reasoning that a
refused connection means the pod is gone. It does not. A pod whose
listen backlog is full refuses connections while being perfectly alive
and busy -- and refuses them on /healthz too, so the "is anything still
there?" probe that was added to protect against exactly this agrees that
the pod has vanished.
Observed 2026-10-19, mid-campaign, ~58 concurrent connections per pod:
08:36:13 pod 2n69g0vt47h0hl terminated (connection refused)
08:36:30 pod c79lv31ayh28l1 terminated (connection refused)
... eight healthy A40s in 69 seconds ...
08:37:22 pod n9fpbskd7owl2h terminated (connection refused)
522 pages were in flight on them. The warm floor then re-rented eight
more, which booted for ~12 minutes, began refusing, and were killed
again -- $3.72 of pods billed, more than the campaign's whole spend,
on boots that never produced a page. Throughput was 0.577 pages/s
against a measured 0.557 PER POD.
A refusal under load is backpressure. So the pod is quarantined instead:
out of rotation, still ours, still counted against max_pods (it is still
billing, and pretending otherwise is how a pool rents a replacement for
a card it already has). `maintain` re-probes it when the quarantine
expires, and only a pod that fails several probes WITH NO WORK ON IT is
finally treated as dead -- see `_recheck_quarantined`.
"""
now = time.time()
with self._cv:
if pod not in self.pods:
return
pod.healthy = False
self.pods.remove(pod)
pod.quarantined_until = now + QUARANTINE_S
self._bump_cards()
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()
held = int(pod.inflight)
print(f"[runpod] pod {pod.pod_id} set aside for {QUARANTINE_S:.0f}s: "
f"{reason}. NOT terminated — a refused connection is as likely to "
f"mean the pod is overloaded as gone, and it still has {held} "
f"page(s) charged to it. It is re-probed when the quarantine "
f"expires.", flush=True)
def forget(self, pod: PodHandle, reason: str = "gone") -> bool:
"""Drop a pod that no longer exists. Like ``discard``, minus the kill.
......@@ -5093,6 +5228,10 @@ class RunpodPodPool:
except Exception:
pass
self._reclaim_stranded_inflight(now)
# Before anything else decides how many pods are serving: a quarantined
# pod is not healthy, so leaving it set aside inflates the warm-floor
# deficit and rents a replacement for a card we are already paying for.
self._recheck_quarantined(now)
self._scale_down(now)
with self._cv:
pods = list(self.pods)
......@@ -5208,7 +5347,15 @@ class RunpodPodPool:
# already running. Observed 2026-10-18: a pod adopted at 17:41,
# a restart at 17:43, a second pod rented at 17:46, $1.18/hr for
# a configuration that asked for $0.59.
healthy_in_hand = len([p for p in self.pods if p.healthy]) + self._booting
# A QUARANTINED POD COUNTS TOWARD THE FLOOR. It is not
# healthy, so it cannot serve a page right now -- but it is ours,
# it is billing, and it is expected back within QUARANTINE_S.
# Renting a replacement means ~700 s of paid boot to cover a 45 s
# gap, and then two cards where the floor asked for one. That is
# precisely the loop that turned eight refused connections into
# sixteen pods and $3.72 of boots on 2026-10-19.
healthy_in_hand = len([p for p in self.pods
if p.healthy or p.quarantined_until]) + self._booting
billing_in_hand = len(self.pods) + self._booting
parallel = max(1, int(getattr(self.mcfg,
"scale_up_max_parallel_boots", 4) or 4))
......@@ -5338,6 +5485,19 @@ def _forget_unavailable(gpu_type_id: str, dc: str = "") -> None:
#: holding a phantom a few minutes longer costs only those minutes.
INFLIGHT_TTL_S = 900.0
#: How long a pod that refused a connection stays out of rotation before it is
#: re-probed. Long enough for a full listen backlog to drain -- the condition
#: that makes a healthy pod look dead -- and short enough that a fleet does not
#: spend minutes at reduced width over a momentary burst.
QUARANTINE_S = 45.0
#: Consecutive failed probes, each a QUARANTINE_S apart and each with NO work in
#: flight, before a quarantined pod is finally terminated. Three is ~2 minutes
#: of a pod that exists, bills, and will not answer while carrying nothing:
#: past that it is dead in every sense that matters and keeping it costs money
#: for nothing. Fewer would start guessing again.
QUARANTINE_STRIKES = 3
def _slots_of(pod) -> list:
"""A pod's in-flight timestamp list, created on demand.
......@@ -5728,7 +5888,12 @@ def fleet_summary(pods: list = None) -> dict:
"""
rows = pods_status() if pods is None else pods
ready = [p for p in rows if p.get("state") == "ready" and p.get("healthy")]
booting = [p for p in rows if p.get("state") != "ready"]
quarantined = [p for p in rows if p.get("state") == "quarantined"]
# A set-aside pod is not booting. Counting it as one told an operator a
# settled fleet was still starting up, and told a client sizing its
# in-flight window from pods_ready to wait for pods that were already here.
booting = [p for p in rows
if p.get("state") not in ("ready", "quarantined")]
def _tally(key, src):
out = {}
......@@ -5740,6 +5905,7 @@ def fleet_summary(pods: list = None) -> dict:
return {
"pods_total": len(rows),
"pods_ready": len(ready),
"pods_quarantined": len(quarantined),
"pods_booting": len(booting),
"pods_unhealthy": len([p for p in rows
if p.get("state") == "ready" and not p.get("healthy")]),
......@@ -6142,7 +6308,18 @@ def pods_status() -> list:
"is_spot": p.is_spot, "healthy": p.healthy, "inflight": p.inflight,
"hourly_usd": p.hourly_usd, "uptime_s": int(now - p.started_at),
"live_cost_usd": round((now - p.started_at) / 3600.0 * p.hourly_usd, 4),
"console_url": p.console_url, "state": "ready",
"console_url": p.console_url,
# A quarantined pod is NOT "ready" and is not gone either.
# Reporting it as one or the other is how an operator sees
# either a fleet that is fine or a fleet that shrank, when
# what happened is that a pod was set aside and is coming
# back.
"state": ("quarantined"
if getattr(p, "quarantined_until", 0) else "ready"),
"quarantined_for_s": (
max(0, int(p.quarantined_until - now))
if getattr(p, "quarantined_until", 0) else 0),
"probe_failures": int(getattr(p, "probe_failures", 0) or 0),
"data_center": getattr(p, "data_center", "") or "",
"cloud_type": getattr(p, "cloud_type", "") or "",
"pool": (pool.mcfg.pool or "") if getattr(pool, "mcfg", None) else "",
......
codai/backends/runpod.py
View file @ 83e40d03
......@@ -332,11 +332,17 @@ class RunpodBackend(ModelBackend):
release()
def _unreachable(self) -> None:
"""The pod refused the connection: drop it and let the caller retry.
"""The pod refused the connection: set it aside 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.
`discard` takes it out of rotation and does NOT terminate it: a refused
connection is as likely to mean the pod's listen backlog is full as that
the pod is gone, and the pool re-probes it before deciding. Reporting a
refusal here is therefore cheap and safe, which it was not when this
call destroyed the card.
"""
pod = getattr(self, "_last_pod", None)
if pod is not None and self._pool is not None:
......
tests/test_runpod_client_scale.py
View file @ 83e40d03
......@@ -359,8 +359,8 @@ def test_the_warm_top_up_does_not_stack_on_a_booting_pod():
"""
src = __import__("pathlib").Path(rw.__file__).read_text()
blk = src.split("def maintain")[1].split("def close")[0]
assert ("healthy_in_hand = len([p for p in self.pods if p.healthy]) "
"+ self._booting") in blk
assert "if p.healthy or p.quarantined_until])" in blk
assert "healthy_in_hand" in blk
assert "billing_in_hand = len(self.pods) + self._booting" in blk
assert "want = min(keep_min - healthy_in_hand," in blk
# the cap must be the billing count, or the regression returns
......
tests/test_runpod_refusal_is_not_death.py 0 → 100644
View file @ 83e40d03
"""A refused connection is backpressure, not a verdict that the pod is gone.
`discard()` terminated the pod, on the reasoning that a connection can only be
refused by a pod that no longer exists. A pod whose listen backlog is full
refuses connections while being perfectly alive and busy -- and refuses them on
/healthz too, so the "is anything still there?" probe added to protect against
exactly this agrees that it has vanished.
Observed 2026-10-19, mid-campaign, ~58 concurrent connections per pod:
08:36:13 pod 2n69g0vt47h0hl terminated (connection refused); billed ~$0.417
08:36:30 pod c79lv31ayh28l1 terminated (connection refused); billed ~$0.449
... eight healthy A40s in 69 seconds, 522 pages in flight on them ...
08:37:22 pod n9fpbskd7owl2h terminated (connection refused); billed ~$0.458
The warm floor then re-rented eight more, which booted ~12 minutes, began
refusing, and were killed again: $3.72 of pods billed -- more than the whole
campaign's spend -- on boots that never produced a page. Measured throughput
0.577 pages/s against 0.557 PER POD.
What these tests pin, in order of what getting it wrong costs:
1. a refusal NEVER terminates. This is the defect; everything else is detail.
2. a pod with work in flight is never terminated by this path at all, however
many probes fail -- killing it loses those pages and pays for them.
3. a quarantined pod still counts toward the warm floor, or the pool rents
~700 s of boot to cover a 45 s gap and ends up with two cards for one.
4. a pod that answers again comes back by itself.
5. a pod that really is dead is still eventually terminated -- the fix must
not turn a leak into the new failure mode.
"""
import threading
import time
import types
import pytest
from codai.api import runpod_worker as rw
def _pool(max_pods=8, min_pods=0):
class _Acct:
enabled = True
p = rw.RunpodPodPool.__new__(rw.RunpodPodPool)
p._cv = threading.Condition()
p.model_key = "datalab-to/surya-ocr-2"
p._closed = False
p._booting = 0
p._waiting = 0
p._surplus_since = 0.0
p._denied = []
p.pods = []
p.account = _Acct()
p.api_key = "k"
cfg = rw.RunpodModelConfig()
cfg.max_pods = max_pods
cfg.min_pods = min_pods
p.mcfg = cfg
p._bump_cards = lambda: None
p._terminate = lambda pod, reason: _TERMINATED.append((pod.pod_id, reason))
return p
def _pod(pod_id="p1", inflight=0, url="https://10.0.0.1:2222"):
return rw.PodHandle(pod_id=pod_id, url=url, hourly_usd=0.59,
started_at=time.time(), gpu="A40", inflight=inflight)
_TERMINATED: list = []
@pytest.fixture(autouse=True)
def _clear():
_TERMINATED.clear()
yield
_TERMINATED.clear()
# --- 1. the regression: a refusal must not kill ---------------------------
def test_discard_does_not_terminate():
"""The eight A40s. This is the whole point."""
p = _pool()
pod = _pod()
p.pods = [pod]
p.discard(pod, "connection refused")
assert _TERMINATED == [], "a refused connection terminated the pod"
def test_discard_keeps_the_pod_in_the_pool():
"""It is still ours and still billing. A pool that forgets it rents a
replacement for a card it is already paying for."""
p = _pool()
pod = _pod()
p.pods = [pod]
p.discard(pod, "connection refused")
assert pod in p.pods
def test_discard_takes_it_out_of_rotation():
"""Right, and unchanged: the next page must not choose it."""
p = _pool()
pod = _pod()
p.pods = [pod]
p.discard(pod, "connection refused")
assert pod.healthy is False
assert pod.quarantined_until > time.time()
def test_discard_is_a_no_op_for_a_pod_we_do_not_hold():
p = _pool()
p.discard(_pod("stranger"), "connection refused")
assert _TERMINATED == []
# --- 2. work in flight is sacred -----------------------------------------
def test_a_pod_with_pages_on_it_is_never_terminated_by_this_path():
"""522 pages were in flight on the pods that were destroyed."""
p = _pool()
pod = _pod(inflight=64)
p.pods = [pod]
p.discard(pod, "connection refused")
p._pod_reachable = lambda _p: False
for _ in range(10): # far past the strike limit
pod.quarantined_until = time.time() - 1
p._recheck_quarantined(time.time())
assert _TERMINATED == [], "a pod carrying pages was terminated"
assert pod in p.pods
def test_in_flight_work_defers_the_probe_rather_than_passing_it():
"""It must not accumulate strikes while it is excused, or it dies the
moment the last page drains."""
p = _pool()
pod = _pod(inflight=1)
p.pods = [pod]
p.discard(pod, "connection refused")
p._pod_reachable = lambda _p: False
for _ in range(5):
pod.quarantined_until = time.time() - 1
p._recheck_quarantined(time.time())
assert pod.probe_failures == 0
# --- 3. the floor must not rent a replacement ----------------------------
def test_a_quarantined_pod_counts_toward_the_warm_floor():
"""Renting over a 45 s gap costs ~700 s of boot and leaves two cards where
the floor asked for one -- the re-rent loop, in one line."""
src = rw.RunpodPodPool.maintain.__doc__ or ""
import inspect
body = inspect.getsource(rw.RunpodPodPool.maintain)
assert ("healthy_in_hand = len([p for p in self.pods\n"
" if p.healthy or p.quarantined_until])"
in body), "the warm-floor deficit ignores quarantined pods"
def test_the_recheck_runs_before_anything_sizes_the_fleet():
import inspect
body = inspect.getsource(rw.RunpodPodPool.maintain)
assert body.index("_recheck_quarantined") < body.index("_scale_down")
# --- 4. recovery is automatic -------------------------------------------
def test_a_pod_that_answers_again_returns_to_rotation():
p = _pool()
pod = _pod()
p.pods = [pod]
p.discard(pod, "connection refused")
p._pod_reachable = lambda _p: True
pod.quarantined_until = time.time() - 1
p._recheck_quarantined(time.time())
assert pod.healthy is True
assert pod.quarantined_until == 0.0
assert pod.probe_failures == 0
assert _TERMINATED == []
def test_a_quarantine_is_not_rechecked_before_it_expires():
p = _pool()
pod = _pod()
p.pods = [pod]
p.discard(pod, "connection refused")
probed = []
p._pod_reachable = lambda _p: probed.append(1) or True
p._recheck_quarantined(time.time()) # still cooling
assert probed == []
assert pod.healthy is False
def test_recovery_clears_the_strikes_so_later_blips_start_fresh():
p = _pool()
pod = _pod()
p.pods = [pod]
p.discard(pod, "connection refused")
p._pod_reachable = lambda _p: False
pod.quarantined_until = time.time() - 1
p._recheck_quarantined(time.time())
assert pod.probe_failures == 1
p._pod_reachable = lambda _p: True
pod.quarantined_until = time.time() - 1
p._recheck_quarantined(time.time())
assert pod.probe_failures == 0
def test_rediscovering_a_pod_clears_its_quarantine():
"""_add_pod is positive evidence the pod is there, which is what the
quarantine was waiting for."""
p = _pool()
pod = _pod()
p.pods = [pod]
p.discard(pod, "connection refused")
p._add_pod(_pod(pod.pod_id, url=pod.url))
assert pod.quarantined_until == 0.0
assert pod.healthy is True
# --- 5. a pod that really is dead still goes ----------------------------
def test_a_pod_that_never_answers_with_no_work_is_terminated():
"""The fix must not turn a terminating bug into a leaking one."""
p = _pool()
pod = _pod()
p.pods = [pod]
p.discard(pod, "connection refused")
p._pod_reachable = lambda _p: False
for _ in range(rw.QUARANTINE_STRIKES):
pod.quarantined_until = time.time() - 1
p._recheck_quarantined(time.time())
assert [t[0] for t in _TERMINATED] == [pod.pod_id]
assert pod not in p.pods
def test_it_takes_several_strikes_not_one():
p = _pool()
pod = _pod()
p.pods = [pod]
p.discard(pod, "connection refused")
p._pod_reachable = lambda _p: False
pod.quarantined_until = time.time() - 1
p._recheck_quarantined(time.time())
assert _TERMINATED == [], "one failed probe is still one failed probe"
assert rw.QUARANTINE_STRIKES >= 2
def test_the_give_up_window_is_minutes_not_seconds():
"""Long enough that a full backlog has certainly drained."""
assert rw.QUARANTINE_S * rw.QUARANTINE_STRIKES >= 90
def test_an_unreachable_probe_is_anything_that_does_not_answer():
"""A pod that answers an ERROR exists, and must not be destroyed here."""
p = _pool()
pod = _pod()
p.pods = [pod]
calls = []
class _R:
status_code = 503
import codai.api.pod_http as ph
old = ph.get
ph.get = lambda url, **kw: calls.append(url) or _R()
try:
assert p._pod_reachable(pod) is True
finally:
ph.get = old
assert calls and calls[0].endswith("/healthz")
def test_a_probe_that_raises_is_unreachable():
p = _pool()
pod = _pod()
import codai.api.pod_http as ph
old = ph.get
def _boom(url, **kw):
raise OSError("connection refused")
ph.get = _boom
try:
assert p._pod_reachable(pod) is False
finally:
ph.get = old
# --- the operator can see it --------------------------------------------
def test_the_snapshot_reports_quarantine_as_its_own_state():
src = open("codai/api/runpod_worker.py").read()
assert '"state": ("quarantined"' in src
assert '"quarantined_for_s"' in src
assert '"pods_quarantined": len(quarantined)' in src
def test_a_quarantined_pod_is_not_counted_as_booting():
"""It told an operator a settled fleet was still starting, and told a
client to wait for pods that were already here."""
import inspect
body = inspect.getsource(rw.fleet_summary)
assert 'p.get("state") not in ("ready", "quarantined")' in body
tests/test_runpod_warm_floor_respects_max_pods.py
View file @ 83e40d03
......@@ -104,7 +104,12 @@ def test_an_unhealthy_pod_does_not_satisfy_the_floor_when_there_is_room():
def test_worker_uses_two_distinct_counts():
src = open("codai/api/runpod_worker.py").read()
assert "billing_in_hand = len(self.pods) + self._booting" in src
assert "healthy_in_hand = len([p for p in self.pods if p.healthy]) + self._booting" in src
# The deficit count also credits a QUARANTINED pod since v0.4.37: it is
# ours and billing and due back in seconds, so renting over it buys ~700s
# of boot to cover a 45s gap. Still a count of pods that can SERVE soon,
# as distinct from pods that BILL.
assert "if p.healthy or p.quarantined_until])" in src
assert "healthy_in_hand" in src and "+ self._booting" in src
assert "max(0, self.effective_max_pods() - billing_in_hand)" in src, \
"the max_pods cap must be tested against pods that bill"
assert "want = min(keep_min - healthy_in_hand," in src, \
......
Markdown is supported
0% or
You are about to add 0 people to the discussion. Proceed with caution.
Finish editing this message first!
Please register or to comment