From bbf351fc69ceee0645bd34cf5e28ec8306e6a14a Mon Sep 17 00:00:00 2001 From: Ahmed Janabi Date: Sun, 11 Oct 2026 09:16:44 +0000 Subject: [PATCH] fix(net): per-route coordinator body caps and counted baseline skips Follow-ups from the PR #49 review: - Per-route request-body caps under the PW_MAX_BODY_BYTES ceiling (4 MiB default): /tasks/{id}/deliver is capped at 2,416,384 bytes (a 200k-char deliverable fully JSON-escaped, plus signature fields), field-capped small bodies (/users, /nodes/register, heartbeat, progress, accept, feedback, rate, /admin/enroll) and non-POST requests at 64 KiB. The free-form routes (/result dicts, /jobs, blob uploads) keep the ceiling because they have no documented smaller bound. A lower ceiling still lowers every cap. - A baseline skipped because every PW_BASELINE_WORKERS slot is busy is now counted (baseline_skipped_busy on GET /metrics) and the log line names the running total. - Tests: an exact-cap /deliver body passes and cap+1 is 413, small-route caps (declared and chunked), the route-to-cap mapping, and the skip counter. README documents PW_MAX_BODY_BYTES and PW_BASELINE_WORKERS. Co-Authored-By: Claude Opus 5.5 --- README.md | 13 +++ passiveworkers/net/baseline.py | 4 +- passiveworkers/net/coordinator_app.py | 55 ++++++++++--- tests/test_coordinator_limits.py | 110 +++++++++++++++++++++++++- 4 files changed, 169 insertions(+), 13 deletions(-) diff --git a/README.md b/README.md index 7977683..fe7e4be 100644 --- a/README.md +++ b/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 — diff --git a/passiveworkers/net/baseline.py b/passiveworkers/net/baseline.py index 9b389f7..d4d588b 100644 --- a/passiveworkers/net/baseline.py +++ b/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 diff --git a/passiveworkers/net/coordinator_app.py b/passiveworkers/net/coordinator_app.py index e54bbb6..1f35845 100644 --- a/passiveworkers/net/coordinator_app.py +++ b/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") diff --git a/tests/test_coordinator_limits.py b/tests/test_coordinator_limits.py index 8d8128d..643ae14 100644 --- a/tests/test_coordinator_limits.py +++ b/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