From 7af3a6fc7c9ea139bf4c6ebe5575c0cabffd4756 Mon Sep 17 00:00:00 2001 From: Josh Tsai <128559392+bounce12340@users.noreply.github.com> Date: Wed, 22 Jul 2026 13:40:46 +0800 Subject: [PATCH] fix(gateway): detect and escape silent event-loop freezes - A self-rescheduling 5s call_later floor timer, armed before any adapter connects, guarantees the selector always has a finite timeout, so the existing async defenses (polling heartbeat, timeout guards) regain a chance to run after a zero-pending-timer stall. - A resident daemon-thread liveness watchdog probes the loop via call_soon_threadsafe every 30s; after 3 consecutive 10s-timeout misses (~120s of total unresponsiveness) it dumps all thread tracebacks and exits with the established GATEWAY_SERVICE_RESTART_EXIT_CODE (75) so a supervisor restarts the gateway - async-level recovery cannot run on a frozen loop. - stop() disarms both guards before any teardown await so a busy shutdown is never misjudged as a freeze. HERMES_GATEWAY_LOOP_WATCHDOG=0 disables; _INTERVAL/_TIMEOUT/_STRIKES tune the thresholds. Fixes #69089 Co-Authored-By: Claude Fable 5 --- gateway/run.py | 44 +++- gateway/shutdown_watchdog.py | 178 ++++++++++++- tests/gateway/test_loop_liveness_watchdog.py | 264 +++++++++++++++++++ 3 files changed, 484 insertions(+), 2 deletions(-) create mode 100644 tests/gateway/test_loop_liveness_watchdog.py diff --git a/gateway/run.py b/gateway/run.py index ec8e9db07a9a..e25a97ad610f 100644 --- a/gateway/run.py +++ b/gateway/run.py @@ -2130,9 +2130,11 @@ from gateway.platforms.base import ( ) from gateway.shutdown_watchdog import ( DEFAULT_HEARTBEAT_INTERVAL_S, + _arm_loop_floor_timer, arm_shutdown_watchdog, loop_heartbeat_forever, resolve_shutdown_watchdog_delay, + start_loop_liveness_watchdog, ) from gateway.restart import ( DEFAULT_GATEWAY_RESTART_DRAIN_TIMEOUT, @@ -3286,10 +3288,12 @@ class GatewayRunner(GatewayAuthorizationMixin, GatewayKanbanWatchersMixin, Gatew _session_ephemeral_pin: Dict[str, tuple] = {} _session_vc_last: Dict[str, str] = {} _startup_restore_in_progress: bool = False - # Loop-liveness heartbeat / shutdown-watchdog handles (#66892). Class-level + # Loop-liveness heartbeat / watchdog handles (#66892, #69089). Class-level # defaults so partial construction in tests doesn't blow up on access; the # real values are set in __init__ / start() / stop(). _loop_heartbeat_task: Optional["asyncio.Task"] = None + _loop_floor_timer_handle: Optional[Any] = None + _loop_liveness_watchdog: Optional[Any] = None _gateway_started_at: float = 0.0 _shutdown_watchdog_done: Optional["threading.Event"] = None _platform_lock_takeover_on_start: bool = False @@ -3673,6 +3677,8 @@ class GatewayRunner(GatewayAuthorizationMixin, GatewayKanbanWatchersMixin, Gatew # updated_at to distinguish "process alive" from "loop frozen". self._gateway_started_at: float = time.time() self._loop_heartbeat_task: Optional[asyncio.Task] = None + self._loop_floor_timer_handle = None + self._loop_liveness_watchdog = None # scale-to-zero (Phase 0, F13): gateway-scoped "last inbound seen" clock. # There is no such clock today (only a per-agent _last_activity_ts), so the @@ -7756,6 +7762,39 @@ class GatewayRunner(GatewayAuthorizationMixin, GatewayKanbanWatchersMixin, Gatew ) return True + def _start_loop_liveness_guards(self, loop: asyncio.AbstractEventLoop) -> None: + """Arm the selector floor and out-of-loop watchdog before adapters.""" + if getattr(self, "_loop_floor_timer_handle", None) is None: + try: + self._loop_floor_timer_handle = _arm_loop_floor_timer(loop) + except Exception: + logger.debug("Failed to arm gateway loop floor timer", exc_info=True) + + watchdog = getattr(self, "_loop_liveness_watchdog", None) + if watchdog is None or not watchdog.is_alive(): + try: + self._loop_liveness_watchdog = start_loop_liveness_watchdog(loop) + except Exception: + logger.debug("Failed to start gateway loop liveness watchdog", exc_info=True) + + def _stop_loop_liveness_guards(self) -> None: + """Disarm lifetime liveness guards before shutdown can load the loop.""" + watchdog = getattr(self, "_loop_liveness_watchdog", None) + self._loop_liveness_watchdog = None + if watchdog is not None: + try: + watchdog.stop() + except Exception: + logger.debug("Failed to stop gateway loop liveness watchdog", exc_info=True) + + floor_timer = getattr(self, "_loop_floor_timer_handle", None) + self._loop_floor_timer_handle = None + if floor_timer is not None: + try: + floor_timer.cancel() + except Exception: + logger.debug("Failed to cancel gateway loop floor timer", exc_info=True) + async def start(self) -> bool: """ Start the gateway and all configured platform adapters. @@ -7796,6 +7835,8 @@ class GatewayRunner(GatewayAuthorizationMixin, GatewayKanbanWatchersMixin, Gatew self._gateway_loop = asyncio.get_running_loop() except RuntimeError: self._gateway_loop = None + if self._gateway_loop is not None: + self._start_loop_liveness_guards(self._gateway_loop) logger.info("Session storage: %s", self.config.sessions_dir) # Sanity-check that systemd's TimeoutStopSec covers our drain @@ -9383,6 +9424,7 @@ class GatewayRunner(GatewayAuthorizationMixin, GatewayKanbanWatchersMixin, Gatew service_restart: bool = False, ) -> None: """Stop the gateway and disconnect all adapters.""" + self._stop_loop_liveness_guards() if restart: self._restart_requested = True self._restart_detached = detached_restart diff --git a/gateway/shutdown_watchdog.py b/gateway/shutdown_watchdog.py index 492160996c79..5dbc4a9b81bf 100644 --- a/gateway/shutdown_watchdog.py +++ b/gateway/shutdown_watchdog.py @@ -1,4 +1,4 @@ -"""Out-of-loop shutdown backstop + event-loop liveness heartbeat (#66892). +"""Out-of-loop shutdown and event-loop liveness backstops (#66892, #69089). When the asyncio loop freezes mid-drain, every asyncio-based recovery path is structurally unable to fire: the drain deadline, status rewrites, and forensics @@ -15,6 +15,10 @@ This module provides: 2. An event-loop heartbeat file at ``/state/gateway.heartbeat`` so external supervision can distinguish "process alive" from "loop frozen" (``gateway_state.json`` alone can't — it only rewrites on transitions/turns). +3. A lifetime thread watchdog that can still diagnose and hard-exit when the + event loop is too frozen to run its own heartbeat or timeout callbacks. +4. A self-rescheduling floor timer that keeps the loop selector's timeout + finite, giving existing async recovery tasks a chance to resume. """ from __future__ import annotations @@ -31,6 +35,7 @@ from datetime import datetime, timezone from pathlib import Path from typing import Any, Callable, Dict, Optional +from gateway.restart import GATEWAY_SERVICE_RESTART_EXIT_CODE from hermes_constants import get_hermes_home from utils import atomic_json_write @@ -40,10 +45,181 @@ logger = logging.getLogger(__name__) # drain is not cut short. Matches the issue #66892 suggested hardening. DEFAULT_SHUTDOWN_WATCHDOG_GRACE_S = 60.0 DEFAULT_HEARTBEAT_INTERVAL_S = 30.0 +DEFAULT_LOOP_FLOOR_TIMER_INTERVAL_S = 5.0 +DEFAULT_LOOP_WATCHDOG_INTERVAL_S = 30.0 +DEFAULT_LOOP_WATCHDOG_TIMEOUT_S = 10.0 +DEFAULT_LOOP_WATCHDOG_MAX_STRIKES = 3 _HEARTBEAT_RELATIVE = ("state", "gateway.heartbeat") _WATCHDOG_DUMP_RELATIVE = ("logs", "gateway-shutdown-watchdog.log") +class _LoopFloorTimerHandle: + """Cancelable owner for the currently scheduled selector floor timer.""" + + def __init__(self, loop: asyncio.AbstractEventLoop, interval: float): + self._loop = loop + self._interval = interval + self._cancelled = False + self._timer: Optional[asyncio.TimerHandle] = None + self._schedule() + + def _schedule(self) -> None: + self._timer = self._loop.call_later(self._interval, self._tick) + + def _tick(self) -> None: + if not self._cancelled: + self._schedule() + + def cancel(self) -> None: + self._cancelled = True + if self._timer is not None: + self._timer.cancel() + + +class _LoopLivenessWatchdogHandle: + """Small lifecycle handle for the daemon liveness thread.""" + + def __init__(self, stop_event: threading.Event, thread: threading.Thread): + self._stop_event = stop_event + self._thread = thread + + def stop(self) -> None: + self._stop_event.set() + + def join(self, timeout: Optional[float] = None) -> None: + self._thread.join(timeout=timeout) + + def is_alive(self) -> bool: + return self._thread.is_alive() + + +def _arm_loop_floor_timer( + loop: asyncio.AbstractEventLoop, + interval: float = DEFAULT_LOOP_FLOOR_TIMER_INTERVAL_S, +) -> _LoopFloorTimerHandle: + """Keep at least one timer pending so selector waits remain bounded.""" + try: + resolved_interval = float(interval) + if resolved_interval <= 0: + raise ValueError + except (TypeError, ValueError): + resolved_interval = DEFAULT_LOOP_FLOOR_TIMER_INTERVAL_S + return _LoopFloorTimerHandle(loop, resolved_interval) + + +def _positive_float_env(name: str, default: float) -> float: + raw = os.environ.get(name, "").strip() + try: + value = float(raw) if raw else float(default) + except (TypeError, ValueError): + return float(default) + return value if value > 0 else float(default) + + +def _positive_int_env(name: str, default: int) -> int: + raw = os.environ.get(name, "").strip() + try: + value = int(raw) if raw else int(default) + except (TypeError, ValueError): + return int(default) + return value if value > 0 else int(default) + + +def start_loop_liveness_watchdog( + loop: asyncio.AbstractEventLoop, + *, + probe_interval: float = DEFAULT_LOOP_WATCHDOG_INTERVAL_S, + probe_timeout: float = DEFAULT_LOOP_WATCHDOG_TIMEOUT_S, + max_strikes: int = DEFAULT_LOOP_WATCHDOG_MAX_STRIKES, + exit_code: int = GATEWAY_SERVICE_RESTART_EXIT_CODE, +) -> Optional[_LoopLivenessWatchdogHandle]: + """Start an out-of-loop watchdog that hard-exits after missed probes. + + Environment overrides are intentionally read here, the owner module, like + other gateway runtime-only safety knobs. Set + ``HERMES_GATEWAY_LOOP_WATCHDOG=0`` to disable it. + """ + enabled = os.environ.get("HERMES_GATEWAY_LOOP_WATCHDOG", "1").strip().lower() + if enabled in {"0", "false", "no", "off"}: + return None + + interval = _positive_float_env( + "HERMES_GATEWAY_LOOP_WATCHDOG_INTERVAL", probe_interval + ) + timeout = _positive_float_env("HERMES_GATEWAY_LOOP_WATCHDOG_TIMEOUT", probe_timeout) + strikes_limit = _positive_int_env( + "HERMES_GATEWAY_LOOP_WATCHDOG_STRIKES", max_strikes + ) + stop_event = threading.Event() + + def _wait_for_probe(probe_event: threading.Event) -> Optional[bool]: + deadline = time.monotonic() + timeout + while True: + if stop_event.is_set(): + return None + remaining = deadline - time.monotonic() + if remaining <= 0: + return probe_event.is_set() + if probe_event.wait(timeout=min(remaining, 0.05)): + return True + + def _watchdog() -> None: + strikes = 0 + while not stop_event.wait(timeout=interval): + probe_event = threading.Event() + try: + loop.call_soon_threadsafe(probe_event.set) + except RuntimeError: + # A normally closed loop cannot be probed and no longer needs + # a process-liveness backstop. + return + except Exception: + logger.debug( + "Failed to schedule gateway loop liveness probe", exc_info=True + ) + return + + responded = _wait_for_probe(probe_event) + if responded is None: + return + if responded: + strikes = 0 + continue + + strikes += 1 + if strikes < strikes_limit: + continue + + try: + logger.critical( + "Gateway event loop missed %d consecutive liveness probes; " + "dumping all thread stacks and exiting with code %d so the " + "service supervisor can restart it.", + strikes, + exit_code, + ) + except Exception: + pass + try: + faulthandler.dump_traceback(all_threads=True) + except Exception: + logger.debug("Loop liveness faulthandler dump failed", exc_info=True) + os._exit(exit_code) + return + + thread = threading.Thread( + target=_watchdog, + daemon=True, + name="gateway-loop-liveness-watchdog", + ) + try: + thread.start() + except Exception: + logger.debug("Failed to start gateway loop liveness watchdog", exc_info=True) + return None + return _LoopLivenessWatchdogHandle(stop_event, thread) + + def _process_hermes_home() -> Path: """HERMES_HOME for process-level identity files (ignore profile overrides).""" val = os.environ.get("HERMES_HOME", "").strip() diff --git a/tests/gateway/test_loop_liveness_watchdog.py b/tests/gateway/test_loop_liveness_watchdog.py new file mode 100644 index 000000000000..1add915341aa --- /dev/null +++ b/tests/gateway/test_loop_liveness_watchdog.py @@ -0,0 +1,264 @@ +"""Gateway event-loop freeze backstops for issue #69089.""" + +from __future__ import annotations + +import asyncio +import threading +import time +from unittest.mock import MagicMock, patch + +import pytest + +from gateway.shutdown_watchdog import ( + _arm_loop_floor_timer, + start_loop_liveness_watchdog, +) + + +def _immediate_loop() -> MagicMock: + loop = MagicMock(spec=asyncio.AbstractEventLoop) + loop.call_soon_threadsafe.side_effect = lambda callback: callback() + return loop + + +def test_loop_liveness_watchdog_responsive_probe_does_not_fire(): + loop = _immediate_loop() + exit_codes = [] + + with ( + 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=2 + ) + assert handle is not None + deadline = time.monotonic() + 2.0 + while loop.call_soon_threadsafe.call_count < 3 and time.monotonic() < deadline: + time.sleep(0.01) + handle.stop() + handle.join(timeout=1.0) + + assert loop.call_soon_threadsafe.call_count >= 3 + assert not handle.is_alive() + dump.assert_not_called() + assert exit_codes == [] + + +def test_loop_liveness_watchdog_exits_after_consecutive_misses(): + loop = MagicMock(spec=asyncio.AbstractEventLoop) + fired = threading.Event() + exit_codes = [] + + def fake_exit(code: int) -> None: + exit_codes.append(code) + fired.set() + + with ( + patch("gateway.shutdown_watchdog.faulthandler.dump_traceback") as dump, + patch("gateway.shutdown_watchdog.os._exit", side_effect=fake_exit), + ): + handle = start_loop_liveness_watchdog( + loop, probe_interval=0.01, probe_timeout=0.01, max_strikes=2 + ) + assert handle is not None + assert fired.wait(timeout=2.0), "loop liveness watchdog did not fire" + handle.join(timeout=1.0) + + assert not handle.is_alive() + assert loop.call_soon_threadsafe.call_count == 2 + dump.assert_called_once_with(all_threads=True) + assert exit_codes == [75] + + +def test_loop_liveness_watchdog_recovery_resets_strikes(): + loop = MagicMock(spec=asyncio.AbstractEventLoop) + four_probes = threading.Event() + + def alternate_response(callback) -> None: + count = loop.call_soon_threadsafe.call_count + if count in {2, 4}: + callback() + if count >= 4: + four_probes.set() + + loop.call_soon_threadsafe.side_effect = alternate_response + with ( + 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=2 + ) + assert handle is not None + assert four_probes.wait( + timeout=2.0 + ), "watchdog did not complete recovery probes" + handle.stop() + handle.join(timeout=1.0) + + assert not handle.is_alive() + dump.assert_not_called() + hard_exit.assert_not_called() + + +def test_loop_liveness_watchdog_stop_exits_thread_and_stops_probes(): + loop = MagicMock(spec=asyncio.AbstractEventLoop) + first_probe = threading.Event() + loop.call_soon_threadsafe.side_effect = lambda callback: first_probe.set() + + handle = start_loop_liveness_watchdog( + loop, probe_interval=0.01, probe_timeout=0.5, max_strikes=10 + ) + assert handle is not None + assert first_probe.wait(timeout=2.0) + handle.stop() + handle.join(timeout=1.0) + calls_after_stop = loop.call_soon_threadsafe.call_count + time.sleep(0.05) + + assert not handle.is_alive() + assert loop.call_soon_threadsafe.call_count == calls_after_stop + + +def test_loop_liveness_watchdog_env_can_disable(monkeypatch): + monkeypatch.setenv("HERMES_GATEWAY_LOOP_WATCHDOG", "0") + loop = MagicMock(spec=asyncio.AbstractEventLoop) + + handle = start_loop_liveness_watchdog( + loop, probe_interval=0.01, probe_timeout=0.01, max_strikes=1 + ) + + assert handle is None + loop.call_soon_threadsafe.assert_not_called() + + +def test_loop_liveness_watchdog_env_overrides_probe_settings(monkeypatch): + monkeypatch.setenv("HERMES_GATEWAY_LOOP_WATCHDOG_INTERVAL", "0.01") + monkeypatch.setenv("HERMES_GATEWAY_LOOP_WATCHDOG_TIMEOUT", "0.01") + monkeypatch.setenv("HERMES_GATEWAY_LOOP_WATCHDOG_STRIKES", "1") + loop = MagicMock(spec=asyncio.AbstractEventLoop) + fired = threading.Event() + + with ( + patch("gateway.shutdown_watchdog.faulthandler.dump_traceback"), + patch( + "gateway.shutdown_watchdog.os._exit", + side_effect=lambda code: fired.set(), + ), + ): + handle = start_loop_liveness_watchdog( + loop, probe_interval=10.0, probe_timeout=10.0, max_strikes=10 + ) + assert handle is not None + assert fired.wait(timeout=2.0), "env overrides were not applied" + handle.join(timeout=1.0) + + assert not handle.is_alive() + assert loop.call_soon_threadsafe.call_count == 1 + + +@pytest.mark.asyncio +async def test_loop_liveness_watchdog_detects_real_loop_sync_freeze(): + loop = asyncio.get_running_loop() + fired = threading.Event() + exit_codes = [] + + def fake_exit(code: int) -> None: + exit_codes.append(code) + fired.set() + + with ( + patch("gateway.shutdown_watchdog.faulthandler.dump_traceback") as dump, + patch("gateway.shutdown_watchdog.os._exit", side_effect=fake_exit), + ): + handle = start_loop_liveness_watchdog( + loop, probe_interval=0.02, probe_timeout=0.03, max_strikes=2 + ) + assert handle is not None + await asyncio.sleep(0.03) + time.sleep(0.25) + assert fired.wait(timeout=1.0), "watchdog did not detect the frozen real loop" + handle.stop() + handle.join(timeout=1.0) + + dump.assert_called_once_with(all_threads=True) + assert exit_codes == [75] + + +@pytest.mark.asyncio +async def test_loop_liveness_watchdog_leaves_responsive_real_loop_running(): + loop = asyncio.get_running_loop() + with ( + 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.02, probe_timeout=0.03, max_strikes=2 + ) + assert handle is not None + await asyncio.sleep(0.25) + handle.stop() + handle.join(timeout=1.0) + + assert not handle.is_alive() + dump.assert_not_called() + hard_exit.assert_not_called() + + +def test_loop_floor_timer_reschedules_until_cancelled(): + loop = MagicMock(spec=asyncio.AbstractEventLoop) + scheduled = [] + + def fake_call_later(delay, callback): + timer = MagicMock(spec=asyncio.TimerHandle) + scheduled.append((delay, callback, timer)) + return timer + + loop.call_later.side_effect = fake_call_later + handle = _arm_loop_floor_timer(loop, interval=5.0) + + assert len(scheduled) == 1 + assert scheduled[0][0] == 5.0 + scheduled[0][1]() + assert len(scheduled) == 2 + assert scheduled[1][0] == 5.0 + + handle.cancel() + scheduled[1][2].cancel.assert_called_once_with() + scheduled[1][1]() + assert len(scheduled) == 2 + + +def test_gateway_runner_liveness_guards_start_and_stop(): + from gateway.run import GatewayRunner + + runner = object.__new__(GatewayRunner) + runner._loop_floor_timer_handle = None + runner._loop_liveness_watchdog = None + loop = MagicMock(spec=asyncio.AbstractEventLoop) + floor_timer = MagicMock() + watchdog = MagicMock() + watchdog.is_alive.return_value = True + + with ( + patch( + "gateway.run._arm_loop_floor_timer", return_value=floor_timer + ) as arm_floor, + patch( + "gateway.run.start_loop_liveness_watchdog", return_value=watchdog + ) as start_watchdog, + ): + runner._start_loop_liveness_guards(loop) + + arm_floor.assert_called_once_with(loop) + start_watchdog.assert_called_once_with(loop) + assert runner._loop_floor_timer_handle is floor_timer + assert runner._loop_liveness_watchdog is watchdog + + runner._stop_loop_liveness_guards() + + watchdog.stop.assert_called_once_with() + floor_timer.cancel.assert_called_once_with() + assert runner._loop_liveness_watchdog is None + assert runner._loop_floor_timer_handle is None