an OCR run starved the server of threads to answer on (v0.3.102)

The admin RunPod page showed:

    error: Unexpected token '<', "<!doctype "... is not valid JSON

That '<' is nginx's HTML error page. The page was not failing to find data; its
fetch never got an answer, the proxy gave up, and JSON.parse was handed HTML.
Measured from inside the host, mid-run, with 64 pages in flight:

    /healthz            200 in 0.002s     no thread needed
    /v1/models          200 in 0.006s     no thread needed
    /metrics            200 in 0.18s      no thread needed
    /v1/runpod/status   no answer in 30s  needs a thread

`asyncio.to_thread` hands work to the loop's default executor, and a loop that
was never told otherwise builds one with min(32, cpu_count + 4) — 20 threads on
a 16-core host. Nothing in this codebase ever called set_default_executor. That
number is for incidental background work measured in milliseconds; what we put
on it is `pool.acquire()` waiting out a pod boot (MINUTES) and a per-page OCR
forward (30-300s each), `ocr.max_concurrency` of them at once. The log says it
plainly: "engine 'surya': 8 instance(s) ready, 8 lane(s) each — 64 pages at
once". 64 blocking calls, 20 threads. Everything else waits behind a page.

Nothing was broken. The server could not get a thread to answer on.

codai/api/thread_pool.py now sizes two pools from the concurrency the install
is CONFIGURED for rather than from its CPU count — these threads wait on other
machines, so what matters is how many remote calls may be outstanding, not how
many cores can run Python:

  * the DEFAULT executor, floor 32, for everything to_thread is used for today;
  * a separate POD executor, floor 64, for the calls that block for minutes, so
    that however many pages are in flight a control-plane request still finds a
    free thread.

Both entry points install it (the front blocks on pool.acquire for whole pod
boots, so it needs this at least as much as the engine). The pod pool copies
the context the way to_thread does — the per-page options and the token trace
are ContextVars, and run_in_executor does not copy by itself; a page losing its
budget because it changed pools would be a nasty way to learn that.

/v1/runpod/spend also took its snapshot ON the loop: pods_status() takes the
pool locks and the ledger reads a file, so a lock held by a provisioning thread
stalled every request rather than one. It runs in a thread now.

A thread waiting on a socket costs a stack and nothing else. The asymmetry is
the point: too small costs availability, too large costs memory we have.

Also fixes a test of my own from an hour ago that asserted "tomorrow 02:00" is
within 24 hours and went red at 00:22. The test was wrong about the clock, not
the code about the booking.

15 tests, including the live failure reproduced: fill the pod pool and check
the control plane still answers. Suite: 3166 passed, the same 13 known-stale
failures.
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.3.101"
+__version__ = "0.3.102"
 __tagline__ = "Complete orchestration, distribution and escalation of remotizable advanced inference."
 
 # Configure the CUDA caching allocator BEFORE torch is imported anywhere.

codai/api/app.py

@@ -102,6 +102,16 @@ async def lifespan(app: FastAPI):
     import asyncio as _asyncio
     boot_phase("lifespan start")
 
+    # Before anything schedules work: every `asyncio.to_thread` in this process
+    # uses the loop's default executor, and an untold loop sizes it for
+    # incidental background work rather than for a fan-out of blocking pod
+    # calls. See codai/api/thread_pool.py.
+    try:
+        from codai.api.thread_pool import install_default_executor
+        install_default_executor()
+    except Exception as exc:
+        print(f"[threads] could not size the default pool: {exc}", flush=True)
+
     async def _archive_cleanup_loop():
         from codai.api.archive import archive_manager
         while True:
@@ -853,18 +863,26 @@ async def runpod_spend(request: Request):
         raise HTTPException(status_code=403,
                             detail="this endpoint needs an admin-scoped API key")
     from codai.api import runpod_worker, runpod_ledger
-    try:
-        pods = runpod_worker.pods_status()
-    except Exception:
-        pods = []
-    try:
-        ledger = runpod_ledger.totals()
-    except Exception:
-        ledger = {"per_model": {}, "global": {}}
-    try:
-        schedules = runpod_worker.pools_schedule_status()
-    except Exception:
-        schedules = []
+
+    def _snapshot():
+        # Off the event loop. Each of these takes the pool locks and reads a
+        # file; none is slow by itself, but a lock held by a provisioning
+        # thread would otherwise stall the whole loop rather than one request.
+        try:
+            pods = runpod_worker.pods_status()
+        except Exception:
+            pods = []
+        try:
+            ledger = runpod_ledger.totals()
+        except Exception:
+            ledger = {"per_model": {}, "global": {}}
+        try:
+            schedules = runpod_worker.pools_schedule_status()
+        except Exception:
+            schedules = []
+        return pods, ledger, schedules
+
+    pods, ledger, schedules = await asyncio.to_thread(_snapshot)
     live_hourly = sum((p.get("hourly_usd") or 0.0) for p in pods if p.get("healthy"))
     live_cost = sum((p.get("live_cost_usd") or 0.0) for p in pods)
     caps = {}

codai/api/remote_gateway.py

@@ -956,7 +956,11 @@ class RemoteGatewayMiddleware:
             return pod_http.request(scope.get("method", "POST"), dest, data=body or None,
                                     headers=fwd, stream=True, timeout=(30, 3600))
         try:
-            resp = await asyncio.to_thread(_call)
+            # On the pod pool: one forwarded page can hold this thread for
+            # minutes, and `ocr.max_concurrency` of them must not be able to
+            # exhaust the executor every other endpoint shares.
+            from codai.api.thread_pool import run_on_pod_pool
+            resp = await run_on_pod_pool(_call)
             # The first request into a pod that came up seconds ago can land
             # before the far side is really serving — RunPod's proxy answers
             # 404 until its route settles, a catalogue may lag its /v1/models

codai/api/thread_pool.py

+# CoderAI - OpenAI-compatible API server
+# Copyright (C) 2026 Stefy Lanza <stefy@nexlab.net>
+#
+# This program is free software: you can redistribute it and/or modify
+# it under the terms of the GNU General Public License as published by
+# the Free Software Foundation, either version 3 of the License, or
+# (at your option) any later version.
+
+"""Thread pools sized for THIS server's work, not for Python's default guess.
+
+``asyncio.to_thread`` hands work to the event loop's default executor, and a
+loop that was never told otherwise builds one with ``min(32, cpu_count + 4)``
+threads — 20 on a 16-core host. That number is for incidental background work:
+a file read, a hash, something measured in milliseconds.
+
+coderai puts two very different kinds of call on it:
+
+* **Short and bounded** — a stat, a config read, a GPU probe.
+* **Long and blocking** — ``pool.acquire()`` waiting for a RunPod pod to boot
+  (MINUTES), and a per-page OCR forward to a pod (30-300 seconds each).
+
+With ``ocr.max_concurrency`` at 64, the second kind alone asks for more threads
+than the default pool has. Every one of them sits in a blocking HTTP call, the
+rest queue, and then an unrelated request that merely wants a thread — the
+admin page asking what the fleet is doing — waits behind a page of OCR.
+
+Observed on a live orchestrator, mid-run, 64 pages in flight:
+
+    /healthz            200 in 0.002s     (no thread needed)
+    /v1/models          200 in 0.006s     (no thread needed)
+    /metrics            200 in 0.18s      (no thread needed)
+    /v1/runpod/status   no answer in 30s  (needs a thread)
+
+The admin page's fetch timed out, nginx answered with its HTML error page, and
+the page showed `Unexpected token '<'`. Nothing was broken; the server was
+simply unable to get a thread to answer on.
+
+Two pools, because the two kinds of work must not compete:
+
+* the DEFAULT executor, sized for the install rather than the CPU count, for
+  everything ``asyncio.to_thread`` is used for today;
+* a separate POD executor for the calls that block for minutes, so that however
+  many pages are in flight, a control-plane request still finds a free thread.
+
+A thread waiting on a socket costs a stack and nothing else. The asymmetry is
+the point: being too small costs availability, being too large costs memory we
+have.
+"""
+
+import asyncio
+import contextvars
+import functools
+import os
+import threading
+from concurrent.futures import ThreadPoolExecutor
+from typing import Optional
+
+#: Floor for the general pool. Above Python's default, because the default is
+#: the thing that was too small, and below it nothing is gained by sizing.
+_MIN_DEFAULT = 32
+#: Floor for the pod pool. Every in-flight page may hold one for minutes.
+_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
+
+_lock = threading.Lock()
+_pod_pool: Optional[ThreadPoolExecutor] = None
+_installed = False
+
+
+def _configured_concurrency() -> int:
+    """The widest fan-out this install is configured for.
+
+    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.
+    """
+    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
+    return want
+
+
+def pool_sizes() -> tuple:
+    """(default_workers, pod_workers) for this install."""
+    want = _configured_concurrency()
+    general = max(_MIN_DEFAULT, min(_MAX_DEFAULT, want))
+    # Room for every concurrent page to be waiting on a pod at once, plus the
+    # pool's own boots and probes.
+    pod = max(_MIN_POD, min(_MAX_POD, want * 2))
+    return general, pod
+
+
+def install_default_executor(loop=None) -> Optional[ThreadPoolExecutor]:
+    """Give the running loop an executor sized for this server.
+
+    Idempotent and safe to call from either entry point — the front and the
+    engine are separate processes and each needs its own.
+    """
+    global _installed
+    with _lock:
+        if _installed:
+            return None
+        try:
+            loop = loop or asyncio.get_running_loop()
+        except RuntimeError:
+            return None
+        general, _pod = pool_sizes()
+        ex = ThreadPoolExecutor(max_workers=general,
+                                thread_name_prefix="coderai")
+        loop.set_default_executor(ex)
+        _installed = True
+        print(f"[threads] default pool {general} workers, pod pool {_pod} "
+              f"(python's default here would be "
+              f"{min(32, (os.cpu_count() or 1) + 4)})", flush=True)
+        return ex
+
+
+def pod_executor() -> ThreadPoolExecutor:
+    """The pool for calls that block on a POD rather than on this machine.
+
+    Separate from the default one on purpose: a page read that takes five
+    minutes must not be able to make the server unable to answer a question
+    about itself.
+    """
+    global _pod_pool
+    with _lock:
+        if _pod_pool is None:
+            _pod_pool = ThreadPoolExecutor(max_workers=pool_sizes()[1],
+                                           thread_name_prefix="coderai-pod")
+        return _pod_pool
+
+
+async def run_on_pod_pool(func, /, *args, **kwargs):
+    """``asyncio.to_thread``, but on the pod pool.
+
+    Copies the context the same way ``to_thread`` does, so anything reading a
+    ContextVar — the OCR per-page options, the token trace — behaves
+    identically whichever pool it lands on. ``run_in_executor`` does NOT do
+    this by itself, and a page losing its budget because it was moved to
+    another pool would be a nasty way to find that out.
+    """
+    loop = asyncio.get_running_loop()
+    ctx = contextvars.copy_context()
+    call = functools.partial(ctx.run, func, *args, **kwargs)
+    return await loop.run_in_executor(pod_executor(), call)

codai/frontproxy/app.py

@@ -977,8 +977,13 @@ class FrontProxy:
             if shared:
                 _rw.shared_pool_key(mcfg, model or "", "")   # raises if not shareable
             pool = _rw.get_pod_pool(key, rp, mcfg, served, shared=bool(shared))
-            # Provisioning blocks for minutes on a cold pod; keep the loop free.
-            handle, url = await _asyncio.to_thread(pool.acquire)
+            # Provisioning blocks for MINUTES on a cold pod. Keeping it off the
+            # loop was never enough: on the shared default executor, N of these
+            # waits occupy every thread the process has, and an unrelated
+            # request that merely needs one — the admin page asking what the
+            # fleet is doing — queues behind a pod boot. Its own pool.
+            from codai.api.thread_pool import run_on_pod_pool
+            handle, url = await run_on_pod_pool(pool.acquire)
         except Exception as exc:
             print("[runpod-spill] pod acquire failed, falling back to local: %s" % exc,
                   flush=True)
@@ -2904,6 +2909,13 @@ def build_app(config, config_dir=None) -> FastAPI:
 
     @app.on_event("startup")
     async def _startup():
+        # The front blocks on pool.acquire() for whole pod boots, so it needs
+        # the sized pools at least as much as the engine does.
+        try:
+            from codai.api.thread_pool import install_default_executor
+            install_default_executor()
+        except Exception as exc:
+            print(f"[threads] could not size the default pool: {exc}", flush=True)
         front.supervisor = EngineSupervisor(config, None, front.registry,
                                             models_path=front._models_path,
                                             internal_token=front.internal_token,

tests/test_autoupgrade_force_options.py

@@ -60,7 +60,12 @@ def test_force_at_books_the_time_it_was_given(tmp_path):
     assert at is not None
     lt = time.localtime(at)
     assert (lt.tm_hour, lt.tm_min) == (2, 0), time.ctime(at)
-    assert time.time() < at <= time.time() + 86400 + 60
+    # "tomorrow 02:00" is between 2 and 26 hours away depending on the hour the
+    # suite runs. The first version of this asserted "within 24h" and went red
+    # at 00:22 — the test was wrong about the clock, not the code about the
+    # booking. What matters is that it is in the future and is tomorrow.
+    assert time.time() < at <= time.time() + 2 * 86400
+    assert lt.tm_mday == time.localtime(time.time() + 86400).tm_mday
 
 
 def test_force_at_takes_a_relative_time(tmp_path):

tests/test_thread_pool_sizing.py

+"""An OCR run must not make the server unable to answer questions about itself.
+
+`asyncio.to_thread` uses the loop's default executor, and a loop that was never
+told otherwise builds one with min(32, cpu_count + 4) — 20 threads on a 16-core
+host. That is a number for incidental background work. coderai puts calls that
+block for MINUTES on it: `pool.acquire()` waiting out a pod boot, and a per-page
+OCR forward.
+
+Measured live 2026-10-16 with 64 pages in flight ("engine 'surya': 8 instance(s)
+ready, 8 lane(s) each — 64 pages at once"):
+
+    /healthz            200 in 0.002s     no thread needed
+    /v1/models          200 in 0.006s     no thread needed
+    /metrics            200 in 0.18s      no thread needed
+    /v1/runpod/status   no answer in 30s  needs a thread
+
+The admin page's fetch timed out, nginx returned its HTML error page, and the
+page showed `Unexpected token '<', "<!doctype "... is not valid JSON`. Nothing
+was broken — the server could not get a thread to answer on.
+"""
+
+import asyncio
+import contextvars
+import types
+
+import pytest
+
+from codai.api import thread_pool
+
+
+def _cfg(monkeypatch, max_concurrency):
+    mgr = types.SimpleNamespace(config=types.SimpleNamespace(
+        ocr=types.SimpleNamespace(max_concurrency=max_concurrency)))
+    import codai.config
+    monkeypatch.setattr(codai.config, "config_manager", mgr, raising=False)
+
+
+# ------------------------------------------------------------------ sizing #
+def test_the_pools_are_bigger_than_pythons_default(monkeypatch):
+    """Python's default is the thing that was too small."""
+    import os
+    _cfg(monkeypatch, 64)
+    general, pod = thread_pool.pool_sizes()
+    assert general > min(32, (os.cpu_count() or 1) + 4)
+    assert pod > general
+
+
+def test_the_pod_pool_holds_every_page_at_once(monkeypatch):
+    """Each in-flight page may sit on a pod call for minutes; if they cannot
+    all have a thread, the ones that cannot are just waiting for a thread
+    instead of for a pod, which is the bug."""
+    _cfg(monkeypatch, 128)
+    general, pod = thread_pool.pool_sizes()
+    assert pod >= 128
+    assert general >= 128
+
+
+def test_a_small_install_still_gets_a_floor(monkeypatch):
+    _cfg(monkeypatch, 2)
+    general, pod = thread_pool.pool_sizes()
+    assert general == thread_pool._MIN_DEFAULT
+    assert pod == thread_pool._MIN_POD
+
+
+def test_an_absurd_config_cannot_ask_for_ten_thousand_stacks(monkeypatch):
+    _cfg(monkeypatch, 100000)
+    general, pod = thread_pool.pool_sizes()
+    assert general == thread_pool._MAX_DEFAULT
+    assert pod == thread_pool._MAX_POD
+
+
+def test_no_config_is_not_an_error(monkeypatch):
+    import codai.config
+    monkeypatch.setattr(codai.config, "config_manager", None, raising=False)
+    general, pod = thread_pool.pool_sizes()
+    assert general == thread_pool._MIN_DEFAULT
+    assert pod == thread_pool._MIN_POD
+
+
+# ------------------------------------------------------------- the install #
+def test_installing_replaces_the_loops_executor(monkeypatch):
+    monkeypatch.setattr(thread_pool, "_installed", False, raising=False)
+    _cfg(monkeypatch, 64)
+    seen = {}
+
+    async def _main():
+        loop = asyncio.get_running_loop()
+        loop.set_default_executor = lambda ex: seen.setdefault("ex", ex)
+        thread_pool.install_default_executor()
+
+    asyncio.run(_main())
+    assert seen.get("ex") is not None
+    assert seen["ex"]._max_workers >= 64
+
+
+def test_installing_twice_is_a_no_op(monkeypatch):
+    """The front and the engine are separate processes, but a process that
+    calls it twice must not swap the pool out from under running work."""
+    monkeypatch.setattr(thread_pool, "_installed", False, raising=False)
+    _cfg(monkeypatch, 64)
+    calls = []
+
+    async def _main():
+        loop = asyncio.get_running_loop()
+        loop.set_default_executor = lambda ex: calls.append(ex)
+        thread_pool.install_default_executor()
+        thread_pool.install_default_executor()
+
+    asyncio.run(_main())
+    assert len(calls) == 1
+
+
+def test_installing_outside_a_loop_does_not_raise(monkeypatch):
+    monkeypatch.setattr(thread_pool, "_installed", False, raising=False)
+    assert thread_pool.install_default_executor() is None
+
+
+# --------------------------------------------------------- the separation #
+def test_the_pod_pool_is_not_the_default_pool(monkeypatch):
+    """The whole point: a page read cannot consume the threads the control
+    plane needs."""
+    monkeypatch.setattr(thread_pool, "_pod_pool", None, raising=False)
+    monkeypatch.setattr(thread_pool, "_installed", False, raising=False)
+    _cfg(monkeypatch, 64)
+    seen = {}
+
+    async def _main():
+        loop = asyncio.get_running_loop()
+        real = loop.set_default_executor
+        loop.set_default_executor = lambda ex: (seen.setdefault("default", ex),
+                                                real(ex))
+        thread_pool.install_default_executor()
+        return thread_pool.pod_executor()
+
+    pod = asyncio.run(_main())
+    assert pod is not seen["default"]
+
+
+def test_a_control_plane_call_gets_a_thread_while_the_pod_pool_is_full(monkeypatch):
+    """The live failure, reproduced: saturate the pod pool and check that work
+    on the default pool still runs."""
+    import threading
+    monkeypatch.setattr(thread_pool, "_pod_pool", None, raising=False)
+    _cfg(monkeypatch, 2)          # pod pool = the floor
+    hold = threading.Event()
+    lock = threading.Lock()
+    started = set()
+
+    def _block(i):
+        with lock:
+            started.add(i)
+        hold.wait(10)
+        return i
+
+    async def _main():
+        size = thread_pool.pool_sizes()[1]
+        tasks = [asyncio.ensure_future(thread_pool.run_on_pod_pool(_block, i))
+                 for i in range(size)]
+        # Yield, rather than block: waiting for them on the loop thread would
+        # stop the loop ever starting them. (It did, the first time.)
+        for _ in range(1000):
+            await asyncio.sleep(0.01)
+            with lock:
+                if len(started) >= size:
+                    break
+        with lock:
+            assert len(started) >= size, "the pod pool never filled"
+        # …and the control plane still answers, on the other pool.
+        answered = await asyncio.wait_for(
+            asyncio.to_thread(lambda: "fleet status"), timeout=5)
+        hold.set()
+        await asyncio.gather(*tasks)
+        return answered
+
+    assert asyncio.run(_main()) == "fleet status"
+
+
+# -------------------------------------------------------------- the context #
+def test_the_pod_pool_carries_contextvars_like_to_thread():
+    """`run_in_executor` does not copy the context by itself. The OCR per-page
+    options and the token trace are ContextVars, so a page losing its budget
+    because it moved pools would be a nasty way to discover that."""
+    var = contextvars.ContextVar("page_budget", default=0)
+
+    async def _main():
+        var.set(3072)
+        return await thread_pool.run_on_pod_pool(var.get)
+
+    assert asyncio.run(_main()) == 3072
+
+
+def test_the_pod_pool_propagates_exceptions():
+    async def _main():
+        with pytest.raises(ValueError):
+            await thread_pool.run_on_pod_pool(
+                lambda: (_ for _ in ()).throw(ValueError("boom")))
+
+    asyncio.run(_main())
+
+
+def test_arguments_reach_the_callable():
+    async def _main():
+        return await thread_pool.run_on_pod_pool(lambda a, b=0: a + b, 2, b=3)
+
+    assert asyncio.run(_main()) == 5
+
+
+# ------------------------------------------------------- the call sites moved #
+def test_the_blocking_pod_waits_use_the_pod_pool():
+    """Both are minutes-long; on the shared pool they were the starvation."""
+    import pathlib
+    from codai import frontproxy
+    from codai.api import remote_gateway
+
+    front = pathlib.Path(frontproxy.app.__file__).read_text()
+    assert "run_on_pod_pool(pool.acquire)" in front
+    assert "_asyncio.to_thread(pool.acquire)" not in front
+
+    gw = pathlib.Path(remote_gateway.__file__).read_text()
+    assert "run_on_pod_pool(_call)" in gw
+
+
+def test_the_fleet_snapshot_is_off_the_event_loop():
+    """pods_status() takes the pool locks; on the loop a contended lock stalls
+    every request rather than one."""
+    import pathlib
+    from codai.api import app as api_app
+    src = pathlib.Path(api_app.__file__).read_text()
+    blk = src.split("async def runpod_spend")[1].split("@app.get")[0]
+    assert "asyncio.to_thread(_snapshot)" in blk