fix(net): blob route cap, root_path/trailing-slash cap matching, throttled baseline-skip log

README.md

@@ -214,11 +214,13 @@ market, money only ever at the edges. Deeper design:
 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
+  route's cap gets a 413 before auth or parsing. The free-form routes (node results
+  and `/jobs`) are capped at the ceiling. Blob uploads are capped at 528 KiB: one
+  512 KiB chunk plus 16 KiB of slack. `/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.
+  these caps. The cap is chosen from the path the app routes on, so a trailing slash
+  or a `--root-path` prefix gets the same cap as the plain route.
 - `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.

passiveworkers/net/coordinator_app.py

@@ -26,9 +26,10 @@
 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.
+/jobs). Routes with a documented smaller bound get that instead: /deliver ≤ _DELIVER_BODY_CAP (a
+200k-char deliverable, JSON-escaped), blob uploads ≤ _BLOB_BODY_CAP (one 512 KiB chunk + slack),
+and the small field-capped bodies (/users, heartbeat, progress, enroll, …) ≤ 64 KiB. The route is
+matched on the app-relative path (root_path and trailing slashes stripped), as Starlette routes it.
 
 Run:  PW_TOKEN=… python -m passiveworkers.net.coordinator_app
 """
@@ -85,12 +86,17 @@ def _startup_guard() -> None:
 # 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)
+_BLOB_CAP = 512 * 1024   # max bytes for one content-addressed chunk
+# A blob upload is the raw chunk (application/octet-stream, no JSON/multipart envelope), so the
+# route cap only adds slack: put_blob's exact _BLOB_CAP check ("chunk too large") decides at the edge.
+_BLOB_BODY_CAP = _BLOB_CAP + 16 * 1024
 
 # 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.
+# (64 KiB); unlisted POSTs — /result's free-form dict, /jobs — 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"/jobs/[^/]+/blobs/[^/]+"), _BLOB_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),
@@ -107,11 +113,21 @@ def _max_body_bytes() -> int:
     return cap if cap > 0 else _MAX_BODY_DEFAULT
 
 
-def _body_cap_for(method: str, path: str) -> int:
+def _route_path(path: str, root_path: str) -> str:
+    """The path the router sees: uvicorn --root-path (and ASGI mounts) put root_path in front of
+    scope["path"], and Starlette strips it the same way (a whole segment only) before matching.
+    Trailing slashes go too, so /users/ gets /users's cap, not the ceiling."""
+    if root_path and (path == root_path or path.startswith(root_path + "/")):
+        path = path[len(root_path):]
+    return path.rstrip("/")
+
+
+def _body_cap_for(method: str, path: str, root_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)
+    path = _route_path(path, root_path)
     for pattern, cap in _ROUTE_CAPS:
         if pattern.fullmatch(path):
             return min(cap, ceiling)
@@ -137,7 +153,7 @@ async def __call__(self, scope, receive, send):
         if scope["type"] != "http":
             await self.app(scope, receive, send)
             return
-        cap = _body_cap_for(scope.get("method", ""), scope.get("path", ""))
+        cap = _body_cap_for(scope.get("method", ""), scope.get("path", ""), scope.get("root_path", ""))
         for name, value in scope.get("headers", []):
             if name == b"content-length":
                 try:
@@ -416,9 +432,6 @@ def deliver_assisted(task_id: str, body: DeliverBody,
     return out
 
 
-_BLOB_CAP = 512 * 1024   # max bytes for one content-addressed chunk
-
-
 @app.post("/jobs/{job_id}/blobs/{blob_hash}")
 async def put_blob(job_id: str, blob_hash: str, request: Request,
                    x_pw_token: str | None = Header(default=None),
@@ -543,18 +556,24 @@ def _baseline_workers() -> int:
 # `baseline_skipped_busy`, so a degraded single-model compare under load is visible, not silent.
 _baseline_skipped = 0
 _baseline_skipped_lock = threading.Lock()
+# One log line for the first skip and then every Nth, not one per job: open signup means a busy or
+# hostile coordinator could otherwise write thousands a minute. The counter itself stays exact.
+_BASELINE_SKIP_LOG_EVERY = 100
 
 
 def _start_baseline(job_id: str, question: str) -> bool:
-    """Start the baseline if a slot is free. False (a log line + the skip counter) when all are busy."""
+    """Start the baseline if a slot is free. False (the skip counter + a throttled log line) 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):
         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)
+        if skipped == 1 or skipped % _BASELINE_SKIP_LOG_EVERY == 0:
+            print(f"[baseline] skipped for job {job_id[:8]}: all {_BASELINE_MAX} slots busy "
+                  f"(PW_BASELINE_WORKERS); {skipped} skipped since start (logged for the first and "
+                  f"every {_BASELINE_SKIP_LOG_EVERY}th; exact count on /metrics)", flush=True)
         return False
 
     def run() -> None:

tests/test_coordinator_limits.py

@@ -3,7 +3,8 @@
     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.
+    job that finds every slot busy is skipped with an exact /metrics counter and a throttled log
+    line (the first skip, then every Nth).
   • F13 — the hot lookups have SQLite indexes, created idempotently on every open.
 """
 import asyncio
@@ -291,7 +292,8 @@ 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", "/jobs/j1/blobs/abc") == capp._BLOB_BODY_CAP
+    assert capp._BLOB_BODY_CAP == capp._BLOB_CAP + 16 * 1024    # derived from the handler's cap
     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
@@ -317,5 +319,127 @@ def test_busy_slots_skip_the_baseline_with_a_log_line_and_a_counter(capp, client
     _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 "1 skipped since start" in out
+    assert "jobbusy2" not in out                 # throttled: only the first and every Nth are logged
     assert client.get("/metrics").json()["baseline_skipped_busy"] == 2
+
+
+def test_skip_log_is_throttled_but_the_counter_stays_exact(capp, client, monkeypatch, capsys):
+    """Low 5 (#50 review): one line for the first skip and then every Nth, not one per job."""
+    monkeypatch.setattr(capp, "_BASELINE_SKIP_LOG_EVERY", 3)
+    monkeypatch.setattr(capp, "_baseline_slots", threading.BoundedSemaphore(1))
+    assert capp._baseline_slots.acquire(blocking=False)          # every slot busy
+    try:
+        for i in range(1, 8):
+            assert capp._start_baseline(f"busy{i:04d}", "q") is False
+    finally:
+        capp._baseline_slots.release()
+    lines = [ln for ln in capsys.readouterr().out.splitlines() if ln.startswith("[baseline] skipped")]
+    assert [ln.split()[4].rstrip(":") for ln in lines] == ["busy0001", "busy0003", "busy0006"]
+    assert "6 skipped since start" in lines[-1]
+    assert client.get("/metrics").json()["baseline_skipped_busy"] == 7
+
+
+# ----------------------------------------------- F11 exact caps on every cap class (#50 review)
+def test_small_route_body_exactly_at_the_cap_passes_and_one_byte_more_is_413(client):
+    cap = 64 * 1024
+    hdr = {"content-type": "application/json"}
+    r = client.post("/users", content=_padded_json(b'{"handle": "alice"', cap), headers=hdr)
+    assert r.status_code == 200 and r.json()["user_secret"]
+    r = client.post("/users", content=_padded_json(b'{"handle": "bob"', cap + 1), headers=hdr)
+    assert r.status_code == 413
+    assert r.json()["detail"] == "request body too large"
+
+
+def test_result_body_exactly_at_the_ceiling_passes_and_one_byte_more_is_413(capp, client,
+                                                                            monkeypatch):
+    calls = _spy_complete(capp, monkeypatch)
+    hdr = {"content-type": "application/json", "X-Node-Secret": _node_secret(client)}
+    ceiling = 4 * 1024 * 1024
+    r = client.post("/tasks/t1/result", content=_padded_json(b'{"text": "x"', ceiling), headers=hdr)
+    assert r.status_code == 200, r.text
+    assert len(calls) == 1
+    r = client.post("/tasks/t1/result", content=_padded_json(b'{"text": "x"', ceiling + 1),
+                    headers=hdr)
+    assert r.status_code == 413
+    assert len(calls) == 1                       # the endpoint never ran
+
+
+def test_jobs_body_exactly_at_the_ceiling_passes_and_one_byte_more_is_413(capp, client,
+                                                                          monkeypatch):
+    calls = []
+    monkeypatch.setattr(capp.store, "create_job",
+                        lambda *a, **k: calls.append(a) or {"job_id": "j1", "status": "queued"})
+    usec = client.post("/users", json={"handle": "alice"}).json()["user_secret"]
+    hdr = {"content-type": "application/json", "X-User-Secret": usec}
+    ceiling = 4 * 1024 * 1024
+    r = client.post("/jobs", content=_padded_json(b'{"question": "q"', ceiling), headers=hdr)
+    assert r.status_code == 200, r.text
+    assert len(calls) == 1
+    r = client.post("/jobs", content=_padded_json(b'{"question": "q"', ceiling + 1), headers=hdr)
+    assert r.status_code == 413
+    assert len(calls) == 1
+
+
+def test_blob_route_cap_is_the_chunk_cap_plus_slack_not_the_ceiling(capp, client, monkeypatch):
+    """Medium 1 (#50 review): a blob upload over _BLOB_BODY_CAP is refused before auth, not let
+    through to 4 MiB; within it, put_blob's own exact _BLOB_CAP check still decides."""
+    stored = []
+    node_secret = _node_secret(client)
+    node_id = capp.store.node_for_secret(node_secret)
+    monkeypatch.setattr(capp.store, "assisted_claimant", lambda job_id: node_id)   # it claimed j1
+    monkeypatch.setattr(capp.store, "put_blob",
+                        lambda job_id, h, data: stored.append(len(data)) or {"ok": True})
+    good = {"content-type": "application/octet-stream", "X-Node-Secret": node_secret}
+    bad = {"content-type": "application/octet-stream", "X-Node-Secret": "nope"}
+    chunk, route_cap = capp._BLOB_CAP, capp._BLOB_BODY_CAP
+
+    r = client.post("/jobs/j1/blobs/h1", content=b"x" * chunk, headers=good)
+    assert r.status_code == 200, r.text
+    assert stored == [chunk]
+    r = client.post("/jobs/j1/blobs/h1", content=b"x" * (chunk + 1), headers=good)
+    assert r.status_code == 413 and r.json()["detail"] == "chunk too large"     # the handler's cap
+    r = client.post("/jobs/j1/blobs/h1", content=b"x" * route_cap, headers=bad)
+    assert r.status_code == 401                  # within the route cap: reaches auth
+    r = client.post("/jobs/j1/blobs/h1", content=b"x" * (route_cap + 1), headers=bad)
+    assert r.status_code == 413 and r.json()["detail"] == "request body too large"   # pre-auth
+    assert stored == [chunk]
+
+
+def test_get_with_a_body_over_64_kib_is_413(client):
+    cap = 64 * 1024
+    assert client.request("GET", "/status", content=b"x" * cap).status_code == 200
+    r = client.request("GET", "/status", content=b"x" * (cap + 1))
+    assert r.status_code == 413
+    assert r.json()["detail"] == "request body too large"
+
+
+def test_trailing_slash_gets_the_route_cap_not_the_ceiling(capp, client):
+    cap = 64 * 1024
+    hdr = {"content-type": "application/json"}
+    r = client.post("/users/", content=_padded_json(b'{"x": 1', cap + 1), headers=hdr,
+                    follow_redirects=False)
+    assert r.status_code == 413                  # was a 307 to /users under the 4 MiB ceiling
+    r = client.post("/users/", content=_padded_json(b'{"x": 1', cap), headers=hdr,
+                    follow_redirects=False)
+    assert r.status_code != 413
+    assert capp._body_cap_for("POST", "/tasks/t1/deliver/") == capp._DELIVER_BODY_CAP
+    assert capp._body_cap_for("POST", "/jobs/j1/blobs/abc/") == capp._BLOB_BODY_CAP
+
+
+def test_root_path_prefix_gets_the_route_cap(capp):
+    """Medium-Low 3 (#50 review): uvicorn --root-path puts the prefix in scope["path"] too; the
+    cap must be chosen from the app-relative path, as Starlette's router does."""
+    from fastapi.testclient import TestClient
+    client = TestClient(capp.app, root_path="/pw")
+    cap = 64 * 1024
+    hdr = {"content-type": "application/json"}
+    r = client.post("/pw/users", content=_padded_json(b'{"handle": "alice"', cap), headers=hdr)
+    assert r.status_code == 200, r.text          # routed to /users under the prefix
+    r = client.post("/pw/users", content=_padded_json(b'{"handle": "bob"', cap + 1), headers=hdr)
+    assert r.status_code == 413
+    small, ceiling = cap, 4 * 1024 * 1024
+    assert capp._body_cap_for("POST", "/pw/users", "/pw") == small
+    assert capp._body_cap_for("POST", "/users", "/pw") == small        # server left path unprefixed
+    assert capp._body_cap_for("POST", "/pw", "/pw") == ceiling
+    assert capp._body_cap_for("POST", "/pwx/users", "/pw") == ceiling  # not a whole-segment prefix