fix(net): cap coordinator request bodies, bound baseline threads, index hot lookups, throttle rate-limit sweep

passiveworkers/net/coordinator_app.py

@@ -23,6 +23,11 @@
   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.
+
 Run:  PW_TOKEN=… python -m passiveworkers.net.coordinator_app
 """
 
@@ -34,7 +39,7 @@
 import threading
 
 from fastapi import FastAPI, Header, HTTPException, Request, Response
-from fastapi.responses import HTMLResponse
+from fastapi.responses import HTMLResponse, JSONResponse
 from pydantic import BaseModel, Field
 
 from passiveworkers.net import geoip
@@ -70,7 +75,81 @@ def _startup_guard() -> None:
     print("[coordinator] PW_TRUST_XFF enabled — rate-limit keys trust X-Forwarded-For. Ensure your "
           "reverse proxy sets AND sanitizes that header (drops client-supplied values), or attackers "
           "can rotate it to defeat per-client limits.", flush=True)
+
+
+_MAX_BODY_DEFAULT = 4 * 1024 * 1024   # see the module docstring for why 4 MiB
+
+
+def _max_body_bytes() -> int:
+    """The request-body cap 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))
+    except (TypeError, ValueError):
+        return _MAX_BODY_DEFAULT
+    return cap if cap > 0 else _MAX_BODY_DEFAULT
+
+
+class _BodyTooLarge(HTTPException):
+    def __init__(self) -> None:
+        super().__init__(status_code=413, detail="request body too large")
+
+
+class BodySizeLimit:
+    """Pure-ASGI request-body cap (F11 review). A declared Content-Length over the cap is refused
+    before the app runs at all; otherwise the body is counted as it streams (chunked or lying
+    clients), and the first message that crosses the cap raises 413 — the app never buffers more
+    than cap + one server-sized chunk. Raised as an HTTPException so FastAPI's body parser re-raises
+    it unchanged (a generic error there would become a 400)."""
+
+    def __init__(self, app) -> None:
+        self.app = app
+
+    async def __call__(self, scope, receive, send):
+        if scope["type"] != "http":
+            await self.app(scope, receive, send)
+            return
+        cap = _max_body_bytes()
+        for name, value in scope.get("headers", []):
+            if name == b"content-length":
+                try:
+                    declared = int(value)
+                except ValueError:
+                    break          # the server rejects malformed lengths; the stream count still applies
+                if declared > cap:
+                    await JSONResponse({"detail": "request body too large"}, status_code=413)(
+                        scope, receive, send)
+                    return
+                break
+        seen = 0
+        started = False
+
+        async def capped_receive():
+            nonlocal seen
+            message = await receive()
+            if message["type"] == "http.request":
+                seen += len(message.get("body", b""))
+                if seen > cap:
+                    raise _BodyTooLarge()
+            return message
+
+        async def tracking_send(message):
+            nonlocal started
+            if message["type"] == "http.response.start":
+                started = True
+            await send(message)
+
+        try:
+            await self.app(scope, capped_receive, tracking_send)
+        except _BodyTooLarge:
+            if started:
+                raise
+            await JSONResponse({"detail": "request body too large"}, status_code=413)(
+                scope, receive, send)
+
+
 app = FastAPI(title="Passive Workers — Council Coordinator")
+app.add_middleware(BodySizeLimit)
 store = Store()
 _rl = RateLimiter()   # per-process sliding-window rate limiter (D36)
 
@@ -419,6 +498,44 @@ def _baseline_async(job_id: str, question: str) -> None:
         print(f"[baseline] store failed for job {job_id[:8]}: {type(e).__name__}: {e}", flush=True)
 
 
+def _baseline_workers() -> int:
+    try:
+        return max(1, int(os.environ.get("PW_BASELINE_WORKERS", "4")))
+    except (TypeError, ValueError):
+        return 4
+
+
+# F12 review: at most `_BASELINE_MAX` baseline threads at once instead of one per job. Each can live
+# up to 900 s (waiting for a local job to settle), so a job submitted while every slot is busy simply
+# gets no baseline — never a queued backlog. Daemon threads, not a ThreadPoolExecutor: executor
+# 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)
+
+
+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."""
+    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)
+        return False
+
+    def run() -> None:
+        try:
+            _baseline_async(job_id, question)
+        except Exception as e:  # best-effort, like _baseline_async's own store failure
+            print(f"[baseline] failed for job {job_id[:8]}: {type(e).__name__}: {e}", flush=True)
+        finally:
+            slots.release()
+
+    try:
+        threading.Thread(target=run, daemon=True, name="pw-baseline").start()
+    except RuntimeError:   # can't start a thread (resource limit / interpreter shutdown)
+        slots.release()
+        return False
+    return True
+
+
 @app.get("/job-types")
 def job_types():
     """The marketplace catalog — what kinds of work computers can request here."""
@@ -442,8 +559,7 @@ def submit_job(body: JobBody, x_user_secret: str | None = Header(default=None)):
     if out.get("status") == "pending_answers" and not task_behavior(body.type).sharded:
         # the honest single-model compare only makes sense for answer/report jobs —
         # a one-shot model can't process a sharded item batch (shard_map/download_extract/code_generation)
-        threading.Thread(target=_baseline_async, args=(out["job_id"], body.question),
-                         daemon=True, name="pw-baseline").start()
+        _start_baseline(out["job_id"], body.question)
     return out
 
 

passiveworkers/net/ratelimit.py

@@ -9,24 +9,29 @@
 Deliberately tiny + dependency-free (one coordinator process today; swap for Redis behind the same
 `allow()` seam when horizontally scaled). Per-key sliding window: at most `limit` events per
 `window_s`. `limit <= 0` disables that key (operator escape hatch). Memory is bounded — a denied
-event is NOT recorded (so a key's deque never exceeds `limit`), and idle keys are swept when the
-table grows past `max_keys`.
+event is NOT recorded (so a key's deque never exceeds `limit`), and once the table grows past
+`max_keys` it is swept — at most once per `_SWEEP_EVERY_S` (F16 review: an unthrottled O(n) sweep
+ran on EVERY call while the table stayed over the cap) — dropping idle keys first, then the
+least-recently-hit keys until it is back at `max_keys`.
 """
 
 from __future__ import annotations
 
+import heapq
 import threading
 import time
 from collections import deque
 
 _SWEEP_HORIZON_S = 3600.0   # keys idle longer than this are dropped on overflow (>> any window)
+_SWEEP_EVERY_S = 10.0       # at most one sweep per this many seconds, however far over the cap
 
 
 class RateLimiter:
     def __init__(self, max_keys: int = 50_000):
         self._hits: dict[str, deque] = {}
         self._lock = threading.Lock()
         self._max_keys = max_keys
+        self._last_sweep = float("-inf")
 
     def allow(self, key: str, limit: int, window_s: float, now: float | None = None) -> bool:
         """True if an event for `key` is within `limit` per `window_s` (and records it); False if it
@@ -46,13 +51,21 @@ def allow(self, key: str, limit: int, window_s: float, now: float | None = None)
             if len(dq) >= limit:
                 return False              # denied — do not record, so the deque stays bounded
             dq.append(t)
-            if len(self._hits) > self._max_keys:
+            # abs(): a clock that stepped backwards must not suppress sweeping until it catches up
+            if len(self._hits) > self._max_keys and abs(t - self._last_sweep) >= _SWEEP_EVERY_S:
+                self._last_sweep = t
                 self._sweep(t)
             return True
 
     def _sweep(self, now: float) -> None:
         """Drop keys that are empty or idle past the horizon (>> any real window, so a swept key's
-        limit window has certainly elapsed) — bounds memory under churny key spaces."""
+        limit window has certainly elapsed). If that still leaves the table over `max_keys` (key
+        rotation keeps every key fresh), evict the least-recently-hit keys down to the cap: memory
+        stays bounded, and an evicted key only loses its own history, so it is never MORE limited."""
         horizon = now - _SWEEP_HORIZON_S
         for k in [k for k, d in self._hits.items() if not d or d[-1] < horizon]:
             del self._hits[k]
+        excess = len(self._hits) - self._max_keys
+        if excess > 0:
+            for k in heapq.nsmallest(excess, self._hits, key=lambda k: self._hits[k][-1]):
+                del self._hits[k]

passiveworkers/net/store.py

@@ -152,4 +152,17 @@ def _init_schema(self) -> None:
             self.conn.execute("ALTER TABLE nodes ADD COLUMN ok_count INTEGER DEFAULT 0")
         if "fail_count" not in ncols:
             self.conn.execute("ALTER TABLE nodes ADD COLUMN fail_count INTEGER DEFAULT 0")
+        # F13 review: the hot lookups (secret auth on every heartbeat/poll, a job's tasks on every
+        # job_view, a node's queue, an asker's jobs, the reaper's status scans) were full-table scans
+        # under the global lock. IF NOT EXISTS → idempotent on every open, existing DBs included.
+        self.conn.executescript(
+            """
+            CREATE INDEX IF NOT EXISTS idx_nodes_secret_hash ON nodes(secret_hash);
+            CREATE INDEX IF NOT EXISTS idx_users_secret_hash ON users(secret_hash);
+            CREATE INDEX IF NOT EXISTS idx_tasks_job_id ON tasks(job_id);
+            CREATE INDEX IF NOT EXISTS idx_tasks_node_status ON tasks(node_id, status);
+            CREATE INDEX IF NOT EXISTS idx_jobs_asker_created ON jobs(asker, created);
+            CREATE INDEX IF NOT EXISTS idx_jobs_status ON jobs(status);
+            """
+        )
         self.conn.commit()