one refusing pod no longer sets the whole OCR pool aside (v0.4.49)

A single pod whose vLLM failed to start (another vLLM still held 37 of its
47 GB) answered 503 to every page. _read_on_pods set the whole pool's pod
pipeline aside for 300 s on that one answer, so every page in that process was
read on the orchestrator. Live, that meant ~10 pages/s from 34 pods, with the
orchestrator's CPU doing the reading.

Now a refusal takes that pod out of rotation (lease.dead -> quarantine) and
the page tries another pod. The pool is set aside only when a page's attempts
were refused by at least two different pods, i.e. when it looks like the
image and not one card.
Co-Authored-By: 's avatarClaude Opus 5.5 <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_01KnfkxBpcZ1v6ypRFFyhrVM
parent 2835beec
...@@ -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.4.48" __version__ = "0.4.49"
__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.
......
...@@ -2413,6 +2413,7 @@ class OcrManager: ...@@ -2413,6 +2413,7 @@ class OcrManager:
return None return None
opts = dict(page_opts or {}) opts = dict(page_opts or {})
refused: list = [] refused: list = []
refusers: set = set()
unreachable: list = [] unreachable: list = []
policy, wait_s = _pod_pipeline_policy(cfg) policy, wait_s = _pod_pipeline_policy(cfg)
# Decided once per document, not per page: a document whose pages # Decided once per document, not per page: a document whose pages
...@@ -2502,8 +2503,29 @@ class OcrManager: ...@@ -2502,8 +2503,29 @@ class OcrManager:
# while the local pipeline sat there able to read it. # while the local pipeline sat there able to read it.
if int(getattr(exc, "status", 0) or 0) not in _POD_CANNOT_RUN_PIPELINE: if int(getattr(exc, "status", 0) or 0) not in _POD_CANNOT_RUN_PIPELINE:
raise raise
refused.append(str(exc)) # BUT ONE POD IS NOT THE POOL. A single pod whose vLLM
return None # failed to start (measured: another vLLM still held
# 37 of its 47 GB) answers 503 to every page, and
# setting the whole pool aside on that one answer sent
# every page of the process to this box's pipeline for
# five minutes at a time -- ~10 pages/s from 34 pods,
# with the orchestrator's CPU doing the reading. So the
# refusing pod is taken out of rotation and the page
# tries ANOTHER one; only a page refused by every pod
# it tried says something about the image.
lease.dead()
refusers.add(lease.base)
_say_rate_limited(
("pod-refused", model_id, lease.base),
f"[ocr] a pod for {model_id!r} at {lease.base} "
f"refused to run the pipeline ({exc}); it is out "
f"of rotation and this page tries another pod "
f"(attempt {attempt} of {POD_PAGE_ATTEMPTS})",
every=60.0)
if attempt >= POD_PAGE_ATTEMPTS:
refused.append(str(exc))
return None
continue
finally: finally:
lease.release() lease.release()
return None return None
...@@ -2536,7 +2558,16 @@ class OcrManager: ...@@ -2536,7 +2558,16 @@ class OcrManager:
# different fault from "nothing is rented", needs a different # different fault from "nothing is rented", needs a different
# fix, and must carry the pod's OWN words: this is the only # fix, and must carry the pod's OWN words: this is the only
# place they are visible. # place they are visible.
if _set_aside_pod_pipeline(model_id, refused[0]): if len(refusers) < 2:
# Every attempt landed on the same pod (a one-pod pool, or
# placement kept choosing it). Still one pod's word: the
# pool is not set aside, and the next document asks again.
_say_rate_limited(
("one-refuser", model_id),
f"[ocr] only one pod refused ({refused[0]}) — reading "
f"this document here, NOT setting the pool aside",
every=60.0)
elif _set_aside_pod_pipeline(model_id, refused[0]):
print(f"[ocr] the pods for {model_id!r} run an image that " print(f"[ocr] the pods for {model_id!r} run an image that "
f"carries the OCR pipeline, but they REFUSE to run " f"carries the OCR pipeline, but they REFUSE to run "
f"it: {refused[0]} — falling back to the pipeline on " f"it: {refused[0]} — falling back to the pipeline on "
......
...@@ -204,17 +204,29 @@ def test_local_serve_mode_never_goes_to_a_pod(monkeypatch): ...@@ -204,17 +204,29 @@ def test_local_serve_mode_never_goes_to_a_pod(monkeypatch):
# -------------------------------------------------------------------------- # --------------------------------------------------------------------------
class _Lease: class _Lease:
"""A lease on a DIFFERENT pod each time, like a pool with many pods."""
kind = "pod" kind = "pod"
base = "https://10.0.0.1:1234/v1"
api_key = "k" api_key = "k"
n = 0
dead_bases: list = []
def __init__(self, base=None):
if base is None:
_Lease.n += 1
base = f"https://10.0.0.{_Lease.n}:1234/v1"
self.base = base
def release(self): def release(self):
pass pass
def dead(self):
_Lease.dead_bases.append(self.base)
@pytest.fixture(autouse=True) @pytest.fixture(autouse=True)
def _forget_set_aside_pools(): def _forget_set_aside_pools():
om._POD_PIPELINE_UNUSABLE.clear() om._POD_PIPELINE_UNUSABLE.clear()
_Lease.dead_bases = []
yield yield
om._POD_PIPELINE_UNUSABLE.clear() om._POD_PIPELINE_UNUSABLE.clear()
...@@ -326,6 +338,52 @@ async def test_the_pool_is_tried_again_once_the_cooldown_passes(monkeypatch): ...@@ -326,6 +338,52 @@ async def test_the_pool_is_tried_again_once_the_cooldown_passes(monkeypatch):
assert len(calls) > before assert len(calls) > before
@pytest.mark.anyio
async def test_one_refusing_pod_does_not_set_the_pool_aside(monkeypatch):
"""THE 2026-10-19 slowdown: one pod whose vLLM never started answered 503,
the whole pool was set aside, and every page of that process was read on
the orchestrator -- ~10 pages/s from 34 pods. The page must go to another
pod, the bad one must leave rotation, and the pool must stay in use."""
_two_page_pdf_split(monkeypatch, pages=4)
from codai.models import placement
bad = "https://10.9.9.9:1/v1"
picks = iter([bad] + [None] * 100)
monkeypatch.setattr(placement, "choose",
lambda *_a, **_k: _Lease(next(picks)))
monkeypatch.setattr(placement, "spill_threshold", lambda *_a, **_k: 0)
def _answer(base, *_a, **_k):
if base == bad:
raise om.OcrError("pod OCR answered 503: local vLLM unusable",
status=503)
return {"engine": "surya", "pages": [{"text": "ok"}]}
_stub_page_post(monkeypatch, _answer)
mgr = om.OcrManager()
got = await _read(mgr, types.SimpleNamespace())
assert got is not None, "the document fell back to this box"
assert bad in _Lease.dead_bases
assert not om._pod_pipeline_refusal("datalab-to/surya-ocr-2")
@pytest.mark.anyio
async def test_a_single_pod_refusing_every_attempt_is_still_not_the_pool(
monkeypatch):
_two_page_pdf_split(monkeypatch, pages=1)
from codai.models import placement
monkeypatch.setattr(placement, "choose",
lambda *_a, **_k: _Lease("https://10.9.9.9:1/v1"))
monkeypatch.setattr(placement, "spill_threshold", lambda *_a, **_k: 0)
def _refuse(*_a, **_k):
raise om.OcrError("pod OCR answered 503: nope", status=503)
_stub_page_post(monkeypatch, _refuse)
mgr = om.OcrManager()
assert await _read(mgr, types.SimpleNamespace()) is None
assert not om._pod_pipeline_refusal("datalab-to/surya-ocr-2")
@pytest.mark.anyio @pytest.mark.anyio
async def test_a_real_page_failure_still_fails(monkeypatch): async def test_a_real_page_failure_still_fails(monkeypatch):
"""Falling back on ANY error would hide a broken page behind a slow one.""" """Falling back on ANY error would hide a broken page behind a slow one."""
......
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