runpod: make a seeded model actually usable, and stop the double rent

The pod finally answered, and answered "Model 'bge-m3' is not available.
Use one of: " — with NOTHING after the colon, despite being seeded with
five models and successfully taught a sixth.

Two causes, both only visible on a real pod.

REGISTERING A MODEL WAS HALF DONE. Both the boot seed and
/v1/models/register wrote models_data, but request validation asks the
MODEL MANAGER for its allowed identifiers. A model in the catalogue that
the manager never heard of is refused. Registration now calls the right
setter per section, maps the alias a request actually uses (bge-m3 ->
BAAI/bge-m3), and widens the engine's assigned-model set — a model
registered at runtime was assigned by nobody and would otherwise be
usable but invisible.

AND THE SEED WAS FILTERED OUT BEFORE IT GOT THERE. A pod runs front plus
engine; the front computes each engine's assignment from models.json, and
the seed is deliberately in-memory only, so the engine was told to
register nothing — dropping exactly the models the pod was rented for.
Seeded entries now survive that filter.

Separately: both engines rented their own pod again, 45 seconds apart,
because a pod takes ~2.5 minutes to boot and the sharing check only saw
READY pods. A pod is now registered at creation, and a sibling that finds
one still booting WAITS for it instead of renting a second card.
Co-Authored-By: 's avatarClaude Opus 5 (1M context) <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_0182DhR1eFNPQyGmDrpHbedY
parent c4c316e5
......@@ -142,25 +142,81 @@ async def register_models(request: Request, _auth=Depends(_require_api_auth)):
return {"added": added, "already_known": known, "refused": refused}
def _register_entry(entry: dict, section: str) -> str:
"""Add one entry to the in-memory catalogue. Returns 'added' or 'known'."""
#: models.json section -> the manager method that makes a model USABLE. Writing
#: models_data alone is not enough: request validation asks the manager for its
#: allowed identifiers, so a model registered only in the catalogue file is still
#: refused with "not available" — observed on a live pod, with an empty list
#: after the colon.
_SECTION_SETTER = {
"text_models": "set_default_model",
"gguf_models": "set_default_model",
"vision_models": "set_vision_model",
"image_models": "set_image_model",
"audio_models": "set_audio_model",
"tts_models": "set_tts_model",
"video_models": "set_video_model",
"audio_gen_models": "set_audio_gen_model",
"embedding_models": "set_embedding_model",
"spatial_models": "set_spatial_model",
}
def register_model_runtime(entry: dict, section: str) -> str:
"""Make a model known AND usable on this instance. 'added' or 'known'.
Two steps, both required: the entry goes into models_data (so the catalogue
and the admin pages see it) and the model manager is told about it (so
request validation accepts it).
"""
path = entry.get("path")
already = False
try:
from codai.admin.routes import config_manager
md = getattr(config_manager, "models_data", None)
if not isinstance(md, dict):
return "known"
if isinstance(md, dict):
lst = md.setdefault(section, [])
if not isinstance(lst, list):
return "known"
path = entry.get("path")
if any(isinstance(m, dict) and m.get("path") == path for m in lst):
return "known"
if isinstance(lst, list):
already = any(isinstance(m, dict) and m.get("path") == path
for m in lst)
if not already:
lst.append(dict(entry))
print(f"[register] {path} added to {section}", flush=True)
return "added"
except Exception as exc:
print(f"[register] failed: {exc}", flush=True)
return "known"
print(f"[register] catalogue update failed: {exc}", flush=True)
setter = _SECTION_SETTER.get(section)
if setter:
try:
from codai.models.manager import multi_model_manager
cfg = {k: v for k, v in entry.items() if k not in ("path",)}
cfg.setdefault("load_mode", "on-request")
getattr(multi_model_manager, setter)(path, config=cfg)
if entry.get("alias"):
# An alias is how a request usually names it (bge-m3 vs BAAI/bge-m3).
try:
multi_model_manager.model_aliases[str(entry["alias"])] = path
except Exception:
pass
# An engine only lists models the front assigned it, and a model
# registered at runtime was assigned by nobody. Widen the set, or it
# would be usable but invisible in /v1/models.
try:
cur = getattr(multi_model_manager, "_assigned_model_keys", None)
if cur is not None:
multi_model_manager.set_assigned_models(
set(cur) | {path, entry.get("alias") or path})
except Exception:
pass
except Exception as exc:
print(f"[register] {path} could not be made usable: {exc}", flush=True)
return "known" if already else "added"
if not already:
print(f"[register] {path} registered in {section}", flush=True)
return "known" if already else "added"
def _register_entry(entry: dict, section: str) -> str:
"""Back-compat wrapper."""
return register_model_runtime(entry, section)
@router.post("/v1/models/upload", summary="Upload model weights")
......
......@@ -1510,6 +1510,16 @@ class RunpodPodPool:
# Before renting: is a sibling process already running a pod for
# this exact pool? One engine per GPU means the same capability
# was being rented a card each.
if sibling_is_provisioning(str(self.model_key)):
# Another engine is already booting one for this pool. Wait
# for it rather than renting a second card for the same work.
print(f"[runpod] another engine is booting a pod for "
f"{self.model_key!r} — waiting for it", flush=True)
with self._cv:
self._provisioning = False
self._cv.notify_all()
self._cv.wait(15.0)
continue
shared_id, shared_url = find_shared_pod(
str(self.model_key), self.health_path, self.api_key)
if shared_url:
......@@ -1920,6 +1930,30 @@ def register_pod(pod_id: str, pool_key: str = "", url: str = "") -> None:
or True)
def sibling_is_provisioning(pool_key: str) -> bool:
"""True when another process has a pod for this pool that is still booting.
Registered at creation with no URL yet. Without this the sibling rents its
own during the ~2.5 minutes a pod takes to boot — which is exactly what
happened live: two pods for one capability, 45 seconds apart.
"""
data = _registry_update(lambda d: False)
if not isinstance(data, dict):
return False
mine = os.getpid()
now = time.time()
for info in data.values():
if not isinstance(info, dict) or info.get("pool") != str(pool_key):
continue
if info.get("pid") == mine or info.get("url"):
continue
# Only while it could plausibly still be booting; a stale entry from a
# dead process must not block provisioning forever.
if now - float(info.get("at") or 0) < REAP_GRACE_SECONDS:
return True
return False
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, '').
......
......@@ -1346,6 +1346,12 @@ class ConfigManager:
continue # a real entry already covers it — leave it alone
lst.append(dict(e))
print(f"[seed] registered {path} in {section}", flush=True)
# The catalogue alone is not enough: request validation asks the MODEL
# MANAGER for its allowed identifiers, and a seeded pod whose manager was
# never told still answers "not available" with an empty list. The
# manager is set up after this runs, so the work is deferred to it.
self._pending_seed_entries = [e for e in entries if isinstance(e, dict)
and e.get("path")]
def save_models(self):
"""Save models.json to disk."""
......
......@@ -870,6 +870,31 @@ def main():
except Exception as _e:
print(f"[engine] assignment filter failed ({_e}); registering all models")
# Models seeded by CODERAI_SEED_MODELS live in memory only — a pod is
# disposable, so they are deliberately not written to models.json. The front
# computes its engine assignment FROM that file, so it assigns none of them
# and the filter above would drop exactly the models this pod was rented to
# serve. Observed on a live pod: "not available. Use one of: " with nothing
# after the colon. Put them back.
_seeded = list(getattr(config_mgr, "_pending_seed_entries", None) or [])
if _seeded:
_keys = set()
for _e in _seeded:
_sec = _e.get("model_type") or "text_models"
_lst = models_config.setdefault(_sec, [])
if isinstance(_lst, list) and not any(
isinstance(x, dict) and x.get("path") == _e.get("path")
for x in _lst):
_lst.append(_e)
_keys.add(_e.get("alias") or _e.get("path"))
try:
_cur = getattr(multi_model_manager, "_assigned_model_keys", None)
if _cur is not None:
multi_model_manager.set_assigned_models(set(_cur) | _keys)
except Exception:
pass
print(f"[seed] {len(_seeded)} seeded model(s) registered on this engine")
# Helper to find model config
def get_model_cfg(model_type, model_id):
key = f"{model_type}:{model_id}"
......
......@@ -485,3 +485,33 @@ def test_the_venv_marker_is_written_last_and_builders_are_serialised():
assert s.index('touch "$MARK"') < s.index('[ -f "$MARK" ] ||')
# mkdir is the atomic lock; a loser waits instead of installing too.
assert 'mkdir "$LOCK"' in s and "another pod is building" in s
def test_a_sibling_waits_for_a_booting_pod_instead_of_renting_one(tmp_path, monkeypatch):
"""A pod takes ~2.5 minutes to boot. Live, the second engine looked during
that window, saw no READY pod, and rented its own — two cards, 45s apart."""
import codai.api.runpod_worker as rw
monkeypatch.setattr(rw, "_pod_registry_path", lambda: str(tmp_path / "pods.json"))
# A sibling registered a pod that has not finished booting (no url yet).
rw._registry_update(lambda d: d.__setitem__(
"booting", {"pool": "capability:embeddings", "pid": 999999,
"at": __import__("time").time(), "url": ""}) or True)
assert rw.sibling_is_provisioning("capability:embeddings") is True
assert rw.sibling_is_provisioning("capability:images") is False
# Once it is serving, it is adoptable rather than merely blocking.
monkeypatch.setattr(rw, "_pod_health_ok", lambda url, **kw: True)
rw._registry_update(lambda d: d.__setitem__(
"booting", {"pool": "capability:embeddings", "pid": 999999,
"at": __import__("time").time(),
"url": "http://pod:8000"}) or True)
assert rw.sibling_is_provisioning("capability:embeddings") is False
assert rw.find_shared_pod("capability:embeddings")[1] == "http://pod:8000"
# A stale entry from a dead process must not block provisioning forever.
rw._registry_update(lambda d: d.__setitem__(
"ancient", {"pool": "capability:video", "pid": 999998,
"at": 0, "url": ""}) or True)
assert rw.sibling_is_provisioning("capability:video") is False
......@@ -1369,3 +1369,45 @@ def test_a_model_the_pod_could_never_fetch_is_refused_with_a_reason(learning_pod
# that would fail later at load time.
assert gw.teach_model(url, "", "local-only") is False
assert _LearningPod.known == set()
def test_registering_a_model_makes_it_usable_not_just_listed(monkeypatch):
"""Writing models_data alone is not enough: request validation asks the
MODEL MANAGER for its allowed identifiers. A live pod proved it — seeded
with 5 models and still answering "not available. Use one of: " with an
empty list."""
import codai.admin.routes as ar
from codai.api import models_transfer as mt
registered = {}
class _MM:
model_aliases = {}
_assigned_model_keys = set()
def set_embedding_model(self, path, config=None):
registered["path"] = path
registered["config"] = config
def set_assigned_models(self, keys):
registered["assigned"] = set(keys)
class _CM:
models_data = {}
monkeypatch.setattr(ar, "config_manager", _CM())
import codai.models.manager as mgr
monkeypatch.setattr(mgr, "multi_model_manager", _MM())
state = mt.register_model_runtime(
{"path": "BAAI/bge-m3", "model_type": "embedding_models", "alias": "bge-m3"},
"embedding_models")
assert state == "added"
# In the catalogue…
assert _CM.models_data["embedding_models"][0]["path"] == "BAAI/bge-m3"
# …AND known to the manager, which is what validation consults.
assert registered["path"] == "BAAI/bge-m3"
# …AND reachable by the alias a request will actually use.
assert _MM.model_aliases["bge-m3"] == "BAAI/bge-m3"
# …AND not filtered out by an engine assignment it was never part of.
assert {"BAAI/bge-m3", "bge-m3"} <= registered["assigned"]
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