one pod, one entry: a pool must not hold the same pod twice (v0.4.29)

Two paths append pods -- a fresh rent and an adoption -- and neither checked
whether the pod was already in the pool. On a restart both fire for the same
machine within the same second:

  [runpod] adopting pod j74fxkxr9vs3fg's bearer token
  [runpod] reclaim: took back a pod for 'datalab-to/surya-ocr-2' ...

and the pool then held one A40 twice. Every capacity question is answered by
walking self.pods, so the duplicate is not cosmetic:

* the pool claimed 2 x max_inflight_per_pod = 128 slots on a card that was
  serving 31, reporting "2 in pool, 2 healthy, 2 at cap" with 255 pages
  queued -- and refusing work the pod could have taken;
* max_pods is enforced against that same count, so a cap of ONE was satisfied
  by two entries for one pod, which is how a second real pod came to be rented
  beside it and billed at $1.18/hr for a configuration that asked for $0.59;
* the in-flight staleness sweep added in 0.4.28 cannot help here, because both
  entries hold legitimate fresh timestamps. A duplicate is not a leak.

Both sites now go through `_add_pod`, which is the only remaining way into the
pool and refuses a pod id it already holds. The EXISTING handle is kept rather
than replaced: it carries the in-flight accounting for work already dispatched
to that pod, and dropping it would strand every slot it holds. What can
legitimately change is refreshed onto it -- the address, since RunPod remaps
ports and the pod id is the identity, not the url.

A test asserts there is exactly one `self.pods.append(` in the file: the guard
is worth nothing if a third path is added later that bypasses it.
Co-Authored-By: 's avatarClaude Opus 5 <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_01KnfkxBpcZ1v6ypRFFyhrVM
parent 9d67ab8b
......@@ -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.28"
__version__ = "0.4.29"
__tagline__ = "Complete orchestration, distribution and escalation of remotizable advanced inference."
# Configure the CUDA caching allocator BEFORE torch is imported anywhere.
......
......@@ -3488,10 +3488,7 @@ class RunpodPodPool:
# knows otherwise.
**_adopted_card(self.account, shared_id),
healthy=True, last_used=time.time())
with self._cv:
self.pods.append(handle)
self._bump_cards()
self._cv.notify_all()
self._add_pod(handle)
register_pod(shared_id, str(self.model_key), shared_url, self.api_key,
hourly_usd=float(shared_rate or 0.0)) # image/gpu kept
return True
......@@ -3823,10 +3820,7 @@ class RunpodPodPool:
if cancellable and self._boot_cancelled(_epoch):
_give_back(pod_id, "just as it came up")
raise BootCancelled(pod_id)
with self._cv:
self.pods.append(h)
self._bump_cards()
self._cv.notify_all()
self._add_pod(h)
with _pools_lock:
_provisioning_ids.discard(pod_id)
_provisioning_info.pop(pod_id, None)
......@@ -4429,6 +4423,53 @@ class RunpodPodPool:
finally:
self._waiting = max(0, self._waiting - 1)
def _add_pod(self, handle: PodHandle) -> bool:
"""Put a pod in the pool, at most ONCE per pod id.
Two paths append pods -- a fresh rent and an adoption -- and neither
used to check whether the pod was already there. On a restart both can
fire for the same pod in the same second (`reclaim: took back a pod`
and `adopting pod <id>'s bearer token` are logged a moment apart), and
the pool then holds two handles for one machine.
That is not a cosmetic duplicate. Every capacity question is answered by
walking `self.pods`, so one A40 counted twice claims 2 x
max_inflight_per_pod slots -- the pool reported `2 in pool, 2 healthy,
2 at cap` with 128 pages in flight while the card itself ran 31 -- and
`max_pods` is enforced against that same inflated count, so a cap of one
can be satisfied by two entries for one pod and a second pod rented
beside them.
Returns True if the pod was added, False if it was already here. The
EXISTING handle is kept rather than replaced: it carries the in-flight
accounting for work already dispatched to that pod, and throwing it away
would strand every slot it holds. What can legitimately change -- the
address, after RunPod remaps a port -- is refreshed onto it.
"""
with self._cv:
for p in self.pods:
if p.pod_id != handle.pod_id:
continue
if handle.url and handle.url != p.url:
print(f"[runpod] pod {p.pod_id} is already in the pool for "
f"{self.model_key!r}; keeping the handle that holds "
f"its {p.inflight} in-flight slot(s) and updating its "
f"address to {handle.url}", flush=True)
p.url = handle.url
else:
print(f"[runpod] pod {p.pod_id} is already in the pool for "
f"{self.model_key!r} — not adding it twice (a second "
f"entry would claim a second pod's worth of capacity "
f"for one card)", flush=True)
p.healthy = True
p.last_used = time.time()
self._cv.notify_all()
return False
self.pods.append(handle)
self._bump_cards()
self._cv.notify_all()
return True
def release(self, pod: PodHandle):
with self._cv:
pod.inflight = max(0, pod.inflight - 1)
......
"""A pool must never hold two handles for one pod.
Two paths append pods -- a fresh rent and an adoption -- and neither checked
whether the pod was already there. On a restart both fire for the same pod in
the same second; the log showed
[runpod] adopting pod j74fxkxr9vs3fg's bearer token
[runpod] reclaim: took back a pod for 'datalab-to/surya-ocr-2' ...
and the pool then held one A40 twice. Every capacity question is answered by
walking self.pods, so the consequences compound:
* the pool claimed 2 x max_inflight_per_pod = 128 slots on a card serving 31,
reporting "2 in pool, 2 healthy, 2 at cap" with 255 pages queued
* max_pods is enforced against that same count, so a cap of ONE can be
satisfied by two entries for one pod -- and a second real pod rented beside
them, which is what was billed at $1.18/hr for a $0.59 configuration
* the in-flight staleness sweep cannot help: both entries hold legitimate
fresh timestamps
"""
import time
import threading
import pytest
import codai.api.runpod_worker as rw
def _handle(pod_id="p1", url="https://a", inflight=0):
h = rw.PodHandle(pod_id=pod_id, url=url, hourly_usd=0.59,
started_at=time.time(), gpu="A40", healthy=True)
h.inflight = inflight
return h
class _Pool:
def __init__(self):
self.pods = []
self._cv = threading.Condition()
self.model_key = "m"
self.cards = {}
def _bump_cards(self):
pass
_add_pod = rw.RunpodPodPool._add_pod
# --- the regression ---
def test_the_same_pod_is_not_added_twice():
pool = _Pool()
assert pool._add_pod(_handle()) is True
assert pool._add_pod(_handle()) is False
assert len(pool.pods) == 1
def test_rent_then_adopt_of_one_pod_leaves_one_entry():
"""The restart sequence: both paths fire for the same machine."""
pool = _Pool()
pool._add_pod(_handle(url="https://first"))
pool._add_pod(_handle(url="https://first"))
assert [p.pod_id for p in pool.pods] == ["p1"]
def test_capacity_is_not_doubled_by_a_duplicate():
"""The observed consequence: one card claiming two pods' worth of slots."""
pool = _Pool()
pool._add_pod(_handle(inflight=64))
pool._add_pod(_handle(inflight=64))
assert sum(p.inflight for p in pool.pods) == 64, \
"a duplicate made one pod's 64 in-flight pages look like 128"
# --- the existing handle must survive, because it holds the accounting ---
def test_the_live_handle_is_kept_not_replaced():
"""Replacing it would strand every in-flight slot it holds."""
pool = _Pool()
first = _handle(inflight=17)
pool._add_pod(first)
pool._add_pod(_handle(inflight=0))
assert pool.pods[0] is first
assert pool.pods[0].inflight == 17
def test_a_changed_address_is_refreshed_onto_the_live_handle():
"""RunPod remaps ports; the pod id is the identity, the url is not."""
pool = _Pool()
first = _handle(url="https://old:1", inflight=5)
pool._add_pod(first)
pool._add_pod(_handle(url="https://new:2"))
assert len(pool.pods) == 1
assert pool.pods[0] is first
assert pool.pods[0].url == "https://new:2"
assert pool.pods[0].inflight == 5, "accounting must survive a re-address"
def test_a_returning_pod_is_marked_healthy_again():
pool = _Pool()
first = _handle()
pool._add_pod(first)
first.healthy = False
pool._add_pod(_handle())
assert first.healthy is True
# --- different pods are still different pods ---
def test_distinct_pods_are_both_added():
pool = _Pool()
assert pool._add_pod(_handle(pod_id="a")) is True
assert pool._add_pod(_handle(pod_id="b")) is True
assert len(pool.pods) == 2
def test_many_distinct_pods_all_land():
pool = _Pool()
for i in range(8):
assert pool._add_pod(_handle(pod_id=f"p{i}")) is True
assert len({p.pod_id for p in pool.pods}) == 8
def test_concurrent_adds_of_one_pod_still_yield_one_entry():
"""Rent and adopt genuinely race; the guard is under the pool lock."""
pool = _Pool()
added = []
def add():
added.append(pool._add_pod(_handle()))
ts = [threading.Thread(target=add) for _ in range(12)]
for t in ts:
t.start()
for t in ts:
t.join()
assert len(pool.pods) == 1
assert added.count(True) == 1, "exactly one caller should have added it"
# --- there must be only one way in ---
def test_the_worker_has_a_single_append_site():
"""Two unguarded append sites are how this happened; the guard is only
worth anything if nothing bypasses it."""
src = open("codai/api/runpod_worker.py").read()
assert src.count("self.pods.append(") == 1, \
"every path into the pool must go through _add_pod"
assert "def _add_pod" in src
assert "self._add_pod(handle)" in src, "the adoption path"
assert "self._add_pod(h)" in src, "the fresh-rent path"
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