From 6fd9e9d8c3a7396c1c00987d844fd9450b6edfee Mon Sep 17 00:00:00 2001 From: xxxigm Date: Sat, 27 Jun 2026 19:16:25 -0700 Subject: [PATCH] fix(gateway/discord): REST liveness probe to detect zombie clients (#26656) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit The Discord adapter could enter a silent zombie state after a network outage / proxy stall: the process is alive, _client looks open, but the underlying socket is dead. discord.py's WebSocket reconnect never sees a RST through a wedged proxy/NAT, so client.start() spins forever without exiting — which means the bot-task done callback (which only fires on task completion) never trips either. The bot stays "offline" in Discord until a manual `hermes gateway restart`. Reported offline for 13-17h. Adds an out-of-band REST liveness probe in DiscordAdapter. Every `discord.liveness_interval_seconds` (default 60s) the adapter issues a cheap fetch_user(bot_id) — the same REST path as message delivery, so it fails when the proxy/NAT is wedged. After `discord.liveness_failure_threshold` consecutive failures (default 3) the probe closes the wedged client and surfaces a retryable fatal error, which trips the gateway's existing _platform_reconnect_watcher and rebuilds the adapter. Operators disable it by setting either knob to 0. Config lives in config.yaml (discord.liveness_*) per the .env-is-secrets policy; _apply_yaml_config bridges it to internal env vars the adapter reads, matching the existing HERMES_DISCORD_TEXT_BATCH_* pattern. Co-authored-by: Hermes Agent --- plugins/platforms/discord/adapter.py | 120 ++++++++++- tests/gateway/test_discord_liveness.py | 188 ++++++++++++++++++ .../docs/reference/environment-variables.md | 2 + 3 files changed, 309 insertions(+), 1 deletion(-) create mode 100644 tests/gateway/test_discord_liveness.py diff --git a/plugins/platforms/discord/adapter.py b/plugins/platforms/discord/adapter.py index 4327ce59afbc6..f5c83aede45b1 100644 --- a/plugins/platforms/discord/adapter.py +++ b/plugins/platforms/discord/adapter.py @@ -103,7 +103,7 @@ def __init__(self, id: int) -> None: # noqa: A002 - matches discord API from gateway.config import Platform, PlatformConfig from gateway.platforms.helpers import MessageDeduplicator, ThreadParticipationTracker, convert_table_to_bullets -from utils import atomic_json_write, env_float +from utils import atomic_json_write, env_float, env_int from gateway.platforms.base import ( BasePlatformAdapter, MessageEvent, @@ -781,6 +781,23 @@ def __init__(self, config: PlatformConfig): self._typing_tasks: Dict[str, asyncio.Task] = {} self._bot_task: Optional[asyncio.Task] = None self._post_connect_task: Optional[asyncio.Task] = None + # REST-level liveness probe. discord.py's WS reconnect handles clean + # drops, but a dead proxy / NAT can wedge the socket without delivering + # a RST — sends time out forever and ``client.start()`` never exits, so + # the bot-task done callback never fires. See #26656. An out-of-band + # ``fetch_user`` exercises the same REST path as message delivery and + # lets us detect the zombie state, close the wedged client, and trip the + # existing retryable-fatal reconnect path. Knobs are surfaced in + # config.yaml as ``discord.liveness_interval_seconds`` / + # ``discord.liveness_failure_threshold`` (bridged to these env vars by + # ``_apply_yaml_config``); set either to 0 to disable. + self._liveness_interval_seconds = env_float( + "HERMES_DISCORD_LIVENESS_INTERVAL_SECONDS", 60.0 + ) + self._liveness_failure_threshold = env_int( + "HERMES_DISCORD_LIVENESS_FAILURE_THRESHOLD", 3 + ) + self._liveness_task: Optional[asyncio.Task] = None # True while disconnect() is intentionally closing discord.py. The # bot task's done callback uses this to distinguish an operator/service # shutdown from a runtime websocket crash. @@ -1139,6 +1156,7 @@ async def on_voice_state_update(member, before, after): await _wait_for_ready_or_bot_exit(self._ready_event, self._bot_task, timeout=30) self._running = True + self._start_liveness_probe() return True except asyncio.TimeoutError: @@ -1169,9 +1187,98 @@ async def _cancel_bot_task(self) -> None: pass self._bot_task = None + def _start_liveness_probe(self) -> None: + """Start the periodic REST liveness probe if configured. + + Idempotent: if a task is already running we leave it alone so a + re-entrant ``connect()`` cannot fork two probes against the same client. + """ + if self._liveness_interval_seconds <= 0 or self._liveness_failure_threshold <= 0: + return + if self._liveness_task and not self._liveness_task.done(): + return + self._liveness_task = asyncio.create_task(self._liveness_loop()) + + async def _liveness_loop(self) -> None: + """Probe Discord REST periodically and force a reconnect on persistent failure. + + See #26656. ``client.start()`` reconnects internally on clean WS drops, + but when the underlying socket is wedged behind a dead proxy the WS never + sees a RST and the adapter sits in a silent zombie state — process alive, + ``client.start()`` spinning, sends timing out forever, and the bot-task + done callback never fires because the task never completes. An + out-of-band ``fetch_user`` exercises the same REST path as message + delivery and lets us detect the wedge. After ``threshold`` consecutive + failures we close the client, set a retryable fatal error, and hand + control back to the gateway's platform reconnect watcher. + """ + interval = self._liveness_interval_seconds + threshold = self._liveness_failure_threshold + fails = 0 + while self._running: + try: + await asyncio.sleep(interval) + except asyncio.CancelledError: + return + client = self._client + if not self._running or client is None or getattr(self, "_disconnecting", False): + return + if hasattr(client, "is_closed") and client.is_closed(): + return + user = getattr(client, "user", None) + if user is None: + continue + try: + await client.fetch_user(user.id) + fails = 0 + except asyncio.CancelledError: + return + except Exception as exc: + fails += 1 + logger.warning( + "[%s] Discord liveness probe failed (%d/%d): %s", + self.name, fails, threshold, exc, + ) + if fails < threshold: + continue + logger.error( + "[%s] Discord client appears dead, forcing reconnect", self.name, + ) + try: + await client.close() + except Exception: + logger.debug( + "[%s] Error closing wedged Discord client", self.name, exc_info=True, + ) + self._set_fatal_error( + "liveness_probe_failed", + f"Discord REST liveness probe failed {fails} times in a row", + retryable=True, + ) + try: + await self._notify_fatal_error() + except Exception: + logger.debug( + "[%s] Fatal-error handler raised", self.name, exc_info=True, + ) + return + + async def _cancel_liveness_task(self) -> None: + """Cancel and await the liveness probe task, if running.""" + if self._liveness_task and not self._liveness_task.done(): + self._liveness_task.cancel() + try: + await self._liveness_task + except asyncio.CancelledError: + pass + self._liveness_task = None + async def disconnect(self) -> None: """Disconnect from Discord.""" self._disconnecting = True + # Cancel the liveness probe first so it can't fire a spurious fatal + # error / reconnect while we're intentionally tearing the adapter down. + await self._cancel_liveness_task() # Cancel the bot task before closing the client. If connect() timed out # and returned False, the background client.start() task may still be # running; calling client.close() alone is not enough to stop it because @@ -1203,6 +1310,7 @@ async def disconnect(self) -> None: self._client = None self._ready_event.clear() self._post_connect_task = None + self._liveness_task = None self._release_platform_lock() @@ -7136,6 +7244,16 @@ def _apply_yaml_config(yaml_cfg: dict, discord_cfg: dict) -> dict | None: if _discord_rtm is not None and not os.getenv("DISCORD_REPLY_TO_MODE"): _rtm_str = "off" if _discord_rtm is False else str(_discord_rtm).lower() os.environ["DISCORD_REPLY_TO_MODE"] = _rtm_str + # liveness probe knobs: detect zombie clients behind dead proxies/NATs and + # force a reconnect (#26656). Bridged to the env vars the adapter reads in + # __init__; set either to 0 to disable. config.yaml is the user-facing + # surface — these env vars are an internal mechanism only. + lis = discord_cfg.get("liveness_interval_seconds") + if lis is not None and not os.getenv("HERMES_DISCORD_LIVENESS_INTERVAL_SECONDS"): + os.environ["HERMES_DISCORD_LIVENESS_INTERVAL_SECONDS"] = str(lis) + lft = discord_cfg.get("liveness_failure_threshold") + if lft is not None and not os.getenv("HERMES_DISCORD_LIVENESS_FAILURE_THRESHOLD"): + os.environ["HERMES_DISCORD_LIVENESS_FAILURE_THRESHOLD"] = str(lft) return None # all settings flow through env; nothing to merge into extras diff --git a/tests/gateway/test_discord_liveness.py b/tests/gateway/test_discord_liveness.py new file mode 100644 index 0000000000000..b54dd6bfdf482 --- /dev/null +++ b/tests/gateway/test_discord_liveness.py @@ -0,0 +1,188 @@ +"""Regression tests for the Discord REST liveness probe (#26656). + +discord.py's WebSocket reconnect handles clean drops, but a wedged proxy / +NAT can leave the underlying socket dead without ever delivering a RST — +sends time out forever while ``client.start()`` happily spins and never +exits, so the bot-task done callback never fires either. The probe in +``DiscordAdapter`` periodically hits Discord REST so we can detect the +zombie state and trip the gateway's existing reconnect path. +""" + +from __future__ import annotations + +import asyncio +from types import SimpleNamespace +from unittest.mock import AsyncMock + +import pytest + +# Re-use the shared discord-stub bootstrap and FakeBot from the connect +# test module so this file doesn't duplicate the (large) mock surface. +from tests.gateway.test_discord_connect import ( # noqa: E402 + FakeBot, + _ensure_discord_mock, +) + +_ensure_discord_mock() + +import plugins.platforms.discord.adapter as discord_platform # noqa: E402 +from gateway.config import PlatformConfig # noqa: E402 +from plugins.platforms.discord.adapter import DiscordAdapter # noqa: E402 + + +class _LiveBot(FakeBot): + """A FakeBot whose ``start()`` stays pending like a real discord.py client. + + The default ``FakeBot.start()`` returns immediately, which would let the + bot-task done callback fire and set a spurious fatal error. Real clients + keep ``start()`` running for the life of the connection; this models that + so the liveness probe is the only thing that can trip a fatal error. + """ + + def __init__(self, *, intents, proxy=None, allowed_mentions=None, **_): + super().__init__(intents=intents, allowed_mentions=allowed_mentions) + self._never = asyncio.Event() + self._closed = False + + async def start(self, token): + if "on_ready" in self._events: + await self._events["on_ready"]() + # Stay alive until close() is called — mirrors a real client. + await self._never.wait() + + def is_closed(self): + return self._closed + + async def close(self): + self._closed = True + self._never.set() + + +def _make_adapter(monkeypatch, *, interval=0.01, threshold=1) -> DiscordAdapter: + monkeypatch.setenv("HERMES_DISCORD_LIVENESS_INTERVAL_SECONDS", str(interval)) + monkeypatch.setenv("HERMES_DISCORD_LIVENESS_FAILURE_THRESHOLD", str(threshold)) + return DiscordAdapter(PlatformConfig(enabled=True, token="test-token")) + + +async def _connect(adapter: DiscordAdapter, monkeypatch, bot_factory): + monkeypatch.setattr( + "gateway.status.acquire_scoped_lock", + lambda scope, identity, metadata=None: (True, None), + ) + monkeypatch.setattr("gateway.status.release_scoped_lock", lambda scope, identity: None) + intents = SimpleNamespace( + message_content=False, dm_messages=False, guild_messages=False, + members=False, voice_states=False, + ) + monkeypatch.setattr(discord_platform.Intents, "default", lambda: intents) + monkeypatch.setattr(discord_platform.commands, "Bot", bot_factory) + monkeypatch.setattr(adapter, "_resolve_allowed_usernames", AsyncMock()) + assert await adapter.connect() is True + + +@pytest.mark.asyncio +async def test_liveness_probe_disabled_when_interval_zero(monkeypatch): + """interval<=0 must skip the probe entirely so users can opt out.""" + adapter = _make_adapter(monkeypatch, interval=0) + + bot_holder: dict = {} + + def factory(**kwargs): + bot = _LiveBot(intents=kwargs["intents"], allowed_mentions=kwargs.get("allowed_mentions")) + bot.fetch_user = AsyncMock() + bot_holder["bot"] = bot + return bot + + await _connect(adapter, monkeypatch, factory) + assert adapter._liveness_task is None + await asyncio.sleep(0.05) + bot_holder["bot"].fetch_user.assert_not_called() + await adapter.disconnect() + + +@pytest.mark.asyncio +async def test_liveness_probe_disabled_when_threshold_zero(monkeypatch): + """threshold<=0 must also skip the probe.""" + adapter = _make_adapter(monkeypatch, interval=0.01, threshold=0) + + def factory(**kwargs): + bot = _LiveBot(intents=kwargs["intents"], allowed_mentions=kwargs.get("allowed_mentions")) + bot.fetch_user = AsyncMock() + return bot + + await _connect(adapter, monkeypatch, factory) + assert adapter._liveness_task is None + await adapter.disconnect() + + +@pytest.mark.asyncio +async def test_liveness_probe_pings_rest_while_healthy(monkeypatch): + """A healthy probe keeps the adapter running and never sets a fatal error.""" + adapter = _make_adapter(monkeypatch, interval=0.01, threshold=3) + + def factory(**kwargs): + bot = _LiveBot(intents=kwargs["intents"], allowed_mentions=kwargs.get("allowed_mentions")) + bot.fetch_user = AsyncMock(return_value=SimpleNamespace(id=999)) + return bot + + await _connect(adapter, monkeypatch, factory) + await asyncio.sleep(0.05) + assert adapter._client.fetch_user.await_count >= 1 + assert adapter._running is True + assert adapter.has_fatal_error is False + await adapter.disconnect() + + +@pytest.mark.asyncio +async def test_liveness_probe_forces_reconnect_after_threshold(monkeypatch): + """Once the probe fails ``threshold`` times in a row, the adapter must + close the wedged client and surface a retryable fatal error so the + gateway's reconnect watcher (gateway/run.py) can rebuild it.""" + adapter = _make_adapter(monkeypatch, interval=0.005, threshold=2) + + def factory(**kwargs): + bot = _LiveBot(intents=kwargs["intents"], allowed_mentions=kwargs.get("allowed_mentions")) + bot.fetch_user = AsyncMock(side_effect=TimeoutError("dead proxy")) + return bot + + handler = AsyncMock() + adapter.set_fatal_error_handler(handler) + await _connect(adapter, monkeypatch, factory) + wedged = adapter._client + + # Wait for the loop to exit (it returns after threshold consecutive + # failures). Bounded by a generous timeout so a regression doesn't hang CI. + for _ in range(200): + if adapter._liveness_task and adapter._liveness_task.done(): + break + await asyncio.sleep(0.01) + else: + pytest.fail("liveness loop did not terminate within 2s") + + assert wedged.is_closed() is True + assert adapter.has_fatal_error is True + assert adapter.fatal_error_code == "liveness_probe_failed" + assert adapter.fatal_error_retryable is True + handler.assert_awaited_once() + + await adapter.disconnect() + + +@pytest.mark.asyncio +async def test_disconnect_cancels_liveness_task(monkeypatch): + """``disconnect()`` must cancel the probe so the gateway can shut down + cleanly without leaking a background task.""" + adapter = _make_adapter(monkeypatch, interval=60, threshold=3) + + def factory(**kwargs): + bot = _LiveBot(intents=kwargs["intents"], allowed_mentions=kwargs.get("allowed_mentions")) + bot.fetch_user = AsyncMock() + return bot + + await _connect(adapter, monkeypatch, factory) + task = adapter._liveness_task + assert task is not None and not task.done() + + await adapter.disconnect() + assert task.done() + assert adapter._liveness_task is None diff --git a/website/docs/reference/environment-variables.md b/website/docs/reference/environment-variables.md index 646b6320b1fdb..71525ef76a203 100644 --- a/website/docs/reference/environment-variables.md +++ b/website/docs/reference/environment-variables.md @@ -569,6 +569,8 @@ Advanced per-platform knobs for throttling the outbound message batcher. Most us | `HERMES_TELEGRAM_DISABLE_FALLBACK_IPS` | Disable the hard-coded Cloudflare fallback IPs used when DNS fails (`true`/`false`). | | `HERMES_DISCORD_TEXT_BATCH_DELAY_SECONDS` | Grace window before flushing a queued Discord text chunk (default: `0.6`). | | `HERMES_DISCORD_TEXT_BATCH_SPLIT_DELAY_SECONDS` | Delay between split chunks when a Discord message exceeds the length limit (default: `2.0`). | +| `HERMES_DISCORD_LIVENESS_INTERVAL_SECONDS` | Internal bridge for `discord.liveness_interval_seconds` (config.yaml). Interval for the Discord REST liveness probe that detects zombie clients behind dead proxies/NATs (default: `60`; set to `0` to disable). Prefer setting `discord.liveness_interval_seconds` in `config.yaml`. | +| `HERMES_DISCORD_LIVENESS_FAILURE_THRESHOLD` | Internal bridge for `discord.liveness_failure_threshold` (config.yaml). Consecutive probe failures before forcing a Discord reconnect (default: `3`). Prefer setting `discord.liveness_failure_threshold` in `config.yaml`. | | `HERMES_MATRIX_TEXT_BATCH_DELAY_SECONDS` / `_SPLIT_DELAY_SECONDS` | Matrix equivalents of the Telegram batch knobs. | | `HERMES_FEISHU_TEXT_BATCH_DELAY_SECONDS` / `_SPLIT_DELAY_SECONDS` / `_MAX_CHARS` / `_MAX_MESSAGES` | Feishu batcher tuning — delay, split delay, max chars per message, max messages per batch. | | `HERMES_FEISHU_MEDIA_BATCH_DELAY_SECONDS` | Feishu media flush delay. |