fix(net): per-route coordinator body caps and counted baseline skips

README.md

@@ -211,6 +211,19 @@ always sees the brief and consents to that one task (D53). No token, no secondar
 market, money only ever at the edges. Deeper design:
 [docs/FEDERATION_V2.md](docs/FEDERATION_V2.md).
 
+Two coordinator limits a self-hoster can tune:
+
+- `PW_MAX_BODY_BYTES` (default 4 MiB) is the request-body ceiling. A body over its
+  route's cap gets a 413 before auth or parsing. The free-form routes (node results,
+  `/jobs`, blob uploads) are capped at the ceiling. `/tasks/{id}/deliver` is capped at
+  2,416,384 bytes, which fits a 200k-char deliverable even when it is fully
+  JSON-escaped. Every other route is capped at 64 KiB. A lower ceiling lowers all of
+  these caps.
+- `PW_BASELINE_WORKERS` (default 4) sets how many single-model baselines run at once.
+  Each one can hold its slot for up to 15 minutes. A job submitted while every slot
+  is busy gets no baseline, and the app falls back to the best single council answer.
+  Each skip is logged and counted as `baseline_skipped_busy` on `GET /metrics`.
+
 ## We work with the ledger open
 
 Every round of work is logged — what shipped, what we tried and reverted, and why —

passiveworkers/net/baseline.py

@@ -14,7 +14,9 @@
   3. None → the UI falls back to best-single-council-answer, clearly labelled.
 
 Runs in a background thread per job (parallel with the council), result stored on the
-job row. Failures are silent by design — the baseline must never block or fail a job.
+job row — but on at most PW_BASELINE_WORKERS (default 4) threads at once: a job submitted
+while every slot is busy gets no baseline, which the coordinator logs and counts
+(`baseline_skipped_busy` on GET /metrics). Failures never block or fail a job; they are logged.
 """
 
 from __future__ import annotations

passiveworkers/net/coordinator_app.py

@@ -23,10 +23,12 @@
   GET  /dashboard                                      → live operator map
   GET  /healthz
 
-Request bodies are capped at PW_MAX_BODY_BYTES (default 4 MiB) BEFORE any parsing or auth: a larger
-declared Content-Length, or a chunked body that streams past the cap, gets a 413 (F11 review). The
-cap sits above the largest legitimate body (a 200k-char /deliver JSON-escaped is ≤ ~2.4 MB); blob
-uploads keep their own tighter 512 KiB cap.
+Request bodies are capped per route BEFORE any parsing or auth: a larger declared Content-Length,
+or a chunked body that streams past the cap, gets a 413 (F11 review). PW_MAX_BODY_BYTES (default
+4 MiB) is the ceiling for every route and the cap for the free-form bodies (node /result dicts,
+/jobs, blob uploads — which keep their own tighter 512 KiB cap). Routes with a documented smaller
+bound get that instead: /deliver ≤ _DELIVER_BODY_CAP (a 200k-char deliverable, JSON-escaped), and
+the small field-capped bodies (/users, heartbeat, progress, enroll, …) ≤ 64 KiB.
 
 Run:  PW_TOKEN=… python -m passiveworkers.net.coordinator_app
 """
@@ -35,6 +37,7 @@
 
 import ipaddress
 import os
+import re
 import secrets
 import threading
 
@@ -78,10 +81,24 @@ def _startup_guard() -> None:
 
 
 _MAX_BODY_DEFAULT = 4 * 1024 * 1024   # see the module docstring for why 4 MiB
+# /deliver: 200k chars at ≤ 12 bytes each (an astral char JSON-escaped as a \uXXXX\uXXXX surrogate
+# pair), plus 16 KiB for signature + signer_pub (≤ 300 chars) and the JSON keys.
+_DELIVER_BODY_CAP = 12 * 200_000 + 16 * 1024
+_SMALL_BODY_CAP = 64 * 1024           # bodies whose every field is length-capped (or that have none)
+
+# POST routes whose body has a documented bound below the ceiling. Non-POST requests carry no body
+# (64 KiB); unlisted POSTs — /result's free-form dict, /jobs, blob uploads — get the ceiling. A new
+# POST route with a small body belongs here.
+_ROUTE_CAPS = (
+    (re.compile(r"/tasks/[^/]+/deliver"), _DELIVER_BODY_CAP),
+    (re.compile(r"/tasks/[^/]+/(?:progress|accept)"), _SMALL_BODY_CAP),
+    (re.compile(r"/jobs/[^/]+/(?:feedback|rate)"), _SMALL_BODY_CAP),
+    (re.compile(r"/(?:users|nodes/register|nodes/heartbeat|admin/enroll)"), _SMALL_BODY_CAP),
+)
 
 
 def _max_body_bytes() -> int:
-    """The request-body cap in bytes (PW_MAX_BODY_BYTES). Invalid or non-positive → the default:
+    """The request-body ceiling in bytes (PW_MAX_BODY_BYTES). Invalid or non-positive → the default:
     there is deliberately no 'unbounded' setting."""
     try:
         cap = int(os.environ.get("PW_MAX_BODY_BYTES", _MAX_BODY_DEFAULT))
@@ -90,6 +107,17 @@ def _max_body_bytes() -> int:
     return cap if cap > 0 else _MAX_BODY_DEFAULT
 
 
+def _body_cap_for(method: str, path: str) -> int:
+    """One request's cap: its route's documented bound, never above PW_MAX_BODY_BYTES."""
+    ceiling = _max_body_bytes()
+    if method != "POST":
+        return min(_SMALL_BODY_CAP, ceiling)
+    for pattern, cap in _ROUTE_CAPS:
+        if pattern.fullmatch(path):
+            return min(cap, ceiling)
+    return ceiling
+
+
 class _BodyTooLarge(HTTPException):
     def __init__(self) -> None:
         super().__init__(status_code=413, detail="request body too large")
@@ -109,7 +137,7 @@ async def __call__(self, scope, receive, send):
         if scope["type"] != "http":
             await self.app(scope, receive, send)
             return
-        cap = _max_body_bytes()
+        cap = _body_cap_for(scope.get("method", ""), scope.get("path", ""))
         for name, value in scope.get("headers", []):
             if name == b"content-length":
                 try:
@@ -511,13 +539,22 @@ def _baseline_workers() -> int:
 # workers are joined at interpreter exit, which would stall a coordinator shutdown for 15 minutes.
 _BASELINE_MAX = _baseline_workers()
 _baseline_slots = threading.BoundedSemaphore(_BASELINE_MAX)
+# Jobs that got no baseline because every slot was busy, since process start — on GET /metrics as
+# `baseline_skipped_busy`, so a degraded single-model compare under load is visible, not silent.
+_baseline_skipped = 0
+_baseline_skipped_lock = threading.Lock()
 
 
 def _start_baseline(job_id: str, question: str) -> bool:
-    """Start the baseline if a slot is free. False (and a log line) when all slots are busy."""
+    """Start the baseline if a slot is free. False (a log line + the skip counter) when all are busy."""
+    global _baseline_skipped
     slots = _baseline_slots   # release the semaphore we acquired, even if the global is rebound
     if not slots.acquire(blocking=False):
-        print(f"[baseline] skipped for job {job_id[:8]}: {_BASELINE_MAX} already running", flush=True)
+        with _baseline_skipped_lock:
+            _baseline_skipped += 1
+            skipped = _baseline_skipped
+        print(f"[baseline] skipped for job {job_id[:8]}: all {_BASELINE_MAX} slots busy "
+              f"(PW_BASELINE_WORKERS); {skipped} skipped since start", flush=True)
         return False
 
     def run() -> None:
@@ -602,7 +639,7 @@ def feedback(job_id: str, body: FeedbackBody, x_user_secret: str | None = Header
 
 @app.get("/metrics")
 def metrics():
-    return store.metrics()
+    return {**store.metrics(), "baseline_skipped_busy": _baseline_skipped}
 
 
 @app.get("/status")

tests/test_coordinator_limits.py

@@ -1,11 +1,14 @@
 """Coordinator resource bounds (review F11, F12, F13):
-  • F11 — request bodies over PW_MAX_BODY_BYTES get a 413 BEFORE parsing or auth, whether the
-    size is declared (Content-Length) or only discovered while streaming (chunked / lying client).
-  • F12 — baseline generation runs on at most PW_BASELINE_WORKERS threads, not one per job.
+  • F11 — request bodies over their route's cap (never above PW_MAX_BODY_BYTES) get a 413 BEFORE
+    parsing or auth, whether the size is declared (Content-Length) or only discovered while
+    streaming (chunked / lying client).
+  • F12 — baseline generation runs on at most PW_BASELINE_WORKERS threads, not one per job; a
+    job that finds every slot busy is skipped with a log line and a /metrics counter.
   • F13 — the hot lookups have SQLite indexes, created idempotently on every open.
 """
 import asyncio
 import importlib
+import json
 import sqlite3
 import threading
 
@@ -215,3 +218,104 @@ def test_indexes_are_added_to_an_existing_db_and_reopen_is_idempotent(tmp_path):
     Store(path)                                  # open → indexes created
     s = Store(path)                              # open again → no error, no duplicates
     assert _indexes(s.conn) == _EXPECTED
+
+
+# ----------------------------------------------------------------------- F11 per-route caps
+def _node_secret(client):
+    r = client.post("/nodes/register", headers=TOK,
+                    json={"owner": "op", "answer_model": "m", "profile": {"models": ["m"]}})
+    return r.json()["node_secret"]
+
+
+def _padded_json(prefix: bytes, size: int) -> bytes:
+    """A valid JSON object `prefix…}` padded with whitespace to exactly `size` bytes."""
+    assert len(prefix) + 1 <= size
+    return prefix + b" " * (size - len(prefix) - 1) + b"}"
+
+
+def test_deliver_body_exactly_at_the_cap_passes_and_one_byte_more_is_413(capp, client, monkeypatch):
+    """The worst-case legitimate /deliver — 200k astral chars, each JSON-escaped to 12 bytes —
+    fits under the route cap; the cap is exact (cap passes, cap+1 is refused before the endpoint)."""
+    calls = []
+    monkeypatch.setattr(capp.store, "deliver_assisted",
+                        lambda task_id, node_id, deliverable, **k:
+                        calls.append(len(deliverable)) or {"ok": True})
+    hdr = {"content-type": "application/json", "X-Node-Secret": _node_secret(client)}
+    prefix = json.dumps({"deliverable": "\U0001F600" * 200_000}).encode()[:-1]   # drop the "}"
+    cap = capp._DELIVER_BODY_CAP
+
+    r = client.post("/tasks/t1/deliver", content=_padded_json(prefix, cap), headers=hdr)
+    assert r.status_code == 200, r.text
+    assert calls == [200_000]
+
+    r = client.post("/tasks/t1/deliver", content=_padded_json(prefix, cap + 1), headers=hdr)
+    assert r.status_code == 413
+    assert r.json()["detail"] == "request body too large"
+    assert calls == [200_000]                    # the endpoint never ran
+
+
+_SMALL_ROUTES = ["/users", "/nodes/register", "/nodes/heartbeat", "/admin/enroll",
+                 "/tasks/t1/progress", "/tasks/t1/accept", "/jobs/j1/feedback", "/jobs/j1/rate"]
+
+
+@pytest.mark.parametrize("path", _SMALL_ROUTES)
+def test_small_body_routes_are_capped_at_64_kib(capp, client, path):
+    cap = 64 * 1024
+    hdr = {"content-type": "application/json"}
+    r = client.post(path, content=_padded_json(b'{"x": 1', cap + 1), headers=hdr)
+    assert r.status_code == 413
+    r = client.post(path, content=_padded_json(b'{"x": 1', cap), headers=hdr)
+    assert r.status_code != 413                  # at the cap: the normal auth/validation path
+
+
+def test_small_route_cap_also_applies_to_a_chunked_body(client):
+    def gen():                                   # no Content-Length → Transfer-Encoding: chunked
+        yield b'{"handle": "'
+        yield b"x" * (64 * 1024)
+        yield b'"}'
+
+    r = client.post("/users", content=gen(), headers={"content-type": "application/json"})
+    assert r.status_code == 413
+
+
+def test_free_form_routes_keep_the_ceiling(client):
+    """/result's dict and /jobs have no documented smaller bound, so a 1 MiB body still reaches
+    the endpoint (its auth check / body validation) under the default 4 MiB ceiling."""
+    body = _padded_json(b'{"text": "x"', 1024 * 1024)
+    hdr = {"content-type": "application/json", "X-Node-Secret": "nope", "X-User-Secret": "nope"}
+    assert client.post("/tasks/t1/result", content=body, headers=hdr).status_code == 401
+    assert client.post("/jobs", content=body, headers=hdr).status_code == 422   # no "question"
+
+
+def test_body_cap_for_routes_and_the_ceiling(capp, monkeypatch):
+    small, ceiling = 64 * 1024, 4 * 1024 * 1024
+    assert capp._body_cap_for("POST", "/tasks/t1/deliver") == capp._DELIVER_BODY_CAP
+    assert capp._body_cap_for("POST", "/tasks/t1/result") == ceiling
+    assert capp._body_cap_for("POST", "/jobs/j1/blobs/abc") == ceiling
+    assert capp._body_cap_for("POST", "/users") == small
+    assert capp._body_cap_for("GET", "/status") == small
+    assert capp._body_cap_for("POST", "/tasks/a/b/deliver") == ceiling   # not the deliver route
+    monkeypatch.setenv("PW_MAX_BODY_BYTES", "4096")     # an operator's lower ceiling wins everywhere
+    assert capp._body_cap_for("POST", "/tasks/t1/deliver") == 4096
+    assert capp._body_cap_for("POST", "/users") == 4096
+
+
+# ----------------------------------------------------------------------- F12 skipped baselines
+def test_busy_slots_skip_the_baseline_with_a_log_line_and_a_counter(capp, client, monkeypatch,
+                                                                    capsys):
+    release = threading.Event()
+    monkeypatch.setattr(capp, "_baseline_async", lambda job_id, question: release.wait(10))
+    before = set(threading.enumerate())
+    assert client.get("/metrics").json()["baseline_skipped_busy"] == 0
+    try:
+        assert capp._start_baseline("jobaaaa1", "q") is True
+        assert capp._start_baseline("jobaaaa2", "q") is True     # PW_BASELINE_WORKERS=2
+        assert capp._start_baseline("jobbusy1", "q") is False
+        assert capp._start_baseline("jobbusy2", "q") is False
+    finally:
+        release.set()
+    _join(_new_baseline_threads(before))
+    out = capsys.readouterr().out
+    assert "[baseline] skipped for job jobbusy1: all 2 slots busy" in out
+    assert "2 skipped since start" in out
+    assert client.get("/metrics").json()["baseline_skipped_busy"] == 2