fix(#828): handle broker connections concurrently with per-conn deadlines + bounded concurrency

docs/scratchpads/828-lease-broker.md

@@ -76,6 +76,9 @@ Verified before code on session start; all exact SHA-256 values matched:
 - Post-remediation evidence: Python unit suite `10/10`, focused real-socket acceptance `35/35`, full root suite `43/43` Turbo tasks, broker coverage `90%` (`395` statements, `38` missed), lint `23/23`, typecheck `42/42`, format check and `git diff --check` GREEN.
 - Independent Codex review of remediation commit `d05465e54736c4966294c4af8fbd6a4ad8fe81aa`: APPROVE, confidence `0.94`, zero findings. Reviewer sandbox could not allocate temp directories; unrestricted parent test evidence above is canonical.
 
+- Remediation session: terra review comment `18072` reproduced a SERIAL-ACCEPT DoS; scope is RED regressions plus bounded concurrent connection handling on PR #836, preserving all existing broker security properties.
+- RED evidence against reviewed daemon: four silent peers delayed registration `3920 ms` beyond the `1500 ms` bound; 16 silent peers were not reaped within `2500 ms`. The first bounded implementation then exposed slot exhaustion by rejecting the valid queued caller with `EPIPE`; admission was corrected to wait for a reclaimed bounded slot. GREEN evidence: queued-peer test `211 ms`; strengthened cap/reap/reclaim test `1118 ms`; complete real-socket acceptance `37/37` and Python persistence suite `10/10`.
+
 ## Coordinator handoff requirements
 
 1. Mandatory Opus-SECREV on the exact PR head; no GPT/terra substitute.

packages/mosaic/framework/tools/lease-broker/daemon.py

@@ -6,6 +6,7 @@ from __future__ import annotations
 import argparse
 import copy
 import errno
+from concurrent.futures import ThreadPoolExecutor
 import json
 import os
 import secrets
@@ -14,6 +15,7 @@ import socket
 import stat
 import struct
 import sys
+import threading
 import time
 from pathlib import Path
 from typing import Final
@@ -21,6 +23,7 @@ from typing import Final
 MAX_FRAME: Final = 64 * 1024
 MAX_STATE: Final = 4 * 1024 * 1024
 MAX_PENDING_TOKENS: Final = 256
+MAX_IN_FLIGHT_CONNECTIONS: Final = 16
 STATE_VERSION: Final = 1
 CONNECTION_DEADLINE_SECONDS: Final = 1.0
 HEX_256_LENGTH: Final = 64
@@ -410,12 +413,51 @@ def read_frame(connection: socket.socket, deadline: float) -> dict[str, object]:
     return value
 
 
+def handle_connection(
+    connection: socket.socket,
+    broker: Broker,
+    broker_lock: threading.Lock,
+) -> None:
+    with connection:
+        deadline = time.monotonic() + CONNECTION_DEADLINE_SECONDS
+        try:
+            raw = connection.getsockopt(socket.SOL_SOCKET, socket.SO_PEERCRED, 12)
+            peer = struct.unpack("3i", raw)
+            request = read_frame(connection, deadline)
+        except BrokerFailure as exc:
+            reply = {"ok": False, "code": exc.code}
+        except OSError:
+            return
+        else:
+            try:
+                with broker_lock:
+                    reply = broker.handle(peer, request)
+            except BrokerFailure as exc:
+                reply = {"ok": False, "code": exc.code}
+        try:
+            remaining = deadline - time.monotonic()
+            if remaining <= 0:
+                return
+            connection.settimeout(remaining)
+            connection.sendall((json.dumps(reply, separators=(",", ":")) + "\n").encode())
+        except OSError:
+            return
+
+
 def serve(socket_path: Path, state_path: Path) -> None:
     secure_parent(socket_path)
     if socket_path.exists() or socket_path.is_symlink():
         raise BrokerFailure("SOCKET_ALREADY_EXISTS")
     store = StateStore(state_path)
     broker = Broker(store)
+    broker_lock = threading.Lock()
+    slots = threading.BoundedSemaphore(MAX_IN_FLIGHT_CONNECTIONS)
+    fatal_lock = threading.Lock()
+    fatal_errors: list[Exception] = []
+    executor = ThreadPoolExecutor(
+        max_workers=MAX_IN_FLIGHT_CONNECTIONS,
+        thread_name_prefix="mosaic-lease-broker",
+    )
     server = socket.socket(socket.AF_UNIX, socket.SOCK_STREAM)
     server.bind(str(socket_path))
     os.chmod(socket_path, 0o600)
@@ -427,42 +469,54 @@ def serve(socket_path: Path, state_path: Path) -> None:
         stopping = True
         server.close()
 
+    def process_connection(connection: socket.socket) -> None:
+        try:
+            handle_connection(connection, broker, broker_lock)
+        except Exception as exc:
+            with fatal_lock:
+                if not fatal_errors:
+                    fatal_errors.append(exc)
+        finally:
+            slots.release()
+
+    def fatal_error() -> Exception | None:
+        with fatal_lock:
+            return fatal_errors[0] if fatal_errors else None
+
     signal.signal(signal.SIGTERM, stop)
     signal.signal(signal.SIGINT, stop)
-    server.listen(16)
+    server.listen(MAX_IN_FLIGHT_CONNECTIONS)
+    server.settimeout(0.1)
     print("READY", flush=True)
     try:
         while not stopping:
+            failure = fatal_error()
+            if failure is not None:
+                raise failure
+            if not slots.acquire(timeout=0.1):
+                continue
             try:
                 connection, _ = server.accept()
+            except socket.timeout:
+                slots.release()
+                continue
             except OSError:
+                slots.release()
+                failure = fatal_error()
+                if failure is not None:
+                    raise failure
                 if stopping:
                     break
                 raise
-            with connection:
-                deadline = time.monotonic() + CONNECTION_DEADLINE_SECONDS
             try:
-                    raw = connection.getsockopt(socket.SOL_SOCKET, socket.SO_PEERCRED, 12)
+                executor.submit(process_connection, connection)
-                    peer = struct.unpack("3i", raw)
+            except Exception:
-                    request = read_frame(connection, deadline)
+                slots.release()
-                except BrokerFailure as exc:
+                connection.close()
-                    reply = {"ok": False, "code": exc.code}
+                raise
-                except OSError:
-                    continue
-                else:
-                    try:
-                        reply = broker.handle(peer, request)
-                    except BrokerFailure as exc:
-                        reply = {"ok": False, "code": exc.code}
-                try:
-                    remaining = deadline - time.monotonic()
-                    if remaining <= 0:
-                        continue
-                    connection.settimeout(remaining)
-                    connection.sendall((json.dumps(reply, separators=(",", ":")) + "\n").encode())
-                except OSError:
-                    continue
     finally:
+        server.close()
+        executor.shutdown(wait=True)
         try:
             current = socket_path.stat()
             if (current.st_dev, current.st_ino) == owned: