runpod: a restart rented a second pod beside its own (v0.2.97)

A warm floor is re-established at every boot, and ensure_ready() went straight
to _provision_one(). The adoption logic — "a pod for this pool is already
running, take it over" — existed only inside acquire(), so the warm path never
saw it. Three restarts on the production orchestrator this afternoon rented
three A40s and left each previous one billing until the reaper's 20-minute
grace expired. An hourly unattended upgrade would have done that every hour.

find_shared_pod already covers the case: after a restart the recorded pid no
longer matches, so our own previous pod is adoptable. It is now one method,
_adopt_shared_pod, called from acquire(), ensure_ready() and maintain()'s warm
top-up.

It also stops lying about the cost. An adopted pod was recorded at $0/hr as
"billed by its owner", which is true of a sibling engine's pod and false of one
we are inheriting from our own restart — a real A40 hidden from the $/hr cap and
from the spend page. register_pod now stores the rate and adoption restores it;
an older registry entry without one still adopts, at 0.
Co-Authored-By: 's avatarClaude Opus 5 <noreply@anthropic.com>
parent afdc1d28
...@@ -16,7 +16,7 @@ ...@@ -16,7 +16,7 @@
# Canonical product version for CoderAI — single source of truth. Both the API # Canonical product version for CoderAI — single source of truth. Both the API
# metadata and the admin web UI read from here. # metadata and the admin web UI read from here.
__version__ = "0.2.96" __version__ = "0.2.97"
__tagline__ = "Complete orchestration, distribution and escalation of remotizable advanced inference." __tagline__ = "Complete orchestration, distribution and escalation of remotizable advanced inference."
# Configure the CUDA caching allocator BEFORE torch is imported anywhere. # Configure the CUDA caching allocator BEFORE torch is imported anywhere.
......
...@@ -2196,6 +2196,45 @@ class RunpodPodPool: ...@@ -2196,6 +2196,45 @@ class RunpodPodPool:
# NEXT candidate rather than failing the request. # NEXT candidate rather than failing the request.
MAX_PROVISION_ATTEMPTS = 3 MAX_PROVISION_ATTEMPTS = 3
def _adopt_shared_pod(self) -> bool:
"""Take over a pod already running for this pool, instead of renting.
Two cases, one mechanism: a sibling engine booted one for the same pool,
or WE booted one before the last restart. The second is the expensive
one — a warm floor is re-established at every boot, so an hourly
unattended upgrade rented a fresh A40 each time and left the previous
one billing until the reaper's 20-minute grace expired.
"""
shared_id, shared_url, shared_key, shared_rate = find_shared_pod(
str(self.model_key), self.health_path, self.api_key)
if not shared_url:
return False
if 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 we end
# up with a pool whose pods want two different tokens.
with self._cv:
if self.pods:
return False
self.api_key = shared_key
print(f"[runpod] adopting pod {shared_id}'s bearer token", flush=True)
print(f"[runpod] reusing the pod already running for {self.model_key!r} "
f"({shared_id}) instead of renting another", flush=True)
handle = PodHandle(pod_id=shared_id, url=shared_url,
# Its recorded price when we know it: a pod we are
# inheriting after our own restart is ours to pay for,
# and reporting it free hides it from the $/hr cap.
hourly_usd=float(shared_rate or 0.0),
started_at=time.time(), gpu="(adopted)",
healthy=True, last_used=time.time())
with self._cv:
self.pods.append(handle)
self._cv.notify_all()
register_pod(shared_id, str(self.model_key), shared_url, self.api_key,
hourly_usd=float(shared_rate or 0.0))
return True
def _data_center(self) -> str: def _data_center(self) -> str:
"""Which RunPod region this pool's pods go to. """Which RunPod region this pool's pods go to.
...@@ -2310,7 +2349,8 @@ class RunpodPodPool: ...@@ -2310,7 +2349,8 @@ class RunpodPodPool:
_provisioning_info.pop(pod_id, None) _provisioning_info.pop(pod_id, None)
# Publish the URL so a sibling engine needing the same pool reuses # Publish the URL so a sibling engine needing the same pool reuses
# this pod instead of renting a second GPU for the same work. # this pod instead of renting a second GPU for the same work.
register_pod(pod_id, str(self.model_key), url, self.api_key) register_pod(pod_id, str(self.model_key), url, self.api_key,
hourly_usd=sel["price"])
print(f"[runpod] pod {pod_id} ready for {self.model_key!r} at {url}", flush=True) print(f"[runpod] pod {pod_id} ready for {self.model_key!r} at {url}", flush=True)
return h return h
raise last_exc or RunpodError("RunPod: could not provision a pod.") raise last_exc or RunpodError("RunPod: could not provision a pod.")
...@@ -2330,6 +2370,11 @@ class RunpodPodPool: ...@@ -2330,6 +2370,11 @@ class RunpodPodPool:
return return
self._provisioning = True self._provisioning = True
try: try:
# A pod for this pool may already be up — a sibling engine's, or our
# own from before the last restart. Renting beside it is how a warm
# floor of one became two A40s at every boot.
if self._adopt_shared_pod():
return
for _ in range(want): for _ in range(want):
try: try:
self._provision_one() self._provision_one()
...@@ -2437,31 +2482,8 @@ class RunpodPodPool: ...@@ -2437,31 +2482,8 @@ class RunpodPodPool:
self._cv.notify_all() self._cv.notify_all()
self._cv.wait(15.0) self._cv.wait(15.0)
continue continue
shared_id, shared_url, shared_key = find_shared_pod( if self._adopt_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)
handle = PodHandle(pod_id=shared_id, url=shared_url,
hourly_usd=0.0, # billed by its owner
started_at=time.time(), gpu="(shared)",
healthy=True, last_used=time.time())
with self._cv: with self._cv:
self.pods.append(handle)
self._provisioning = False self._provisioning = False
self._cv.notify_all() self._cv.notify_all()
continue continue
...@@ -2636,7 +2658,8 @@ class RunpodPodPool: ...@@ -2636,7 +2658,8 @@ class RunpodPodPool:
return return
self._provisioning = True self._provisioning = True
try: try:
self._provision_one() if not self._adopt_shared_pod():
self._provision_one()
except Exception as exc: except Exception as exc:
print(f"[runpod] maintain: warm provision failed: {exc}", flush=True) print(f"[runpod] maintain: warm provision failed: {exc}", flush=True)
finally: finally:
...@@ -3079,7 +3102,7 @@ def _registry_update(fn): ...@@ -3079,7 +3102,7 @@ def _registry_update(fn):
def register_pod(pod_id: str, pool_key: str = "", url: str = "", def register_pod(pod_id: str, pool_key: str = "", url: str = "",
api_key: str = "") -> None: api_key: str = "", hourly_usd: float = 0.0) -> None:
"""Record a pod as owned by this deployment. """Record a pod as owned by this deployment.
Two reasons, both about processes that cannot see each other: no sibling's Two reasons, both about processes that cannot see each other: no sibling's
...@@ -3096,7 +3119,10 @@ def register_pod(pod_id: str, pool_key: str = "", url: str = "", ...@@ -3096,7 +3119,10 @@ def register_pod(pod_id: str, pool_key: str = "", url: str = "",
return return
_registry_update(lambda d: d.__setitem__( _registry_update(lambda d: d.__setitem__(
str(pod_id), {"pool": str(pool_key), "pid": os.getpid(), "at": time.time(), str(pod_id), {"pool": str(pool_key), "pid": os.getpid(), "at": time.time(),
"url": str(url or ""), "key": str(api_key or "")}) "url": str(url or ""), "key": str(api_key or ""),
# So a pod adopted after a restart is still attributed its
# real $/hr instead of looking free.
"hourly_usd": float(hourly_usd or 0.0)})
or True) or True)
...@@ -3126,7 +3152,7 @@ def sibling_is_provisioning(pool_key: str) -> bool: ...@@ -3126,7 +3152,7 @@ def sibling_is_provisioning(pool_key: str) -> bool:
def find_shared_pod(pool_key: str, health_path: str = "/v1/models", def find_shared_pod(pool_key: str, health_path: str = "/v1/models",
api_key: str = "") -> tuple: api_key: str = "") -> tuple:
"""A pod another process already has for this pool: (pod_id, url, key). """A pod already running for this pool: (pod_id, url, key, hourly_usd).
``key`` is the token that pod was launched with, which is not necessarily ``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 ours — see register_pod. Empty when unknown (an entry from before tokens were
...@@ -3136,6 +3162,12 @@ def find_shared_pod(pool_key: str, health_path: str = "/v1/models", ...@@ -3136,6 +3162,12 @@ def find_shared_pod(pool_key: str, health_path: str = "/v1/models",
capability was rented a pod PER ENGINE, paying twice for one job while capability was rented a pod PER ENGINE, paying twice for one job while
max_pods said 1. Checked against the shared registry and health-probed, so a max_pods said 1. Checked against the shared registry and health-probed, so a
dead entry is never handed out. dead entry is never handed out.
It also covers OUR OWN pod from before a restart: the pid no longer matches,
so the entry is adoptable. That matters most for a warm floor, which is
re-established at every boot — an hourly unattended upgrade would otherwise
rent a fresh A40 each time and leave the last one billing until the reaper's
grace ran out.
""" """
data = _registry_update(lambda d: False) data = _registry_update(lambda d: False)
if not isinstance(data, dict): if not isinstance(data, dict):
...@@ -3151,8 +3183,8 @@ def find_shared_pod(pool_key: str, health_path: str = "/v1/models", ...@@ -3151,8 +3183,8 @@ def find_shared_pod(pool_key: str, health_path: str = "/v1/models",
continue continue
key = str(info.get("key") or "") or api_key key = str(info.get("key") or "") or api_key
if _pod_health_ok(url, path=health_path, api_key=key): if _pod_health_ok(url, path=health_path, api_key=key):
return pod_id, url, key return pod_id, url, key, float(info.get("hourly_usd") or 0.0)
return None, "", "" return None, "", "", 0.0
def unregister_pod(pod_id: str) -> 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): ...@@ -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, "pod-from-nvidia", {"pool": "capability:embeddings", "pid": 999999,
"at": 0, "url": "http://pod-a:8000"}) or True) "at": 0, "url": "http://pod-a:8000"}) or True)
pod_id, url, key = rw.find_shared_pod("capability:embeddings") pod_id, url, key, rate = rw.find_shared_pod("capability:embeddings")
assert (pod_id, url) == ("pod-from-nvidia", "http://pod-a:8000") assert (pod_id, url) == ("pod-from-nvidia", "http://pod-a:8000")
# A different pool must not borrow it. # A different pool must not borrow it.
assert rw.find_shared_pod("capability:images") == (None, "", "") assert rw.find_shared_pod("capability:images") == (None, "", "", 0.0)
# Our own pods are left to the local pool, not "borrowed" from ourselves. # Our own pods are left to the local pool, not "borrowed" from ourselves.
import os import os
rw._registry_update(lambda d: d.__setitem__( rw._registry_update(lambda d: d.__setitem__(
"pod-mine", {"pool": "capability:images", "pid": os.getpid(), "pod-mine", {"pool": "capability:images", "pid": os.getpid(),
"at": 0, "url": "http://pod-b:8000"}) or True) "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, "", "", 0.0)
def test_a_borrowed_pod_is_released_not_terminated(monkeypatch): def test_a_borrowed_pod_is_released_not_terminated(monkeypatch):
...@@ -538,7 +538,7 @@ def test_a_surviving_pods_token_is_remembered_across_a_restart(tmp_path, monkeyp ...@@ -538,7 +538,7 @@ def test_a_surviving_pods_token_is_remembered_across_a_restart(tmp_path, monkeyp
"http://pod-a:8000", "cra-original") "http://pod-a:8000", "cra-original")
monkeypatch.setattr(rw.os, "getpid", lambda: 424242) # a NEW process monkeypatch.setattr(rw.os, "getpid", lambda: 424242) # a NEW process
pod_id, url, key = rw.find_shared_pod("capability:embeddings", pod_id, url, key, rate = rw.find_shared_pod("capability:embeddings",
api_key="cra-freshly-generated") api_key="cra-freshly-generated")
assert (pod_id, url, key) == ("pod-survivor", "http://pod-a:8000", "cra-original") 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" assert seen["key"] == "cra-original", "probed with the pod's token, not ours"
......
...@@ -372,3 +372,59 @@ def test_a_pod_is_recorded_even_when_the_region_is_unset(): ...@@ -372,3 +372,59 @@ def test_a_pod_is_recorded_even_when_the_region_is_unset():
pool.mcfg = rw.parse_model_runpod({}) pool.mcfg = rw.parse_model_runpod({})
pool._volume_dc = None pool._volume_dc = None
assert pool._data_center() == "" # no exception, and a usable value assert pool._data_center() == "" # no exception, and a usable value
# ------------------------------- a restart must not rent beside its own pod
def test_the_warm_path_adopts_before_it_rents():
"""A warm floor is re-established at EVERY boot. Without adoption an hourly
unattended upgrade rents a fresh A40 each time and leaves the previous one
billing until the reaper's 20-minute grace expires — observed three times
on the production orchestrator in one afternoon."""
src = __import__("pathlib").Path(rw.__file__).read_text()
ready = src.split("def ensure_ready")[1].split("def _pick")[0]
assert "_adopt_shared_pod()" in ready
assert ready.index("_adopt_shared_pod()") < ready.index("self._provision_one()")
warm = src.split("def maintain")[1].split("def close")[0]
assert "if not self._adopt_shared_pod():" in warm
def test_adoption_is_one_implementation_for_both_callers():
"""It lived inside acquire() only, which is why the warm path never had it."""
src = __input = __import__("pathlib").Path(rw.__file__).read_text()
assert src.count("def _adopt_shared_pod") == 1
assert src.count("_adopt_shared_pod()") >= 3 # acquire + ensure_ready + maintain
def test_an_adopted_pod_keeps_its_price():
"""It used to be recorded at $0/hr as "billed by its owner". After OUR
restart there is no other owner: reporting it free hides a real A40 from
the $/hr cap and from the spend page."""
src = __import__("pathlib").Path(rw.__file__).read_text()
blk = src.split("def _adopt_shared_pod")[1].split("def _data_center")[0]
assert "hourly_usd=float(shared_rate or 0.0)" in blk
assert "hourly_usd=0.0" not in blk
def test_the_registry_records_the_rate_so_adoption_can_read_it():
import json
import tempfile
d = tempfile.mkdtemp()
saved = rw._pod_registry_path
rw._pod_registry_path = lambda: d + "/pods.json"
try:
rw.register_pod("p1", "pool:x", "http://pod:8000", "tok", hourly_usd=0.59)
rec = json.load(open(d + "/pods.json"))["p1"]
assert rec["hourly_usd"] == 0.59
assert rec["url"] == "http://pod:8000" and rec["key"] == "tok"
# An older entry with no rate must not break adoption.
rec.pop("hourly_usd")
json.dump({"p1": rec}, open(d + "/pods.json", "w"))
saved_health = rw._pod_health_ok
rw._pod_health_ok = lambda *a, **k: True
try:
assert rw.find_shared_pod("pool:x")[3] == 0.0
finally:
rw._pod_health_ok = saved_health
finally:
rw._pod_registry_path = saved
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