From 365134f66990e2ac5f70bbb826247bbc271036f1 Mon Sep 17 00:00:00 2001 From: StellarisW Date: Sun, 19 Jul 2026 02:01:51 +0800 Subject: [PATCH] fix(gateway): recover watchdog after transient loop stalls --- gateway/systemd_notify.py | 32 +++++++++-- tests/gateway/test_systemd_notify.py | 67 +++++++++++++++++++++- website/docs/user-guide/messaging/index.md | 13 +++-- 3 files changed, 100 insertions(+), 12 deletions(-) diff --git a/gateway/systemd_notify.py b/gateway/systemd_notify.py index 4fb2b9d5eadf6..671d3d4c195d4 100644 --- a/gateway/systemd_notify.py +++ b/gateway/systemd_notify.py @@ -60,6 +60,8 @@ def watchdog_interval_seconds() -> Optional[float]: class SystemdWatchdog: """Feed systemd while the asyncio event loop continues to make progress.""" + _RECOVERY_TICKS = 2 + def __init__( self, *, @@ -71,6 +73,7 @@ def __init__( self._lag_tolerance_seconds = lag_tolerance_seconds self._task: Optional[asyncio.Task[None]] = None self._unhealthy = False + self._timely_ticks = 0 self._stopping = False self._stopping_notified = False @@ -111,6 +114,7 @@ def start(self) -> bool: return False self._stopping = False self._unhealthy = False + self._timely_ticks = 0 self._stopping_notified = False self._task = asyncio.create_task(self._run(), name="hermes-systemd-watchdog") return True @@ -123,8 +127,14 @@ def ready(self, status: str = "Gateway running") -> bool: return notify(f"READY=1\nSTATUS={safe_status}") def record_tick(self, *, scheduled_at: float, now: float) -> bool: - """Feed systemd only when the event loop woke within its lag budget.""" - if not self.enabled or self._stopping or self._unhealthy: + """Feed systemd whenever the event loop is making progress again. + + A late wake-up is evidence of starvation, but the fact that this method + is running means the loop recovered before systemd killed the process. + Keep feeding the external watchdog while reporting a degraded status; + otherwise one transient delay permanently guarantees a later restart. + """ + if not self.enabled or self._stopping: return False try: lag = float(now) - float(scheduled_at) @@ -132,8 +142,20 @@ def record_tick(self, *, scheduled_at: float, now: float) -> bool: lag = float("inf") if not math.isfinite(lag) or lag > self._lag_tolerance(): self._unhealthy = True - notify("STATUS=watchdog unhealthy: event loop progress is late") - return False + self._timely_ticks = 0 + notify("WATCHDOG=1\nSTATUS=watchdog degraded: event loop progress was late") + return True + + if self._unhealthy: + self._timely_ticks += 1 + if self._timely_ticks >= self._RECOVERY_TICKS: + self._unhealthy = False + self._timely_ticks = 0 + notify( + "WATCHDOG=1\nSTATUS=watchdog healthy: event loop progress recovered" + ) + return True + notify("WATCHDOG=1") return True @@ -145,7 +167,7 @@ async def _run(self) -> None: loop = asyncio.get_running_loop() scheduled_at = loop.time() + cadence try: - while not self._stopping and not self._unhealthy: + while not self._stopping: await asyncio.sleep(max(0.0, scheduled_at - loop.time())) now = loop.time() if not self.record_tick(scheduled_at=scheduled_at, now=now): diff --git a/tests/gateway/test_systemd_notify.py b/tests/gateway/test_systemd_notify.py index b0dea324cb329..f872e9a855aa5 100644 --- a/tests/gateway/test_systemd_notify.py +++ b/tests/gateway/test_systemd_notify.py @@ -4,6 +4,7 @@ import asyncio import socket +import time import pytest @@ -55,6 +56,70 @@ def send(self, payload): assert calls[0] == ("setblocking", False) +def test_watchdog_recovers_after_loop_progress_is_late(monkeypatch): + calls: list[str] = [] + monkeypatch.setenv("NOTIFY_SOCKET", "/tmp/hermes-test-notify") + monkeypatch.setenv("WATCHDOG_USEC", "1000000") + + import gateway.systemd_notify as notify_mod + + monkeypatch.setattr( + notify_mod, "notify", lambda message: calls.append(message) or True + ) + watchdog = notify_mod.SystemdWatchdog(lag_tolerance_seconds=0.1) + + assert watchdog.record_tick(scheduled_at=10.0, now=10.05) is True + assert calls == ["WATCHDOG=1"] + assert watchdog.record_tick(scheduled_at=10.0, now=10.2) is True + assert watchdog.unhealthy is True + assert "WATCHDOG=1" in calls[-1] + assert "STATUS=watchdog degraded" in calls[-1] + + # A recovered event loop must keep feeding systemd. Require two timely + # samples before clearing the degraded status so one lucky wake-up does not + # hide recurring starvation. + assert watchdog.record_tick(scheduled_at=10.3, now=10.35) is True + assert watchdog.unhealthy is True + assert watchdog.record_tick(scheduled_at=10.4, now=10.45) is True + assert watchdog.unhealthy is False + assert "STATUS=watchdog healthy" in calls[-1] + + +@pytest.mark.asyncio +async def test_watchdog_task_survives_a_transient_event_loop_stall(monkeypatch): + calls: list[str] = [] + degraded = asyncio.Event() + healthy = asyncio.Event() + monkeypatch.setenv("NOTIFY_SOCKET", "/tmp/hermes-test-notify") + monkeypatch.setenv("WATCHDOG_USEC", "100000") + + import gateway.systemd_notify as notify_mod + + def _capture(message: str) -> bool: + calls.append(message) + if "STATUS=watchdog degraded" in message: + degraded.set() + if "STATUS=watchdog healthy" in message: + healthy.set() + return True + + monkeypatch.setattr(notify_mod, "notify", _capture) + watchdog = notify_mod.SystemdWatchdog(lag_tolerance_seconds=0.01) + + assert watchdog.start() is True + try: + await asyncio.sleep(0) # Let the watchdog establish its first deadline. + time.sleep(0.08) # Fault injection: delay the loop, but not past WatchdogSec. + + await asyncio.wait_for(degraded.wait(), timeout=2.0) + await asyncio.wait_for(healthy.wait(), timeout=2.0) + assert watchdog.task is not None + assert not watchdog.task.done() + assert calls.count("WATCHDOG=1") >= 1 + finally: + await watchdog.stop() + + @pytest.mark.asyncio async def test_watchdog_sends_ready_heartbeat_and_stopping(monkeypatch): calls: list[str] = [] @@ -77,5 +142,3 @@ async def test_watchdog_sends_ready_heartbeat_and_stopping(monkeypatch): assert "WATCHDOG=1" in calls assert calls[-1] == "STOPPING=1" assert watchdog.unhealthy is False - - diff --git a/website/docs/user-guide/messaging/index.md b/website/docs/user-guide/messaging/index.md index fb6098e003806..914285e7820c1 100644 --- a/website/docs/user-guide/messaging/index.md +++ b/website/docs/user-guide/messaging/index.md @@ -180,11 +180,14 @@ hermes gateway install --force ``` A positive value makes the generated unit use `Type=notify`, -`NotifyAccess=main`, and the matching `WatchdogSec`. Hermes sends heartbeats -only while its event loop is making timely progress; systemd restarts the -process when they stop. The default `0` keeps the existing `Type=simple` -behavior. This setting is Linux/systemd-only and does not treat an ordinary -platform network disconnect as an event-loop failure. +`NotifyAccess=main`, and the matching `WatchdogSec`. Hermes normally sends +heartbeats while its event loop makes timely progress. A late callback shows +that the loop has resumed, so Hermes renews the watchdog lease with a degraded +status and reports healthy again after two consecutive timely samples. A full +stall prevents the callback from running, so no heartbeat is sent and systemd +restarts the process when `WatchdogSec` expires. The default `0` keeps the +existing `Type=simple` behavior. This setting is Linux/systemd-only and does +not treat an ordinary platform network disconnect as an event-loop failure. ## Chat Commands (Inside Messaging)