From b072a7bd429715c82a086f6c36705b10a17c7103 Mon Sep 17 00:00:00 2001 From: ms-lead-reviewer Date: Fri, 17 Jul 2026 21:57:05 -0500 Subject: [PATCH] fix(#828): handle broker connections concurrently with per-conn deadlines + bounded concurrency --- docs/scratchpads/828-lease-broker.md | 3 + .../framework/tools/lease-broker/daemon.py | 102 +++++++++++++----- 2 files changed, 81 insertions(+), 24 deletions(-) diff --git a/docs/scratchpads/828-lease-broker.md b/docs/scratchpads/828-lease-broker.md index fbe0200..6c66923 100644 --- a/docs/scratchpads/828-lease-broker.md +++ b/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. diff --git a/packages/mosaic/framework/tools/lease-broker/daemon.py b/packages/mosaic/framework/tools/lease-broker/daemon.py index d204090..dfcddf1 100644 --- a/packages/mosaic/framework/tools/lease-broker/daemon.py +++ b/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) - peer = struct.unpack("3i", raw) - request = read_frame(connection, deadline) - except BrokerFailure as exc: - reply = {"ok": False, "code": exc.code} - 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 + try: + executor.submit(process_connection, connection) + except Exception: + slots.release() + connection.close() + raise finally: + server.close() + executor.shutdown(wait=True) try: current = socket_path.stat() if (current.st_dev, current.st_ino) == owned: