the pod path was 32 threads wide, and said 512 (v0.4.8)

A 512-page semaphore, a 48-wide admission gate on every pod, 31 A40s at
`Running: 1` with `Waiting: 0`, and the orchestrator delivered 1.65 pages/s.
Raising `ocr.max_concurrency` 128 -> 512 changed nothing. Nor did replacing the
pods' 2-wide gate with 48. The reason both were inert is that neither was ever
the limit:

    32 threads / 19.4 s per page = 1.65 pages/s

Two defects, and the second is why the first was invisible for every release
this code has shipped in:

* **The per-page forward ran on the DEFAULT executor.** `asyncio.to_thread`
  hands work to the loop's default pool -- 32 threads, shared with every other
  blocking call in the process -- while `codai/api/thread_pool.py` exists for
  exactly this call, documents it ("a per-page OCR forward to a pod (30-300
  seconds each)"), and offers `run_on_pod_pool()`. Nothing called it. It was
  dead code next to the problem it was written for.

* **Every sizing helper read a name that does not exist.**
  `thread_pool._configured_concurrency()` and `pod_http._pool_size()` both did
  `from codai.config import config_manager`; the shared manager lives in
  `codai.admin.routes`. The ImportError was swallowed by a bare except, so both
  answered 0 and took their FLOORS -- 32 threads, 64 connections -- on every
  install that has ever run, whatever the config said. `ocr.max_concurrency`
  has never once reached a thread pool.

So the two knobs an operator would reach for were the two that could not
help, and the number in the boot line ("default pool 32 workers") was the
floor wearing the clothes of a configured value.

Fixed:

* the page forward and the liveness probe go to the pod pool;
* sizing reads the holder that exists, tries both, and `sized_from_config()`
  reports which -- the boot line now says FLOORS out loud when it could not
  read the config, because "32 workers" means something entirely different
  when it is chosen than when it is a fallback nobody saw;
* `_MAX_POD` 512 -> 2048: ~100 pages/s at ~20 s per page is ~2000 forwards
  outstanding, and each is a thread blocked on a socket. Past that an async
  client is the answer, not more stacks;
* `configure()` grows the pod pool when `max_concurrency` changes, so raising
  it mid-run widens the threads and not just the semaphore. Growing only:
  shrinking would have to stranded threads mid-page.

With max_concurrency 2048 the pools are now (256, 2048) rather than (32, 64).

Credit where it is due: the client team measured 41-42 established sockets
inside the container's network namespace and a 1.65 pages/s ceiling, and said
the serialisation was on the orchestrator's OCR path. It was. The contrary
reading here came from running `ss` on the HOST, outside the rootless netns,
which can only see forwarded sockets -- a measurement of the wrong thing,
reported with more confidence than it had earned.

11 new tests, including the one-liner that pins the original bug
(`assert not hasattr(codai.config, "config_manager")`) and a static check that
the forward is not back on `asyncio.to_thread`. 3324 passed, 13 pre-existing
failures (broker, studio, manual client) untouched.
Co-Authored-By: Claude Opus 5 <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_01KnfkxBpcZ1v6ypRFFyhrVM

codai/__init__.py

@@ -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.7"
+__version__ = "0.4.8"
 __tagline__ = "Complete orchestration, distribution and escalation of remotizable advanced inference."
 
 # Configure the CUDA caching allocator BEFORE torch is imported anywhere.

codai/api/pod_http.py

@@ -64,9 +64,12 @@ def _pool_size() -> int:
     should not have to know that a connection pool exists.
     """
     try:
-        from codai.config import config_manager
-        cfg = config_manager.config.ocr if config_manager and config_manager.config else None
-        want = int(getattr(cfg, "max_concurrency", 0) or 0)
+        # The SHARED manager is in codai.admin.routes; codai.config has no
+        # `config_manager` at all, so the old read raised ImportError on every
+        # call and this silently returned the floor. See
+        # codai/api/thread_pool.py for what that cost.
+        from codai.api.thread_pool import _configured_concurrency
+        want = _configured_concurrency()
     except Exception:
         want = 0
     return max(_DEFAULT_POOL, min(_MAX_POOL, want))

codai/api/thread_pool.py

@@ -63,7 +63,11 @@ _MIN_POD = 64
 #: Ceilings. A blocked thread is cheap but not free, and a runaway config
 #: should not be able to ask for ten thousand stacks.
 _MAX_DEFAULT = 256
-_MAX_POD = 512
+#: Raised from 512 once the sizing above started working: a run that wants
+#: 100 pages/s at ~20 s per page needs ~2000 forwards outstanding, and every
+#: one of them is a thread blocked on a socket. Beyond this an async client is
+#: the right answer, not more stacks.
+_MAX_POD = 2048
 
 _lock = threading.Lock()
 _pod_pool: Optional[ThreadPoolExecutor] = None
@@ -76,18 +80,47 @@ def _configured_concurrency() -> int:
     Read from the config rather than from the CPU count: these threads are
     waiting on other machines, so the number that matters is how many remote
     calls may be outstanding, not how many cores can run Python.
+
+    THE SHARED MANAGER LIVES IN ``codai.admin.routes``, not in ``codai.config``.
+    This read used to be ``from codai.config import config_manager``, which has
+    no such name -- so it raised ImportError on every call, the except swallowed
+    it, and this returned 0 forever. Both pools then took their FLOORS (32 and
+    64) on every install that has ever run, whatever ``ocr.max_concurrency``
+    said. Measured on a live orchestrator: 32 page forwards in flight against a
+    512-page semaphore and 31 idle pods, a ceiling of 1.65 pages/s
+    (32 / 19.4 s per page) that moved for neither a 4x semaphore nor a 24x pod
+    gate, because neither was ever the limit.
+
+    Hence BOTH holders, and no bare except: a sizing function that silently
+    answers "unconfigured" is indistinguishable from one that answers
+    truthfully, and the difference was a tenfold throughput ceiling nobody
+    could see. ``sized_from_config()`` reports which it was.
     """
     want = 0
-    try:
-        from codai.config import config_manager
-        cfg = config_manager.config if config_manager else None
-        if cfg is not None:
-            want = max(want, int(getattr(cfg.ocr, "max_concurrency", 0) or 0))
-    except Exception:
-        pass
+    for mod, attr in (("codai.admin.routes", "config_manager"),
+                      ("codai.config", "config_manager")):
+        try:
+            import importlib
+            mgr = getattr(importlib.import_module(mod), attr, None)
+            cfg = getattr(mgr, "config", None) if mgr is not None else None
+            ocr = getattr(cfg, "ocr", None) if cfg is not None else None
+            if ocr is not None:
+                want = max(want, int(getattr(ocr, "max_concurrency", 0) or 0))
+        except Exception:
+            continue
     return want
 
 
+def sized_from_config() -> bool:
+    """Whether the pools could read the config at all.
+
+    For the boot line: "32 workers" means something entirely different when it
+    is the configured number than when it is the floor taken because nothing
+    could be read.
+    """
+    return _configured_concurrency() > 0
+
+
 def pool_sizes() -> tuple:
     """(default_workers, pod_workers) for this install."""
     want = _configured_concurrency()
@@ -117,8 +150,11 @@ def install_default_executor(loop=None) -> Optional[ThreadPoolExecutor]:
                                 thread_name_prefix="coderai")
         loop.set_default_executor(ex)
         _installed = True
+        how = ("from ocr.max_concurrency" if sized_from_config()
+               else "FLOORS — the config was not readable from here yet, so "
+                    "ocr.max_concurrency did NOT size these")
         print(f"[threads] default pool {general} workers, pod pool {_pod} "
-              f"(python's default here would be "
+              f"({how}; python's default here would be "
               f"{min(32, (os.cpu_count() or 1) + 4)})", flush=True)
         return ex
 
@@ -138,6 +174,28 @@ def pod_executor() -> ThreadPoolExecutor:
         return _pod_pool
 
 
+def resize_pod_pool() -> int:
+    """Grow the pod pool to the configured width. Returns its new size.
+
+    Called when the config changes: the pool is built on first use, so an
+    install that raises ``ocr.max_concurrency`` mid-run would otherwise keep
+    whatever width it was born with until a restart. Growing only -- shrinking
+    would have to decide what to do with the threads currently blocked on a
+    page, and the answer that does not lose pages is "nothing".
+    """
+    global _pod_pool
+    want = pool_sizes()[1]
+    with _lock:
+        if _pod_pool is None:
+            return want                      # built at the right size on demand
+        cur = getattr(_pod_pool, "_max_workers", 0)
+        if want > cur:
+            _pod_pool._max_workers = want
+            print(f"[threads] pod pool {cur} -> {want} workers "
+                  f"(ocr.max_concurrency changed)", flush=True)
+        return max(cur, want)
+
+
 async def run_on_pod_pool(func, /, *args, **kwargs):
     """``asyncio.to_thread``, but on the pod pool.
 

codai/ocr/manager.py

@@ -33,6 +33,7 @@ import threading
 import time
 from typing import Dict, List, Optional
 
+from codai.api.thread_pool import run_on_pod_pool
 from codai.ocr.base import (OcrEngine, OcrPage, OcrError, ink_coverage,
                             PAGE_OPTIONS, page_estimate, render_pages,
                             iter_pdf_page_pdfs, pdf_page_count, split_pdf_page,
@@ -1160,6 +1161,15 @@ class OcrManager:
                     pass
             self._pools = {}
             self._sem = asyncio.Semaphore(max(1, int(getattr(cfg, "max_concurrency", 4))))
+            # The semaphore is only half of the width: every page in flight on
+            # the pod path also holds a thread, so a raised max_concurrency that
+            # does not reach the pod pool just queues on it instead. Grows only,
+            # and only when the pool already exists.
+            try:
+                from codai.api.thread_pool import resize_pod_pool
+                resize_pod_pool()
+            except Exception:
+                pass
         self._register_releaser()
 
     def _register_releaser(self) -> None:
@@ -2034,7 +2044,16 @@ class OcrManager:
                     if lease is None:
                         return None
                     try:
-                        return await asyncio.to_thread(
+                        # THE POD POOL, not the default executor. A page forward
+                        # blocks for 8-35 s, and `asyncio.to_thread` hands it to
+                        # the loop's default pool -- 32 threads, shared with
+                        # every other blocking call in the process. That was the
+                        # real ceiling on this path: 32 forwards in flight
+                        # against a 512-page semaphore and 31 idle pods, which
+                        # is 1.65 pages/s however wide the semaphore or the
+                        # pods' own gate is set. codai/api/thread_pool.py was
+                        # written for exactly this call and nothing used it.
+                        return await run_on_pod_pool(
                             _post_page_to_pod, lease.base, lease.api_key, blob,
                             f"{idx}.pdf" if is_pdf else (filename or "page"),
                             dpi, opts, detect, engine_name)
@@ -2045,7 +2064,7 @@ class OcrManager:
                         # it on a reused idle connection has cost seven healthy
                         # cards before), and one that does not is reported so
                         # the next attempt is routed elsewhere.
-                        alive = await asyncio.to_thread(
+                        alive = await run_on_pod_pool(
                             _pod_still_alive, lease.base, lease.api_key)
                         if not alive:
                             lease.dead()

tests/test_ocr_pod_pool_width.py

+"""The width of the pod path, which was 32 threads and nobody could see it.
+
+A 512-page semaphore, a 48-wide gate on every pod and 31 idle A40s, and the
+orchestrator still delivered 1.65 pages/s. The cause was two lines:
+
+* every sizing helper read ``codai.config.config_manager``, a name that does
+  not exist (the shared manager is in ``codai.admin.routes``), so the
+  ImportError was swallowed and both pools took their FLOORS forever;
+* the per-page forward used ``asyncio.to_thread``, i.e. the loop's default
+  32-thread pool, rather than the pod pool written for it.
+
+32 threads / 19.4 s per page = 1.65 pages/s. Neither the semaphore nor the pod
+gate could ever have moved it. These tests fail if either half comes back.
+"""
+
+import asyncio
+
+import pytest
+
+from codai.api import pod_http, thread_pool
+
+
+class _Ocr:
+    max_concurrency = 0
+
+
+class _Cfg:
+    def __init__(self, n):
+        self.ocr = _Ocr()
+        self.ocr.max_concurrency = n
+
+
+class _Mgr:
+    def __init__(self, n):
+        self.config = _Cfg(n)
+
+
+@pytest.fixture
+def holder(monkeypatch):
+    """Install a config manager where the REAL one lives."""
+    import codai.admin.routes as routes
+
+    def _set(n):
+        monkeypatch.setattr(routes, "config_manager", _Mgr(n), raising=False)
+    return _set
+
+
+# ------------------------------------------------------- reading the config #
+
+def test_the_pools_read_the_holder_that_actually_exists(holder):
+    holder(512)
+    assert thread_pool._configured_concurrency() == 512
+    assert thread_pool.sized_from_config() is True
+
+
+def test_codai_config_has_no_config_manager():
+    """The regression in one line: the old read could never have worked, on any
+    install, in any version. If this name ever appears, the fallback is live
+    again and the sizing should prefer it."""
+    import codai.config
+    assert not hasattr(codai.config, "config_manager")
+
+
+def test_an_unreadable_config_is_reported_not_disguised(monkeypatch):
+    import codai.admin.routes as routes
+    monkeypatch.setattr(routes, "config_manager", None, raising=False)
+    assert thread_pool._configured_concurrency() == 0
+    assert thread_pool.sized_from_config() is False
+    # ...and the floors are still what you get, because a server that cannot
+    # read its config must still run.
+    assert thread_pool.pool_sizes() == (thread_pool._MIN_DEFAULT,
+                                        thread_pool._MIN_POD)
+
+
+def test_the_configured_width_reaches_both_pools(holder):
+    holder(2048)
+    general, pod = thread_pool.pool_sizes()
+    assert general == thread_pool._MAX_DEFAULT      # 256
+    assert pod == 2048
+    assert pod > thread_pool._MIN_POD * 10
+
+
+def test_the_pod_ceiling_allows_a_hundred_pages_a_second():
+    """~100 pages/s at ~20 s per page is ~2000 forwards outstanding. A ceiling
+    below that caps the install regardless of its config."""
+    assert thread_pool._MAX_POD >= 2000
+
+
+def test_the_connection_pool_is_sized_the_same_way(holder):
+    holder(512)
+    assert pod_http._pool_size() == min(pod_http._MAX_POOL, 512)
+    holder(0)
+    assert pod_http._pool_size() == pod_http._DEFAULT_POOL
+
+
+# ------------------------------------------------------- using the pod pool #
+
+def test_the_page_forward_does_not_use_the_default_executor():
+    """Read from the source: the call must not be `asyncio.to_thread`.
+
+    Asserted statically because the alternative is asserting on a thread name
+    at runtime, which passes just as happily when the call has been moved back
+    and the pool happens to be large.
+    """
+    import inspect
+
+    from codai.ocr.manager import OcrManager
+    src = inspect.getsource(OcrManager._read_on_pods)
+    assert "_post_page_to_pod" in src
+    assert "run_on_pod_pool(" in src
+    # The page forward and the liveness probe both block on a POD.
+    assert "asyncio.to_thread(\n" not in src.replace(" ", "")
+    for call in ("_post_page_to_pod", "_pod_still_alive"):
+        i = src.index(call)
+        before = src[max(0, i - 400):i]
+        assert "run_on_pod_pool" in before, f"{call} is not on the pod pool"
+
+
+def test_the_pod_pool_is_not_the_default_pool(holder):
+    holder(512)
+    thread_pool._pod_pool = None
+    try:
+        ex = thread_pool.pod_executor()
+        assert ex._max_workers == 1024          # 512 * 2
+        assert ex._max_workers > thread_pool.pool_sizes()[0]
+    finally:
+        thread_pool._pod_pool = None
+
+
+def test_run_on_pod_pool_preserves_contextvars(holder):
+    """A page that lost its token budget because it changed pools would be a
+    nasty way to discover the pool changed."""
+    import contextvars
+    holder(64)
+    var = contextvars.ContextVar("probe", default="unset")
+
+    async def go():
+        var.set("set-by-caller")
+        return await thread_pool.run_on_pod_pool(var.get)
+
+    thread_pool._pod_pool = None
+    try:
+        assert asyncio.run(go()) == "set-by-caller"
+    finally:
+        thread_pool._pod_pool = None
+
+
+def test_the_pod_pool_grows_when_the_config_grows(holder):
+    holder(64)
+    thread_pool._pod_pool = None
+    try:
+        assert thread_pool.pod_executor()._max_workers == 128
+        holder(512)
+        assert thread_pool.resize_pod_pool() == 1024
+        assert thread_pool._pod_pool._max_workers == 1024
+        # ...and never shrinks: threads blocked on a page would have to go
+        # somewhere, and the answer that loses no pages is "nowhere".
+        holder(64)
+        assert thread_pool.resize_pod_pool() == 1024
+        assert thread_pool._pod_pool._max_workers == 1024
+    finally:
+        thread_pool._pod_pool = None
+
+
+def test_configure_resizes_the_pod_pool(holder):
+    """Raising max_concurrency through the admin API has to reach the threads,
+    or it only widens a semaphore that then queues on a pool it never touched."""
+    from codai.config import OcrConfig
+    from codai.ocr.manager import OcrManager
+
+    holder(64)
+    thread_pool._pod_pool = None
+    try:
+        thread_pool.pod_executor()
+        m = OcrManager()
+        cfg = OcrConfig()
+        cfg.max_concurrency = 1
+        m.configure(cfg)
+        holder(400)
+        cfg2 = OcrConfig()
+        cfg2.max_concurrency = 400          # a change, so pools are rebuilt
+        m.configure(cfg2)
+        assert thread_pool._pod_pool._max_workers == 800
+    finally:
+        thread_pool._pod_pool = None