From c2cb37532c43ae3aa7d9aabc21c1255f52bbf4bb Mon Sep 17 00:00:00 2001 From: Brooklyn Nicholson Date: Sat, 18 Jul 2026 21:31:15 -0400 Subject: [PATCH] fix(telegram): add cause-agnostic wedged-recovery watchdog so the reconnect ladder can't freeze silently (#66377) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit The Telegram gateway could go silently deaf for hours: the reconnect ladder stalled mid-way (e.g. "attempt 4/10, reconnecting in 40s" then nothing) while the process stayed active(running), so Restart=always never fired. Root class: every recovery path — the ladder's re-entry (_schedule_polling_recovery), the pending-update probe (_probe_pending_updates), and PTB's error callback — gates new recovery on _polling_error_task.done(). If that single task wedges on any hung await, all recovery returns early forever and nothing retries. The heartbeat loop is a separate task, so make it an independent, cause-agnostic watchdog: if the same recovery task stays in-flight past _POLLING_ERROR_TASK_STUCK_TIMEOUT (300s — well beyond a healthy ladder attempt's bounded stop+drain+start+backoff), force a retryable-fatal so the background reconnector rebuilds the adapter instead of relying on the frozen ladder. This guarantees progress regardless of *where* the stall is (issue direction #1), tracked locally so no task-assignment site needs to change. Also salvages @koduri-mahesh-bhushan-chowdary's #66492 (drain-await timeout), which closes the one concrete wedge vector documented in the incident (_drain_polling_connections' unbounded shutdown()/initialize() on a wedged CLOSE-WAIT pool). The watchdog covers the rest of the class. Co-authored-by: Koduri Mahesh Bhushan Chowdary --- plugins/platforms/telegram/adapter.py | 57 +++++++++++++++++++ .../test_telegram_network_reconnect.py | 50 ++++++++++++++++ 2 files changed, 107 insertions(+) diff --git a/plugins/platforms/telegram/adapter.py b/plugins/platforms/telegram/adapter.py index f37c41c0c6db..d91f2a799828 100644 --- a/plugins/platforms/telegram/adapter.py +++ b/plugins/platforms/telegram/adapter.py @@ -17,6 +17,7 @@ import os import html as _html import re import threading +import time from contextvars import ContextVar from datetime import datetime, timezone from typing import Dict, List, Optional, Set, Any @@ -551,6 +552,16 @@ _UPDATER_START_TIMEOUT = 30.0 # so the ladder always advances toward the fatal-restart escalation. Matches # _UPDATER_STOP_TIMEOUT. Refs: NousResearch/hermes-agent#66377 _DRAIN_TIMEOUT = 15.0 +# Cause-agnostic wedged-recovery watchdog (#66377). Every recovery path (the +# reconnect ladder's re-entry, the pending-update probe, PTB's error callback) +# gates new recovery on ``_polling_error_task.done()``; if that task ever wedges +# on a hung await that no local bound covers, the whole gateway goes silently +# deaf with nothing retrying. The heartbeat loop force-escalates a recovery task +# that stays in-flight far longer than any healthy ladder attempt could take — +# stop (_UPDATER_STOP_TIMEOUT) + drain (2x_DRAIN_TIMEOUT) + start +# (_UPDATER_START_TIMEOUT) + max backoff (60s) is ~135s, so 300s is +# unambiguously stuck. +_POLLING_ERROR_TASK_STUCK_TIMEOUT = 300.0 # A generation is not healthy until the dedicated getUpdates request returns # successfully. This exceeds a normal long-poll cycle for healthy idle bots. _POLLING_PROGRESS_TIMEOUT = 60.0 @@ -2464,6 +2475,16 @@ class TelegramAdapter(BasePlatformAdapter): HEARTBEAT_INTERVAL = 90 # seconds between probes PROBE_TIMEOUT = 15 # seconds before declaring the path dead + # Wedged-recovery watchdog state (#66377). Tracked locally so no + # _polling_error_task assignment site needs to stamp a timestamp: the + # heartbeat notes when it first observes a given recovery task still + # in-flight, and force-escalates if the *same* task object is still + # running after _POLLING_ERROR_TASK_STUCK_TIMEOUT. A healthy ladder + # attempt completes (task done) or chains to a new task well before + # then, so a single long-lived task is unambiguously wedged. + stuck_task_ref: Optional[asyncio.Task] = None + stuck_task_since = 0.0 + while True: try: await asyncio.sleep(HEARTBEAT_INTERVAL) @@ -2471,6 +2492,42 @@ class TelegramAdapter(BasePlatformAdapter): return if self.has_fatal_error: return + + # Independent wedged-recovery watchdog (#66377): if the tracked + # recovery task has hung (any await no local bound covers), every + # other recovery path is gated behind it and returns early + # forever — the gateway stays alive but deaf. Force a + # retryable-fatal so the background reconnector rebuilds the + # adapter instead of relying on the frozen ladder. + recovery_task = self._polling_error_task + if recovery_task is not None and not recovery_task.done(): + now = time.monotonic() + if recovery_task is not stuck_task_ref: + stuck_task_ref = recovery_task + stuck_task_since = now + elif now - stuck_task_since > _POLLING_ERROR_TASK_STUCK_TIMEOUT: + stuck_for = now - stuck_task_since + logger.error( + "[%s] Telegram reconnect task wedged for %.0fs with no " + "ladder progress; forcing retryable-fatal so the gateway " + "reconnects instead of staying silently deaf.", + self.name, stuck_for, + ) + try: + recovery_task.cancel() + except Exception: + pass + self._set_fatal_error( + "telegram_network_error", + "Telegram reconnect task wedged for %.0fs; forcing " + "gateway reconnect." % stuck_for, + retryable=True, + ) + await self._notify_fatal_error() + return + else: + stuck_task_ref = None + bot = self._app.bot if self._app else None if bot is None: continue diff --git a/tests/gateway/test_telegram_network_reconnect.py b/tests/gateway/test_telegram_network_reconnect.py index 7b8ed87a0ba4..4ffa4a99dd88 100644 --- a/tests/gateway/test_telegram_network_reconnect.py +++ b/tests/gateway/test_telegram_network_reconnect.py @@ -373,6 +373,56 @@ async def test_reconnect_continues_if_drain_hangs(monkeypatch): ) +@pytest.mark.asyncio +async def test_heartbeat_force_escalates_wedged_recovery_task(monkeypatch): + """#66377: the heartbeat is an independent, cause-agnostic watchdog. + + Every recovery path (ladder re-entry, pending-update probe, PTB error + callback) gates new recovery on ``_polling_error_task.done()``. If that task + wedges on ANY hung await — not just the drain closed by #66492 — the gateway + stays alive but deaf with nothing retrying. The heartbeat must detect a + recovery task that stays in-flight past ``_POLLING_ERROR_TASK_STUCK_TIMEOUT`` + and force a retryable-fatal so the background reconnector rebuilds the + adapter. + """ + adapter = _make_adapter() + + async def _wedged(): + await asyncio.Event().wait() # never completes — simulates the hang + + wedged_task = asyncio.ensure_future(_wedged()) + adapter._polling_error_task = wedged_task + + mock_bot = MagicMock() + mock_bot.get_me = AsyncMock() + mock_app = MagicMock() + mock_app.bot = mock_bot + adapter._app = mock_app + adapter._probe_pending_updates = AsyncMock() + adapter._notify_fatal_error = AsyncMock() + + # Controllable monotonic clock advanced by each (mocked) heartbeat sleep so + # the same wedged task is observed across the stuck threshold deterministically. + clock = [1000.0] + + async def _fake_sleep(*_a, **_k): + clock[0] += 200.0 + + monkeypatch.setattr(tg_adapter.time, "monotonic", lambda: clock[0]) + + with patch("asyncio.sleep", new=AsyncMock(side_effect=_fake_sleep)): + await asyncio.wait_for(adapter._polling_heartbeat_loop(), timeout=5) + + assert adapter.has_fatal_error, "wedged recovery task must force a fatal escalation" + adapter._notify_fatal_error.assert_awaited() + + wedged_task.cancel() + try: + await wedged_task + except asyncio.CancelledError: + pass + + @pytest.mark.asyncio async def test_conflict_retry_also_drains_polling_connections(): """_handle_polling_conflict must also drain the polling pool on retry."""