runpod: make a seeded pod actually serve its models, and survive a restart

Two bugs, both found by the first end-to-end embeddings test on a rented pod.

A capability image's entrypoint is `uvicorn codai.api.app:app` — it never runs
codai/main.py, so set_config_manager() was never called and
codai.admin.routes.config_manager stayed None. Both the CODERAI_SEED_MODELS list
and the runtime catalogue write read through that global, so a pod seeded with
one model answered "not available. Use one of: " with nothing after the colon.
The lifespan now builds its own ConfigManager when none was set — loading it is
what parses the seed list — and says how many seeds it registered.

The bearer token a pool launches its pods with is generated in memory when none
is configured. A pod that outlived the process that rented it was then adopted
with a DIFFERENT token and answered 401 for the rest of its paid life (the test
run reported serves: "(HTTP 401)"). The token is now recorded in the shared pod
registry and adopted along with the pod.

Verified live: bge-m3 on a rented RTX 2000 Ada returned a real 1024-dimension
embedding, 67.9s cold.
Co-Authored-By: 's avatarClaude Opus 5 (1M context) <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_0182DhR1eFNPQyGmDrpHbedY
parent e6aa7e3a
......@@ -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.2.9"
__version__ = "0.2.10"
# Configure the CUDA caching allocator BEFORE torch is imported anywhere.
# expandable_segments lets the allocator return freed pages to the driver even
......
......@@ -95,6 +95,38 @@ async def lifespan(app: FastAPI):
except Exception:
pass
# Models this instance was seeded with (CODERAI_SEED_MODELS). A pod image
# runs uvicorn on this app directly — it never executes codai/main.py — so
# the startup registration that lives there does not happen here, and a
# seeded pod would answer "not available" for everything. Registering in the
# lifespan covers BOTH entry points.
try:
import codai.admin.routes as _admin_routes
from codai.api.models_transfer import register_model_runtime
_cm = getattr(_admin_routes, "config_manager", None)
if _cm is None:
# Nobody built one: under bare uvicorn neither main.py nor
# build_system_app ran, so config_manager is still the module's
# initial None and BOTH the seed list and the catalogue write below
# would silently do nothing. Build it here — it is the same
# ConfigManager over the same config_dir the other entry points use,
# and loading it is what parses CODERAI_SEED_MODELS.
from codai.platform_paths import legacy_style_config_dir
from codai.config import ConfigManager
_cm = ConfigManager(str(legacy_style_config_dir()))
_cm.load()
_admin_routes.set_config_manager(_cm)
print("[seed] config manager bootstrapped for a bare-uvicorn start",
flush=True)
_seeds = [e for e in (getattr(_cm, "_pending_seed_entries", None) or [])
if isinstance(e, dict) and e.get("path")]
print(f"[seed] {len(_seeds)} seeded model(s) to register", flush=True)
for _entry in _seeds:
register_model_runtime(_entry,
_entry.get("model_type") or "text_models")
except Exception as _exc:
print(f"[seed] could not register seeded models: {_exc}", flush=True)
# Start any RunPod pod configured to stay warm. Pools are otherwise created
# on first use, so the first request after a restart would still pay the cold
# boot that `keep_warm` exists to avoid. In a thread: provisioning blocks for
......
......@@ -1382,7 +1382,7 @@ class RunpodPodPool:
"console_url": console, "started_at": time.time()}
# …and from every SIBLING engine's reaper, which cannot see the set
# above: each engine has its own pools and its own reaper.
register_pod(pod_id, str(self.model_key)) # url recorded once ready
register_pod(pod_id, str(self.model_key), api_key=self.api_key) # url once ready
print(f"[runpod] pod {pod_id} created for {self.model_key!r} "
f"({sel['display_name']} {sel['cloud_type']}) — booting; logs: {console}",
flush=True)
......@@ -1443,7 +1443,7 @@ class RunpodPodPool:
_provisioning_info.pop(pod_id, None)
# Publish the URL so a sibling engine needing the same pool reuses
# this pod instead of renting a second GPU for the same work.
register_pod(pod_id, str(self.model_key), url)
register_pod(pod_id, str(self.model_key), url, self.api_key)
print(f"[runpod] pod {pod_id} ready for {self.model_key!r} at {url}", flush=True)
return h
raise last_exc or RunpodError("RunPod: could not provision a pod.")
......@@ -1559,8 +1559,22 @@ class RunpodPodPool:
self._cv.notify_all()
self._cv.wait(15.0)
continue
shared_id, shared_url = find_shared_pod(
shared_id, shared_url, shared_key = find_shared_pod(
str(self.model_key), self.health_path, self.api_key)
if shared_url and shared_key and shared_key != self.api_key:
# That pod only answers to the token it was launched with.
# Adopt the token so this pool — and the pods it rents next —
# speak the same one. Only safe while we hold no pod of our
# own; otherwise leave it to be reaped rather than end up
# with a pool whose pods want two different tokens.
with self._cv:
adoptable = not self.pods
if not adoptable:
shared_id, shared_url = None, ""
else:
self.api_key = shared_key
print(f"[runpod] adopting pod {shared_id}'s bearer token",
flush=True)
if shared_url:
print(f"[runpod] reusing pod {shared_id} from another engine "
f"for {self.model_key!r}", flush=True)
......@@ -1954,18 +1968,25 @@ def _registry_update(fn):
return {}
def register_pod(pod_id: str, pool_key: str = "", url: str = "") -> None:
def register_pod(pod_id: str, pool_key: str = "", url: str = "",
api_key: str = "") -> None:
"""Record a pod as owned by this deployment.
Two reasons, both about processes that cannot see each other: no sibling's
reaper may treat it as an orphan, and a sibling that needs the SAME pool can
reuse this pod instead of renting a second GPU for the same work.
The bearer token goes in too. A pool that was not given one generates a
random token and launches its pods with it, so a pod that outlives the
process that rented it — a restart, a crash — is adopted by the next pool
with a DIFFERENT token and answers 401 to every request for the rest of its
paid life. Observed live: a surviving pod reported "serves: (HTTP 401)".
"""
if not pod_id:
return
_registry_update(lambda d: d.__setitem__(
str(pod_id), {"pool": str(pool_key), "pid": os.getpid(), "at": time.time(),
"url": str(url or "")})
"url": str(url or ""), "key": str(api_key or "")})
or True)
......@@ -1995,7 +2016,11 @@ def sibling_is_provisioning(pool_key: str) -> bool:
def find_shared_pod(pool_key: str, health_path: str = "/v1/models",
api_key: str = "") -> tuple:
"""A pod another process already has for this pool: (pod_id, url) or (None, '').
"""A pod another process already has for this pool: (pod_id, url, key).
``key`` is the token that pod was launched with, which is not necessarily
ours — see register_pod. Empty when unknown (an entry from before tokens were
recorded), in which case the caller's own token is tried.
coderai runs one engine per GPU, and each has its own pools — so the same
capability was rented a pod PER ENGINE, paying twice for one job while
......@@ -2004,7 +2029,7 @@ def find_shared_pod(pool_key: str, health_path: str = "/v1/models",
"""
data = _registry_update(lambda d: False)
if not isinstance(data, dict):
return None, ""
return None, "", ""
mine = os.getpid()
for pod_id, info in data.items():
if not isinstance(info, dict) or info.get("pool") != str(pool_key):
......@@ -2014,9 +2039,10 @@ def find_shared_pod(pool_key: str, health_path: str = "/v1/models",
url = str(info.get("url") or "")
if not url:
continue
if _pod_health_ok(url, path=health_path, api_key=api_key):
return pod_id, url
return None, ""
key = str(info.get("key") or "") or api_key
if _pod_health_ok(url, path=health_path, api_key=key):
return pod_id, url, key
return None, "", ""
def unregister_pod(pod_id: str) -> None:
......
......@@ -348,18 +348,18 @@ def test_a_sibling_engines_pod_is_reused_not_duplicated(tmp_path, monkeypatch):
"pod-from-nvidia", {"pool": "capability:embeddings", "pid": 999999,
"at": 0, "url": "http://pod-a:8000"}) or True)
pod_id, url = rw.find_shared_pod("capability:embeddings")
pod_id, url, key = rw.find_shared_pod("capability:embeddings")
assert (pod_id, url) == ("pod-from-nvidia", "http://pod-a:8000")
# A different pool must not borrow it.
assert rw.find_shared_pod("capability:images") == (None, "")
assert rw.find_shared_pod("capability:images") == (None, "", "")
# Our own pods are left to the local pool, not "borrowed" from ourselves.
import os
rw._registry_update(lambda d: d.__setitem__(
"pod-mine", {"pool": "capability:images", "pid": os.getpid(),
"at": 0, "url": "http://pod-b:8000"}) or True)
assert rw.find_shared_pod("capability:images") == (None, "")
assert rw.find_shared_pod("capability:images") == (None, "", "")
def test_a_borrowed_pod_is_released_not_terminated(monkeypatch):
......@@ -515,3 +515,30 @@ def test_a_sibling_waits_for_a_booting_pod_instead_of_renting_one(tmp_path, monk
"ancient", {"pool": "capability:video", "pid": 999998,
"at": 0, "url": ""}) or True)
assert rw.sibling_is_provisioning("capability:video") is False
def test_a_surviving_pods_token_is_remembered_across_a_restart(tmp_path, monkeypatch):
"""A pool generates a random bearer token and launches its pods with it. When
the process dies and the pod does not, the next pool generates a DIFFERENT
token and the surviving pod answers 401 to everything for the rest of its
paid life — observed live: the test run reported 'serves: (HTTP 401)'."""
import codai.api.runpod_worker as rw
monkeypatch.setattr(rw, "_pod_registry_path", lambda: str(tmp_path / "pods.json"))
seen = {}
def _health(url, path="/v1/models", api_key="", **kw):
seen["key"] = api_key
return api_key == "cra-original" # the pod only knows its own token
monkeypatch.setattr(rw, "_pod_health_ok", _health)
# The pod the dead process rented, with the token it was launched with.
rw.register_pod("pod-survivor", "capability:embeddings",
"http://pod-a:8000", "cra-original")
monkeypatch.setattr(rw.os, "getpid", lambda: 424242) # a NEW process
pod_id, url, key = rw.find_shared_pod("capability:embeddings",
api_key="cra-freshly-generated")
assert (pod_id, url, key) == ("pod-survivor", "http://pod-a:8000", "cra-original")
assert seen["key"] == "cra-original", "probed with the pod's token, not ours"
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