From d505b4554508a86f692c2d3c5adbf0b73fb0ee5a Mon Sep 17 00:00:00 2001 From: ai-ag2026 <261867348+ai-ag2026@users.noreply.github.com> Date: Wed, 8 Jul 2026 20:24:42 +0200 Subject: [PATCH] fix(bg): serialize daemon-thread start under a lifecycle lock start_drain_thread and start_session_channel_reaper checked is_alive() and then created + started the daemon thread without holding a lock, a check-then-act race: two concurrent callers could both see "not alive" and each spawn a thread. The loser's thread was never stored in the module global and ran forever, un-joinable by the matching stop_*. Both check-then-start sequences now run under a dedicated _THREAD_LIFECYCLE_LOCK (kept separate from the purpose-bound SESSION_CHANNELS_LOCK / _EMIT_COALESCE_LOCK), so exactly one thread is ever created. Co-Authored-By: Claude Opus 4.8 (1M context) --- api/background_process.py | 50 +++++++++------ tests/test_bg_thread_start_lock.py | 98 ++++++++++++++++++++++++++++++ 2 files changed, 128 insertions(+), 20 deletions(-) create mode 100644 tests/test_bg_thread_start_lock.py diff --git a/api/background_process.py b/api/background_process.py index 01fc8128..7147291d 100644 --- a/api/background_process.py +++ b/api/background_process.py @@ -57,6 +57,14 @@ _REAPER_THREAD: Optional[threading.Thread] = None _REAPER_STOP = threading.Event() _REAPER_INTERVAL_SECS = 60.0 +# Serializes the check-then-start of the module's daemon threads +# (``start_drain_thread`` / ``start_session_channel_reaper``). Without it two +# concurrent callers can both observe ``is_alive() == False`` and each spawn a +# thread; the loser's thread is never referenced by the module global and runs +# forever, un-joinable. A dedicated lock (not the purpose-bound +# ``SESSION_CHANNELS_LOCK`` / ``_EMIT_COALESCE_LOCK``) keeps this narrow. +_THREAD_LIFECYCLE_LOCK = threading.Lock() + # T3: per-session coalesce gate for the public bg_task_complete SSE emit. # The server-side wakeup path remains immediate; only the browser-observation # frame is throttled so a burst of background task completions does not flood an @@ -413,16 +421,17 @@ def _reaper_loop() -> None: def start_session_channel_reaper() -> bool: """Start the SessionChannel reaper thread. Idempotent; returns True on first start.""" global _REAPER_THREAD - if _REAPER_THREAD is not None and _REAPER_THREAD.is_alive(): - return False - _REAPER_STOP.clear() - _REAPER_THREAD = threading.Thread( - target=_reaper_loop, - name="hermes-webui-session-channel-reaper", - daemon=True, - ) - _REAPER_THREAD.start() - return True + with _THREAD_LIFECYCLE_LOCK: + if _REAPER_THREAD is not None and _REAPER_THREAD.is_alive(): + return False + _REAPER_STOP.clear() + _REAPER_THREAD = threading.Thread( + target=_reaper_loop, + name="hermes-webui-session-channel-reaper", + daemon=True, + ) + _REAPER_THREAD.start() + return True def stop_session_channel_reaper(timeout: float = 2.0) -> None: @@ -1372,16 +1381,17 @@ def forget_bg_task_completion_dedup(session_id: str) -> None: def start_drain_thread() -> bool: """Start the background drain thread idempotently. Returns True on first start.""" global _DRAIN_THREAD - if _DRAIN_THREAD is not None and _DRAIN_THREAD.is_alive(): - return False - _DRAIN_STOP.clear() - _DRAIN_THREAD = threading.Thread( - target=_drain_loop, - name="hermes-webui-bg-task-complete-drain", - daemon=True, - ) - _DRAIN_THREAD.start() - return True + with _THREAD_LIFECYCLE_LOCK: + if _DRAIN_THREAD is not None and _DRAIN_THREAD.is_alive(): + return False + _DRAIN_STOP.clear() + _DRAIN_THREAD = threading.Thread( + target=_drain_loop, + name="hermes-webui-bg-task-complete-drain", + daemon=True, + ) + _DRAIN_THREAD.start() + return True def stop_drain_thread(timeout: float = 2.0) -> None: diff --git a/tests/test_bg_thread_start_lock.py b/tests/test_bg_thread_start_lock.py new file mode 100644 index 00000000..f96d84f8 --- /dev/null +++ b/tests/test_bg_thread_start_lock.py @@ -0,0 +1,98 @@ +"""Regression test for the thread-start check-then-act race in background_process. + +``start_drain_thread`` and ``start_session_channel_reaper`` checked +``is_alive()`` and then created + started the daemon thread WITHOUT a lock, so +two concurrent callers could both observe "not alive" and each spawn a thread. +The loser's thread was never stored in the module global and ran forever, +un-joinable. Both check-then-start sequences are now serialized under +``_THREAD_LIFECYCLE_LOCK``, so exactly one thread is ever created. + +The test drives N callers through a barrier so they hit the check-and-start +simultaneously, and asserts exactly one caller reports "started" (returns True) +and exactly one live thread exists. +""" + +from __future__ import annotations + +import threading + +import pytest + + +# (start_fn, stop_fn, loop_attr, thread_attr, thread_name_fragment) +_CASES = [ + ( + "start_drain_thread", + "stop_drain_thread", + "_drain_loop", + "_DRAIN_THREAD", + "bg-task-complete-drain", + ), + ( + "start_session_channel_reaper", + "stop_session_channel_reaper", + "_reaper_loop", + "_REAPER_THREAD", + "session-channel-reaper", + ), +] + + +@pytest.mark.parametrize( + "start_name,stop_name,loop_attr,thread_attr,name_frag", _CASES +) +def test_concurrent_start_creates_exactly_one_thread( + monkeypatch, start_name, stop_name, loop_attr, thread_attr, name_frag +): + from api import background_process as bp + + release = threading.Event() + + def _blocking_loop() -> None: + # Keep the started thread alive during the assertions so a genuinely + # started thread reads as is_alive()==True deterministically. + release.wait(5.0) + + monkeypatch.setattr(bp, loop_attr, _blocking_loop, raising=True) + # Fresh slate: no pre-existing thread reference. + monkeypatch.setattr(bp, thread_attr, None, raising=True) + + start_fn = getattr(bp, start_name) + n = 12 + barrier = threading.Barrier(n) + results: list[bool] = [] + results_lock = threading.Lock() + + def _caller() -> None: + barrier.wait() + started = start_fn() + with results_lock: + results.append(started) + + callers = [threading.Thread(target=_caller) for _ in range(n)] + try: + for c in callers: + c.start() + for c in callers: + c.join(timeout=5.0) + + # Exactly one caller won the create; the rest saw a live thread. + assert sum(1 for r in results if r) == 1, ( + f"expected exactly one start to win, got {results}" + ) + # Exactly one live daemon thread carrying this loop's name. + live = [ + t + for t in threading.enumerate() + if name_frag in t.name and t.is_alive() + ] + assert len(live) == 1, f"expected 1 live {name_frag} thread, got {len(live)}" + # The module global points at that one live thread. + assert getattr(bp, thread_attr) is live[0] + finally: + release.set() + getattr(bp, stop_name)() + + +if __name__ == "__main__": # pragma: no cover - manual invocation + raise SystemExit(pytest.main([__file__, "-v"]))