From aff415b653a80f7070f0710eb9ae6c7cd74c425c Mon Sep 17 00:00:00 2001 From: Josh Tsai <128559392+bounce12340@users.noreply.github.com> Date: Thu, 23 Jul 2026 17:25:12 +0800 Subject: [PATCH] fix(gateway): close watchdog shutdown race against final-strike exit - Re-check stop_event after a missed probe (before the strike increment) and again on entering the final-strike branch (before the critical log, dump, and hard exit), so a normal stop() landing between the last timeout check and the exit path can no longer be misclassified as a freeze and trigger a supervisor restart. - Deterministic boundary tests pin both re-checks independently (mutation-verified: removing either check turns its own test red); frozen-loop semantics are unchanged. Addresses the shutdown-race review on #69164. Co-Authored-By: Claude Fable 5 --- gateway/shutdown_watchdog.py | 4 + tests/gateway/test_loop_liveness_watchdog.py | 98 ++++++++++++++++++++ 2 files changed, 102 insertions(+) diff --git a/gateway/shutdown_watchdog.py b/gateway/shutdown_watchdog.py index 5dbc4a9b81bf..116c7fe2ba10 100644 --- a/gateway/shutdown_watchdog.py +++ b/gateway/shutdown_watchdog.py @@ -186,10 +186,14 @@ def start_loop_liveness_watchdog( strikes = 0 continue + if stop_event.is_set(): + return strikes += 1 if strikes < strikes_limit: continue + if stop_event.is_set(): + return try: logger.critical( "Gateway event loop missed %d consecutive liveness probes; " diff --git a/tests/gateway/test_loop_liveness_watchdog.py b/tests/gateway/test_loop_liveness_watchdog.py index 1add915341aa..91bc21c33572 100644 --- a/tests/gateway/test_loop_liveness_watchdog.py +++ b/tests/gateway/test_loop_liveness_watchdog.py @@ -71,6 +71,104 @@ def test_loop_liveness_watchdog_exits_after_consecutive_misses(): assert exit_codes == [75] +def test_loop_liveness_watchdog_stop_during_final_miss_disarms_hard_exit(): + loop = MagicMock(spec=asyncio.AbstractEventLoop) + probe_scheduled = threading.Event() + release_probe = threading.Event() + probe_event_ref = {} + handle_ref = {} + exit_codes = [] + + class FinalStrikeLimit: + def __gt__(self, _strikes: int) -> bool: + # If strike evaluation is reached, keep recheck #2 from masking a + # missing post-probe recheck #1 in this boundary test. + handle_ref["handle"]._stop_event.clear() + return False + + def hold_scheduled_probe(callback) -> None: + probe_event_ref["event"] = callback.__self__ + probe_scheduled.set() + assert release_probe.wait(timeout=2.0) + + loop.call_soon_threadsafe.side_effect = hold_scheduled_probe + with ( + patch( + "gateway.shutdown_watchdog._positive_int_env", + return_value=FinalStrikeLimit(), + ), + patch("gateway.shutdown_watchdog.logger.critical") as critical, + patch("gateway.shutdown_watchdog.faulthandler.dump_traceback") as dump, + patch("gateway.shutdown_watchdog.os._exit", side_effect=exit_codes.append), + ): + handle = start_loop_liveness_watchdog( + loop, probe_interval=0.01, probe_timeout=0.01, max_strikes=1 + ) + assert handle is not None + handle_ref["handle"] = handle + assert probe_scheduled.wait(timeout=2.0), "watchdog did not schedule a probe" + + def stop_during_miss() -> bool: + handle.stop() + return False + + probe_event_ref["event"].is_set = stop_during_miss + release_probe.set() + handle.join(timeout=1.0) + + assert not handle.is_alive() + assert exit_codes == [] + critical.assert_not_called() + dump.assert_not_called() + + +def test_loop_liveness_watchdog_stop_after_first_recheck_skips_final_actions(): + loop = MagicMock(spec=asyncio.AbstractEventLoop) + probe_scheduled = threading.Event() + release_probe = threading.Event() + + def hold_scheduled_probe(callback) -> None: + probe_scheduled.set() + assert release_probe.wait(timeout=2.0) + + loop.call_soon_threadsafe.side_effect = hold_scheduled_probe + with ( + patch("gateway.shutdown_watchdog.logger.critical") as critical, + patch("gateway.shutdown_watchdog.faulthandler.dump_traceback") as dump, + patch("gateway.shutdown_watchdog.os._exit") as hard_exit, + ): + handle = start_loop_liveness_watchdog( + loop, probe_interval=0.01, probe_timeout=0.01, max_strikes=1 + ) + assert handle is not None + assert probe_scheduled.wait(timeout=2.0), "watchdog did not schedule a probe" + + original_is_set = handle._stop_event.is_set + is_set_calls = 0 + + def stop_on_final_recheck() -> bool: + nonlocal is_set_calls + is_set_calls += 1 + # With the forced immediate timeout: _wait_for_probe is call 1, + # recheck #1 is call 2, and recheck #2 is call 3. + if is_set_calls == 3: + handle.stop() + return original_is_set() + + handle._stop_event.is_set = stop_on_final_recheck + with patch( + "gateway.shutdown_watchdog.time.monotonic", side_effect=[0.0, 1.0] + ): + release_probe.set() + handle.join(timeout=1.0) + + assert is_set_calls == 3 + assert not handle.is_alive() + critical.assert_not_called() + dump.assert_not_called() + hard_exit.assert_not_called() + + def test_loop_liveness_watchdog_recovery_resets_strikes(): loop = MagicMock(spec=asyncio.AbstractEventLoop) four_probes = threading.Event()