From acc1f27df37932ff81d943d7ed4330e9f0abab91 Mon Sep 17 00:00:00 2001 From: Ahmed Janabi Date: Sun, 11 Oct 2026 09:27:25 +0000 Subject: [PATCH] fix(net): blob route cap, root_path/trailing-slash cap matching, throttled baseline-skip log Follow-ups from the PR #50 review: - Blob uploads get their own route cap, _BLOB_BODY_CAP = _BLOB_CAP + 16 KiB slack, derived from the handler's chunk constant (now defined with the other caps), instead of the 4 MiB ceiling. - _body_cap_for matches the app-relative path: a whole-segment root_path prefix (uvicorn --root-path puts it in scope["path"]) and trailing slashes are stripped first, as Starlette's router does, so /pw/users and /users/ get /users's 64 KiB cap instead of the ceiling. - The "baseline skipped" log line is printed for the first skip and every 100th; the /metrics counter stays exact. - README cap wording fixed (blob uploads are no longer "at the ceiling"). - Tests: exact-cap passes and cap+1 is 413 for /users, /result, /jobs and blobs; GET with a >64 KiB body is 413; trailing-slash and root_path-prefixed paths get the route cap; the log throttle keeps an exact count. Co-Authored-By: Claude Opus 5.5 --- README.md | 8 +- passiveworkers/net/coordinator_app.py | 45 ++++++--- tests/test_coordinator_limits.py | 130 +++++++++++++++++++++++++- 3 files changed, 164 insertions(+), 19 deletions(-) diff --git a/README.md b/README.md index fe7e4be..b2b698e 100644 --- a/README.md +++ b/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. diff --git a/passiveworkers/net/coordinator_app.py b/passiveworkers/net/coordinator_app.py index 1f35845..09e653f 100644 --- a/passiveworkers/net/coordinator_app.py +++ b/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: diff --git a/tests/test_coordinator_limits.py b/tests/test_coordinator_limits.py index 643ae14..e0faeb1 100644 --- a/tests/test_coordinator_limits.py +++ b/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