From b1c7b965476acb321fa517d62927c91b9e6782fa Mon Sep 17 00:00:00 2001 From: kshitijk4poor <82637225+kshitijk4poor@users.noreply.github.com> Date: Sat, 4 Jul 2026 21:03:46 +0530 Subject: [PATCH] fix(telegram): bound the 3 sibling updater.stop() calls with the same CLOSE-WAIT timeout MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit The salvaged fix (#58272) guarded the primary network-error reconnect path. Issue #58270's Scope section names three more unguarded await updater.stop() sites that can hang identically on a CLOSE-WAIT socket: - conflict handler (before the retry back-off sleep) - conflict-retries-exhausted teardown (before the fatal notify) - disconnect() teardown (would hang gateway shutdown/restart) Each is now wrapped in asyncio.wait_for(..., _UPDATER_STOP_TIMEOUT) with a warning on timeout, matching the primary path, so no reconnect/teardown ladder can wedge on a dead socket. Also hoist the shared 15.0s bound to a single module constant _UPDATER_STOP_TIMEOUT (self-documenting + DRY across all 4 sites), and update the CLOSE-WAIT regression test to patch that constant instead of monkeypatching asyncio.wait_for process-wide. Fatal-notify idempotency: bounding the conflict-exhausted teardown stop() adds an await AFTER _set_fatal_error, which yields the loop and lets a concurrent retry task (scheduled by an earlier conflict, already suspended past the entry guard) reach the fatal branch too — double-firing the fatal handler (surfaced as a Python 3.11 CI failure in test_polling_conflict_becomes_fatal_after_retries). Snapshot the pre-transition fatal state and only notify on the first transition. --- plugins/platforms/telegram/adapter.py | 59 ++++++++++++++++--- .../test_telegram_network_reconnect.py | 11 ++-- 2 files changed, 56 insertions(+), 14 deletions(-) diff --git a/plugins/platforms/telegram/adapter.py b/plugins/platforms/telegram/adapter.py index 870fc326d23..2e723223aef 100644 --- a/plugins/platforms/telegram/adapter.py +++ b/plugins/platforms/telegram/adapter.py @@ -397,6 +397,14 @@ def _rich_normalize_linebreaks(text: str) -> str: return ''.join(out) +# Watchdog bound for `await updater.stop()`. When the underlying TCP socket is +# in CLOSE-WAIT the PTB polling task is blocked on epoll on the dead socket and +# never wakes, so an unguarded stop() hangs indefinitely and wedges the whole +# reconnect/teardown ladder. This is an internal safety bound (not a user knob), +# applied identically at every stop() site so no path can hang on a dead socket. +_UPDATER_STOP_TIMEOUT = 15.0 + + class TelegramAdapter(BasePlatformAdapter): """ Telegram bot adapter. @@ -2001,7 +2009,7 @@ class TelegramAdapter(BasePlatformAdapter): # the gateway silently drops messages for hours. # Bounding stop() lets the reconnect ladder always advance. # Refs: NousResearch/hermes-agent#58270 - await asyncio.wait_for(app.updater.stop(), timeout=15.0) + await asyncio.wait_for(app.updater.stop(), timeout=_UPDATER_STOP_TIMEOUT) except asyncio.TimeoutError: logger.warning( "[%s] updater.stop() timed out during network-error " @@ -2382,10 +2390,19 @@ class TelegramAdapter(BasePlatformAdapter): ) # Stop the local updater cleanly before sleeping. If it's already # stopped (e.g. PTB raised before updater.running was set) this is - # a no-op. + # a no-op. Bounded with a timeout for the same reason as the + # network-error path: a CLOSE-WAIT socket can wedge stop() on epoll + # forever, which would stall the conflict-retry ladder. try: if self._app and self._app.updater and self._app.updater.running: - await self._app.updater.stop() + try: + await asyncio.wait_for(self._app.updater.stop(), timeout=_UPDATER_STOP_TIMEOUT) + except asyncio.TimeoutError: + logger.warning( + "[%s] updater.stop() timed out during conflict " + "retry (likely CLOSE-WAIT socket); continuing", + self.name, + ) except Exception: pass @@ -2453,16 +2470,33 @@ class TelegramAdapter(BasePlatformAdapter): "[%s] %s Original error: %s", self.name, message, error, ) + # Snapshot whether we are the call that actually transitions to fatal. + # A concurrent retry task scheduled by an earlier conflict may already + # be suspended past the entry guard; once _set_fatal_error flips the + # flag, adding an await below (the bounded stop()) yields the loop and + # lets that task reach this branch too — double-notifying the fatal + # handler. Only the first transition notifies. + _already_fatal = ( + self.has_fatal_error + and self.fatal_error_code == "telegram_polling_conflict" + ) self._set_fatal_error("telegram_polling_conflict", message, retryable=False) try: if self._app and self._app.updater: - await self._app.updater.stop() + await asyncio.wait_for(self._app.updater.stop(), timeout=_UPDATER_STOP_TIMEOUT) + except asyncio.TimeoutError: + logger.warning( + "[%s] updater.stop() timed out after exhausting conflict " + "retries (likely CLOSE-WAIT socket); proceeding to fatal notify", + self.name, + ) except Exception as stop_error: logger.warning( "[%s] Failed stopping Telegram updater after exhausting conflict retries: %s", self.name, stop_error, exc_info=True, ) - await self._notify_fatal_error() + if not _already_fatal: + await self._notify_fatal_error() async def _create_dm_topic( self, @@ -3352,9 +3386,20 @@ class TelegramAdapter(BasePlatformAdapter): if self._app: try: - # Only stop the updater if it's running + # Only stop the updater if it's running. Bounded with a + # timeout: a CLOSE-WAIT socket can wedge stop() on epoll + # indefinitely, which would hang disconnect() (and any + # gateway shutdown/restart waiting on it) forever. On timeout + # we fall through to app.stop()/shutdown() to force teardown. if self._app.updater and self._app.updater.running: - await self._app.updater.stop() + try: + await asyncio.wait_for(self._app.updater.stop(), timeout=_UPDATER_STOP_TIMEOUT) + except asyncio.TimeoutError: + logger.warning( + "[%s] updater.stop() timed out during disconnect " + "(likely CLOSE-WAIT socket); forcing app shutdown", + self.name, + ) if self._app.running: await self._app.stop() await self._app.shutdown() diff --git a/tests/gateway/test_telegram_network_reconnect.py b/tests/gateway/test_telegram_network_reconnect.py index a13623f75d1..0016b0e32e7 100644 --- a/tests/gateway/test_telegram_network_reconnect.py +++ b/tests/gateway/test_telegram_network_reconnect.py @@ -856,15 +856,12 @@ async def test_handle_polling_network_error_updater_stop_timeout(): app.updater.start_polling = AsyncMock(side_effect=_fake_start_polling) - # Patch the timeout constant so the test completes fast. + # Shrink the stop() watchdog bound so the test completes fast instead of + # waiting the full _UPDATER_STOP_TIMEOUT. Patching the named constant is + # cleaner than monkeypatching asyncio.wait_for process-wide. import plugins.platforms.telegram.adapter as _mod - original_wait_for = asyncio.wait_for - async def _fast_wait_for(coro, timeout): - # Substitute a 0.05s timeout so the test doesn't take 15s. - return await original_wait_for(coro, 0.05) - - with patch("asyncio.wait_for", side_effect=_fast_wait_for): + with patch.object(_mod, "_UPDATER_STOP_TIMEOUT", 0.05): await adapter._handle_polling_network_error(OSError("CLOSE-WAIT test")) # The reconnect ladder must have advanced past the hung stop().