From 2f7c5f01ee7591a9588a89ed778408fa812f37ec Mon Sep 17 00:00:00 2001 From: nesquena-hermes Date: Fri, 26 Jun 2026 07:53:30 +0000 Subject: [PATCH] fix(#4973): replace per-request diagnostics Timer with a process-global watchdog thread A threading.Timer per RequestDiagnostics spawns one OS thread per request, held alive until finish() cancels it. Under sustained /api/sessions poll load this exhausts the per-process thread cap (RuntimeError: can't start new thread) and the server stops accepting connections. Replace with a single lazy-init watchdog daemon thread that scans pending diags every ~1s; public API unchanged. Caps timeout-tracking at 1 thread/process regardless of request rate. --- api/request_diagnostics.py | 84 ++++++++++++++++-- tests/test_issue4973_diagnostics_watchdog.py | 90 ++++++++++++++++++++ 2 files changed, 166 insertions(+), 8 deletions(-) create mode 100644 tests/test_issue4973_diagnostics_watchdog.py diff --git a/api/request_diagnostics.py b/api/request_diagnostics.py index 4c3ec719..2277a319 100644 --- a/api/request_diagnostics.py +++ b/api/request_diagnostics.py @@ -16,6 +16,73 @@ from typing import Any DEFAULT_SLOW_REQUEST_SECONDS = 5.0 MAX_STACK_FRAMES_PER_THREAD = 40 +# Process-global watchdog: a single daemon thread scans all in-flight +# RequestDiagnostics instances instead of each request spawning its own +# threading.Timer. A Timer-per-request spawns one OS thread per request, held +# alive until finish() cancels it; under sustained /api/sessions poll load that +# exhausts the per-process thread cap ("RuntimeError: can't start new thread") +# and the server stops accepting connections (#4973). One watchdog thread per +# process caps the timeout-tracking cost regardless of request rate. +_WATCHDOG_TICK_SECONDS = 1.0 +_watchdog_lock = threading.Lock() +_watchdog_pending: "dict[str, tuple[float, RequestDiagnostics]]" = {} +_watchdog_cv = threading.Condition(_watchdog_lock) +_watchdog_thread: "threading.Thread | None" = None + + +def _ensure_watchdog_running() -> None: + global _watchdog_thread + # Caller already holds _watchdog_lock. + if _watchdog_thread is not None and _watchdog_thread.is_alive(): + return + t = threading.Thread( + target=_watchdog_loop, + name="request-diagnostics-watchdog", + daemon=True, + ) + _watchdog_thread = t + t.start() + + +def _watchdog_loop() -> None: + # The loop body cannot raise while holding _watchdog_cv (the scan is pure + # dict ops; _on_timeout is fired outside the lock inside a try/except), so + # the watchdog thread never dies with entries pending. _ensure_watchdog_running + # therefore only needs to (re)spawn lazily on register, not supervise a crash. + while True: + fired: list[RequestDiagnostics] = [] + with _watchdog_cv: + # Sleep until the next tick, but wake early if work arrives/changes. + _watchdog_cv.wait(_WATCHDOG_TICK_SECONDS) + if _watchdog_pending: + now = time.monotonic() + expired = [ + rid for rid, (deadline, _diag) in _watchdog_pending.items() + if now >= deadline + ] + for rid in expired: + _deadline, diag = _watchdog_pending.pop(rid) + fired.append(diag) + # Fire _on_timeout outside the watchdog lock so a slow logger / stack + # snapshot can't stall the scan or block registering new requests. + for diag in fired: + try: + diag._on_timeout() + except Exception: # never let one bad record kill the watchdog + pass + + +def _watchdog_register(request_id: str, deadline: float, diag: "RequestDiagnostics") -> None: + with _watchdog_cv: + _watchdog_pending[request_id] = (deadline, diag) + _ensure_watchdog_running() + _watchdog_cv.notify() + + +def _watchdog_unregister(request_id: str) -> None: + with _watchdog_cv: + _watchdog_pending.pop(request_id, None) + def _slow_request_seconds() -> float: raw = os.getenv("HERMES_WEBUI_SLOW_REQUEST_SECONDS", "").strip() @@ -53,11 +120,12 @@ class RequestDiagnostics: self._current_stage_started = self.started_monotonic self._finished = False self._watchdog_logged = False - self._timer: threading.Timer | None = None if auto_start and self.timeout_seconds > 0: - self._timer = threading.Timer(self.timeout_seconds, self._on_timeout) - self._timer.daemon = True - self._timer.start() + _watchdog_register( + self.request_id, + self.started_monotonic + self.timeout_seconds, + self, + ) @classmethod def maybe_start( @@ -91,16 +159,16 @@ class RequestDiagnostics: self._current_stage_started = now def finish(self) -> None: - timer = None record = None with self._lock: if self._finished: return self._finished = True - timer = self._timer record = self._build_record_locked(include_stacks=False) - if timer is not None: - timer.cancel() + # Drop ourselves from the watchdog so the process-global scan never + # fires _on_timeout for a completed request (and the pending dict stays + # bounded by the number of in-flight requests). + _watchdog_unregister(self.request_id) if record and self.timeout_seconds > 0 and record["elapsed_ms"] >= self.timeout_seconds * 1000: self.logger.warning( "Slow WebUI request completed: %s", diff --git a/tests/test_issue4973_diagnostics_watchdog.py b/tests/test_issue4973_diagnostics_watchdog.py new file mode 100644 index 00000000..62b5a1b8 --- /dev/null +++ b/tests/test_issue4973_diagnostics_watchdog.py @@ -0,0 +1,90 @@ +"""Regression coverage for #4973 — RequestDiagnostics must not spawn one OS +thread per request. + +Before the fix, every ``RequestDiagnostics(auto_start=True)`` started its own +``threading.Timer`` (one OS thread per request, held alive until ``finish()``). +Under sustained ``/api/sessions`` poll load that exhausts the per-process thread +cap (``RuntimeError: can't start new thread``) and the server stops accepting +connections. The fix replaces the per-instance timer with a single +process-global watchdog daemon thread. +""" + +import logging +import threading +import time + +import api.request_diagnostics as rd +from api.request_diagnostics import RequestDiagnostics + + +def _watchdog_thread_count() -> int: + return sum( + 1 for t in threading.enumerate() + if t.name == "request-diagnostics-watchdog" + ) + + +def test_many_concurrent_diags_do_not_spawn_a_thread_each(): + """N in-flight auto_start diags must add at most ONE watchdog thread.""" + logger = logging.getLogger("test.issue4973.bounded") + baseline = _watchdog_thread_count() + + diags = [ + RequestDiagnostics( + "GET", "/api/sessions", logger=logger, timeout_seconds=30 + ) + for _ in range(200) + ] + try: + # The whole point: 200 concurrent diags, not 200 new threads. + added = _watchdog_thread_count() - baseline + assert added <= 1, f"expected <=1 watchdog thread, got {added} added" + # All 200 are registered as pending (none lost). + with rd._watchdog_cv: + pending = len(rd._watchdog_pending) + assert pending >= 200 + finally: + for d in diags: + d.finish() + + # finish() unregisters every diag so the pending dict stays bounded. + with rd._watchdog_cv: + remaining = sum( + 1 for _rid, (_dl, diag) in rd._watchdog_pending.items() + if diag in diags + ) + assert remaining == 0 + + +def test_watchdog_fires_on_timeout_for_a_slow_request(caplog): + """A request that never finishes before its deadline still gets logged by + the process-global watchdog (behavioral, not just structural).""" + logger = logging.getLogger("test.issue4973.fires") + # Tiny timeout so the watchdog (1s tick) fires quickly. + diag = RequestDiagnostics( + "GET", "/api/sessions", logger=logger, timeout_seconds=0.05 + ) + diag.stage("slow_stage") + with caplog.at_level(logging.WARNING, logger=logger.name): + deadline = time.monotonic() + 4.0 + while time.monotonic() < deadline: + if any("still running" in (r.getMessage() or "") for r in caplog.records): + break + time.sleep(0.1) + assert any("still running" in (r.getMessage() or "") for r in caplog.records), \ + "watchdog did not log the slow request within the window" + diag.finish() + + +def test_finish_before_deadline_prevents_watchdog_log(caplog): + """A request that finishes fast must NOT be logged by the watchdog.""" + logger = logging.getLogger("test.issue4973.fast") + diag = RequestDiagnostics( + "GET", "/api/sessions", logger=logger, timeout_seconds=0.05 + ) + diag.finish() # completes immediately, well under the deadline + with caplog.at_level(logging.WARNING, logger=logger.name): + time.sleep(1.5) # give the watchdog a couple of ticks + assert not any( + "still running" in (r.getMessage() or "") for r in caplog.records + ), "watchdog logged a request that already finished"