a page forward is a wait on a socket, not a thread (v0.4.9)

0.4.8 moved the forward off the default 32-thread pool and the measured result
on the live orchestrator was 33 -> 483 pages in flight, 1.65-2.2 -> 7.99
pages/s, and a control-plane request that used to time out at 40 s answering in
12 ms. But a thread is still the wrong unit for the wait: ~60 s per page
against a real corpus means 15 documents/second at ~3.5 pages each is ~3,200
outstanding forwards, against a pod-pool ceiling of 2048. Threads cannot get
there, and should not try.

So the hot path is a coroutine. `pod_http` gains `apost`/`aget` on httpx --
already in the stack (codai/cluster/nodes.py, codai/main.py), so no new
dependency -- with everything the sync path learned kept intact:

* the pinned TLS context, unchanged: a direct-TCP pod serves HTTPS with a
  certificate signed by THIS install's CA at an ip the certificate cannot name,
  so the CA is trusted and the hostname check is off. NOT verify=False, which
  would accept anybody;
* one client per event loop AND per pinned-ness, because a client belongs to
  the loop its sockets were opened on and the two trust settings must never
  share one;
* connection limits from the same `_pool_size()` the sync session uses;
* a proxied pod (http://, or https:// to a name) stays an ordinary request.

`_apost_page_to_pod` and `_apod_still_alive` mirror the sync twins, and both
twins now share ONE request builder and ONE response reader -- two copies of
that field list is how an engine name, a dpi or a token budget comes to apply
on one transport and not the other. Every distinction 0.4.2-0.4.6 established
is preserved and tested on the new path: an ANSWER (even 503) is a verdict
about the pod and belongs on the refusal path, nothing-came-back is
PodUnreachable, a dead pooled socket is retried on a fresh one, and a timeout
is never retried -- resending a page the pod is still working on is a second
card doing the same work and a double bill.

`ocr.pod_http_async` (default on, admin API + settings control) puts the
forward back on the pod thread pool. That path is kept working deliberately,
not as dead code: this is a new transport on the hottest path in the
subsystem, and an operator who hits something odd should change one setting
rather than wait for a release.

Four existing test files had to be repaired, and the reason is worth keeping:
they stubbed only the SYNC twin, so the moment the async default landed they
stopped testing their own stubs and began dialling 10.0.0.1 for real -- 300 s
hangs where there should have been failures. Their helpers now stub both twins
via `_stub_page_post`, and the policy tests say `pod_http_async: False` out
loud because they stub `pod_http.post` specifically. A test that quietly
measures something else is worse than one that breaks, and this change
produced four of them.

21 new transport tests. 3345 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.8"
+__version__ = "0.4.9"
 __tagline__ = "Complete orchestration, distribution and escalation of remotizable advanced inference."
 
 # Configure the CUDA caching allocator BEFORE torch is imported anywhere.

codai/admin/routes.py

@@ -4138,6 +4138,7 @@ def build_settings_dict(c, gpu_cards):
             "max_pages": c.ocr.max_pages,
             "pod_pipeline": c.ocr.pod_pipeline,
             "pod_pipeline_wait_s": c.ocr.pod_pipeline_wait_s,
+            "pod_http_async": c.ocr.pod_http_async,
             "page_image_format": c.ocr.page_image_format,
             "page_image_jpeg_quality": c.ocr.page_image_jpeg_quality,
             "page_image_min_dpi": c.ocr.page_image_min_dpi,
@@ -5001,6 +5002,10 @@ async def api_save_settings(request: Request, username: str = Depends(require_ad
                 o.pod_pipeline_wait_s = max(0.0, float(d.get("pod_pipeline_wait_s") or 0))
             except (TypeError, ValueError):
                 pass
+        if "pod_http_async" in d:
+            v = d.get("pod_http_async")
+            o.pod_http_async = (v.strip().lower() in ("1", "true", "on", "yes")
+                                if isinstance(v, str) else bool(v))
         if "page_image_format" in d:
             _pf = str(d.get("page_image_format") or "").strip().lower()
             # An unknown value leaves the setting alone rather than silently

codai/admin/templates/settings.html

@@ -1150,6 +1150,18 @@
         that is momentarily full, not a cold one: after a timeout the next pages go
         straight through for five minutes rather than paying this per page.</span>
       </div>
+      <div class="form-group" style="flex:1;min-width:200px">
+        <label class="form-label">Forward pages to pods</label>
+        <select id="s-ocr-pod-http-async" class="form-input">
+          <option value="1">On the event loop (no thread per page)</option>
+          <option value="0">On the pod thread pool (one thread per page)</option>
+        </select>
+        <span class="form-hint">A forward waits <b>tens of seconds</b> on a socket, and
+        the thread-per-page path is bounded by the pod pool &mdash; 15 documents/second
+        needs thousands of forwards outstanding, which is not a number of threads to
+        carry. Change this only to rule the transport out while diagnosing something;
+        the thread path is kept working for that.</span>
+      </div>
       <div class="form-group" style="flex:1;min-width:190px">
         <label class="form-label">Page image sent to the model</label>
         <select id="s-ocr-page-image-format" class="form-input">
@@ -2311,6 +2323,7 @@ async function loadSettings(){
     document.getElementById('s-ocr-max-pages').value = ocr.max_pages ?? 0;
     document.getElementById('s-ocr-pod-pipeline').value = ocr.pod_pipeline ?? 'prefer';
     document.getElementById('s-ocr-pod-pipeline-wait').value = ocr.pod_pipeline_wait_s ?? 60;
+    document.getElementById('s-ocr-pod-http-async').value = (ocr.pod_http_async ?? true) ? '1' : '0';
     document.getElementById('s-ocr-page-image-format').value = ocr.page_image_format ?? 'auto';
     document.getElementById('s-ocr-page-image-quality').value = ocr.page_image_jpeg_quality ?? 95;
     document.getElementById('s-ocr-page-image-min-dpi').value = ocr.page_image_min_dpi ?? 150;
@@ -2624,6 +2637,7 @@ async function saveSettings(silent){
       max_pages: parseInt(document.getElementById('s-ocr-max-pages').value) || 0,
       pod_pipeline: document.getElementById('s-ocr-pod-pipeline').value || 'prefer',
       pod_pipeline_wait_s: parseFloat(document.getElementById('s-ocr-pod-pipeline-wait').value) || 0,
+      pod_http_async: document.getElementById('s-ocr-pod-http-async').value === '1',
       page_image_format: document.getElementById('s-ocr-page-image-format').value || 'auto',
       page_image_jpeg_quality: parseInt(document.getElementById('s-ocr-page-image-quality').value) || 95,
       page_image_min_dpi: parseFloat(document.getElementById('s-ocr-page-image-min-dpi').value) || 0,

codai/api/pod_http.py

@@ -152,3 +152,114 @@ def head(url: str, **kw):
     if is_pinned_url(url):
         return request("HEAD", url, **kw)
     return requests.head(url, **kw)
+
+
+# --------------------------------------------------------------------------- #
+# the async path
+# --------------------------------------------------------------------------- #
+#
+# A page forward blocks for ~60 s against a real corpus, and the sync path above
+# costs one THREAD for every one of them. That is survivable at tens of pages in
+# flight and not at thousands: 15 documents/second at ~3.5 pages each and ~60 s
+# per page is ~3,200 forwards outstanding, which is not a number of threads any
+# install should be asked to carry (see codai/api/thread_pool.py, where the pod
+# pool is capped at 2048 for exactly that reason).
+#
+# So the hot path gets a coroutine instead. Everything the sync path learned the
+# hard way is kept:
+#
+# * the pod serves HTTPS with a certificate signed by THIS install's CA, at an
+#   ip that was not known when it was issued -- so the CA is trusted and the
+#   hostname check is off, and NOT verify=False, which would accept anyone;
+# * one client per event loop, with connection limits from the same
+#   `_pool_size()` the sync session uses;
+# * a proxied pod (http://, or https:// to a name) is an ordinary request.
+#
+# httpx rather than aiohttp: it is already in the stack (codai/cluster/nodes.py,
+# codai/main.py) and takes an ssl.SSLContext directly, so the pinned context is
+# the same object the sync path builds.
+
+_aclients: dict = {}
+_alock = threading.Lock()
+
+
+def _pinned_ssl_context():
+    """The context the sync adapter installs, built once per call site.
+
+    CA from this install, hostname matching off. The two together are the whole
+    trust decision for a direct-TCP pod: the certificate must be one we signed,
+    and the name on it cannot be the ip we reach it at.
+    """
+    from codai.api.pod_tls import ensure_ca
+    ctx = ssl.create_default_context(cafile=ensure_ca())
+    ctx.check_hostname = False
+    ctx.verify_mode = ssl.CERT_REQUIRED
+    return ctx
+
+
+def async_available() -> bool:
+    """Whether the async transport can be used at all."""
+    try:
+        import httpx            # noqa: F401
+        return True
+    except Exception:
+        return False
+
+
+def aclient(pinned: bool):
+    """The AsyncClient for this event loop, created on first use.
+
+    Keyed by loop AND by pinned-ness: the two need different trust settings, and
+    a client built for one must never serve the other. Keyed by loop because an
+    httpx client belongs to the loop whose transport it opened its sockets on --
+    sharing one across loops is how a test suite ends up with sockets attached
+    to a closed loop.
+    """
+    import asyncio
+
+    import httpx
+
+    try:
+        loop_key = id(asyncio.get_running_loop())
+    except RuntimeError:
+        loop_key = 0
+    key = (loop_key, bool(pinned))
+    with _alock:
+        c = _aclients.get(key)
+        if c is not None and not c.is_closed:
+            return c
+        size = _pool_size()
+        limits = httpx.Limits(max_connections=size,
+                              max_keepalive_connections=size)
+        kw = {"limits": limits, "timeout": httpx.Timeout(600.0, connect=30.0)}
+        if pinned:
+            kw["verify"] = _pinned_ssl_context()
+        c = httpx.AsyncClient(**kw)
+        _aclients[key] = c
+        return c
+
+
+async def apost(url: str, files=None, data=None, headers=None, timeout=600):
+    """POST on the event loop. Answers like the sync path: an object with
+    ``status_code``, ``text`` and ``json()``, so callers need not care which
+    transport ran."""
+    c = aclient(is_pinned_url(url))
+    return await c.post(url, files=files, data=data, headers=headers,
+                        timeout=timeout)
+
+
+async def aget(url: str, headers=None, timeout=15):
+    c = aclient(is_pinned_url(url))
+    return await c.get(url, headers=headers, timeout=timeout)
+
+
+async def aclose_all() -> None:
+    """Close every client (shutdown, tests). Never raises."""
+    with _alock:
+        clients = list(_aclients.values())
+        _aclients.clear()
+    for c in clients:
+        try:
+            await c.aclose()
+        except Exception:
+            pass

codai/config.py

@@ -1088,6 +1088,15 @@ class OcrConfig:
     #   jpeg   lossy, 8 ms and a third of the bytes. Whether it costs accuracy
     #          is a question about a corpus rather than about a codec, so it is
     #          opt-in and not a default.
+    # Forward a page to a pod on the EVENT LOOP rather than on a thread.
+    #
+    # The sync path costs one thread per page in flight, and a thread is the
+    # wrong unit for a ~60 s wait on a socket: 15 documents a second at ~3.5
+    # pages each needs ~3,200 outstanding, against a thread-pool ceiling of
+    # 2048. On by default; off puts the forward back on the pod thread pool,
+    # which is kept working so an operator who hits something odd on a new
+    # transport can change one setting instead of waiting for a release.
+    pod_http_async: bool = True
     page_image_format: str = "auto"
     page_image_jpeg_quality: int = 95
     # The bounds within which a scan may be passed through. Below the floor the
@@ -2106,6 +2115,7 @@ class ConfigManager:
                 "max_pages": self.config.ocr.max_pages,
                 "pod_pipeline": self.config.ocr.pod_pipeline,
                 "pod_pipeline_wait_s": self.config.ocr.pod_pipeline_wait_s,
+                "pod_http_async": self.config.ocr.pod_http_async,
                 "page_image_format": self.config.ocr.page_image_format,
                 "page_image_jpeg_quality": self.config.ocr.page_image_jpeg_quality,
                 "page_image_min_dpi": self.config.ocr.page_image_min_dpi,

tests/test_ocr_dual_mode.py

@@ -229,6 +229,22 @@ def test_an_abandoned_document_stops_splitting():
         next(g)
 
 
+def _stub_page_post(monkeypatch, fn):
+    """Stub BOTH page-forward twins with one callable.
+
+    The forward has two transports -- a coroutine on the event loop, and the
+    sync one on the pod thread pool -- and a test that stubs only one silently
+    stops testing its own stub the moment the default changes. Not
+    hypothetical: when the async path landed these tests began dialling
+    10.0.0.1 for real and hung for 300 s each instead of failing. Whatever
+    `fn` does, including raising, it now does on either path.
+    """
+    async def _async(*a, **k):
+        return fn(*a, **k)
+    monkeypatch.setattr(om, "_post_page_to_pod", fn)
+    monkeypatch.setattr(om, "_apost_page_to_pod", _async)
+
+
 # ------------------------------------------- the pipeline choice at runtime #
 class _Cfg:
     surya_serve = "model"
@@ -334,9 +350,9 @@ def test_a_pod_error_is_not_swallowed(monkeypatch):
     mgr._sem = asyncio.Semaphore(4)
 
     _fake_placement(monkeypatch)
-    monkeypatch.setattr(om, "_post_page_to_pod",
-                        lambda *a, **k: (_ for _ in ()).throw(
-                            OcrError("pod said 500", status=500)))
+    _stub_page_post(monkeypatch,
+                    lambda *a, **k: (_ for _ in ()).throw(
+                        OcrError("pod said 500", status=500)))
     with pytest.raises(OcrError):
         asyncio.run(mgr._read_on_pods(
             _pdf(2), "d.pdf", "application/pdf", "m",
@@ -355,7 +371,7 @@ def test_pages_come_back_in_document_order(monkeypatch):
     def _post(base, key, blob, name, dpi, opts, detect, engine="surya"):
         seen.append(name)
         return {"engine": engine, "pages": [{"text": name, "width": 1, "height": 2}]}
-    monkeypatch.setattr(om, "_post_page_to_pod", _post)
+    _stub_page_post(monkeypatch, _post)
     name, pages, blank, stamps, sigs = asyncio.run(mgr._read_on_pods(
         _pdf(5), "d.pdf", "application/pdf", "m",
         dpi=250, page_opts=None, detect=None, cfg=_Cfg()))
@@ -373,8 +389,8 @@ def test_one_page_is_one_request_so_fan_out_survives(monkeypatch):
     mgr._sem = asyncio.Semaphore(8)
     chose = {"n": 0}
     _fake_placement(monkeypatch, on_choose=lambda: chose.__setitem__("n", chose["n"] + 1))
-    monkeypatch.setattr(om, "_post_page_to_pod",
-                        lambda *a, **k: {"engine": "surya", "pages": [{"text": "x"}]})
+    _stub_page_post(monkeypatch,
+                    lambda *a, **k: {"engine": "surya", "pages": [{"text": "x"}]})
     asyncio.run(mgr._read_on_pods(_pdf(7), "d.pdf", "application/pdf", "m",
                                   dpi=250, page_opts=None, detect=None, cfg=_Cfg()))
     assert chose["n"] == 7, "the upstream must be chosen per PAGE"
@@ -391,9 +407,9 @@ def test_a_single_image_is_sent_whole(monkeypatch):
 
     _fake_placement(monkeypatch)
     sent = []
-    monkeypatch.setattr(om, "_post_page_to_pod",
-                        lambda b, k, blob, name, *a: sent.append(len(blob)) or
-                        {"engine": "surya", "pages": [{"text": "x"}]})
+    _stub_page_post(monkeypatch,
+                    lambda b, k, blob, name, *a: sent.append(len(blob)) or
+                    {"engine": "surya", "pages": [{"text": "x"}]})
     asyncio.run(mgr._read_on_pods(buf.getvalue(), "x.png", "image/png", "m",
                                   dpi=250, page_opts=None, detect=None, cfg=_Cfg()))
     assert len(sent) == 1

tests/test_ocr_pod_pipeline_gate.py

@@ -228,7 +228,23 @@ def _pods_answer(monkeypatch, fn):
     from codai.models import placement
     monkeypatch.setattr(placement, "choose", lambda *_a, **_k: _Lease())
     monkeypatch.setattr(placement, "spill_threshold", lambda *_a, **_k: 0)
+    _stub_page_post(monkeypatch, fn)
+
+
+def _stub_page_post(monkeypatch, fn):
+    """Stub BOTH page-forward twins with one callable.
+
+    The forward has two transports -- a coroutine on the event loop and the
+    sync one on the pod thread pool -- and a test that stubs only one silently
+    stops testing its own stub when the default changes. That is not
+    hypothetical: when the async path landed, these tests started dialling
+    10.0.0.1 for real and hung for 300 s instead of failing. Whatever `fn`
+    does, including raising, it now does on either path.
+    """
+    async def _async(*a, **k):
+        return fn(*a, **k)
     monkeypatch.setattr(om, "_post_page_to_pod", fn)
+    monkeypatch.setattr(om, "_apost_page_to_pod", _async)
 
 
 async def _read(mgr, cfg):

tests/test_ocr_pod_pipeline_policy.py

@@ -65,7 +65,15 @@ def _cfg(**kw):
             "olmocr_serve": "model",
             "olmocr_model_id": "allenai/olmOCR-2-7B-1025-FP8",
             "max_concurrency": 8, "page_window": 4,
-            "pod_pipeline": "prefer", "pod_pipeline_wait_s": 60.0}
+            "pod_pipeline": "prefer", "pod_pipeline_wait_s": 60.0,
+            # EXPLICIT, not incidental: these tests stub the SYNC transport
+            # (`_post_page_to_pod`, `pod_http.post`), so they must say which
+            # path they are exercising. When the async default was introduced
+            # they silently stopped testing their own stubs and started dialling
+            # 10.0.0.1 for real -- 300 s of nothing instead of a failure. A test
+            # that quietly measures a different thing is worse than one that
+            # breaks.
+            "pod_http_async": False}
     base.update(kw)
     return types.SimpleNamespace(**base)
 
@@ -222,9 +230,12 @@ def _one_page(monkeypatch):
 
 
 def _page_read(monkeypatch):
-    monkeypatch.setattr(om, "_post_page_to_pod",
-                        lambda *_a, **_k: {"engine": "surya",
-                                           "pages": [{"text": "ok"}]})
+    reply = {"engine": "surya", "pages": [{"text": "ok"}]}
+
+    async def _aok(*_a, **_k):
+        return reply
+    monkeypatch.setattr(om, "_post_page_to_pod", lambda *_a, **_k: reply)
+    monkeypatch.setattr(om, "_apost_page_to_pod", _aok)
 
 
 async def _read(cfg):