From aaeba213d90c7234772280a335e6324659c004f9 Mon Sep 17 00:00:00 2001 From: kshitijk4poor <82637225+kshitijk4poor@users.noreply.github.com> Date: Tue, 7 Jul 2026 15:19:31 +0530 Subject: [PATCH] fix(telegram): bound start_polling() at bootstrap and conflict-retry sites too; strengthen tests Follow-up on the salvaged fix, which bounded start_polling() only in _handle_polling_network_error. The same wedge (#59614) exists at the two sibling call sites: 1. _start_polling_resilient (bootstrap): an exhausted pool hangs connect() forever. The TimeoutError from wait_for is a builtins TimeoutError (OSError subclass), so the existing except classifies it via _looks_like_network_error and schedules background recovery. 2. _handle_polling_conflict (conflict-retry ladder): identical hang wedges conflict attempt N forever; timeout now converts to RuntimeError and the existing except schedules the next attempt. Tests replaced with a stronger suite: hung-network-ladder repro (RED without the fix), bootstrap hang schedules recovery, success-path sanity, and a bug-class contract test asserting EVERY updater.start_polling( call site is wrapped in wait_for so a new unbounded site can't reintroduce the wedge. Verified RED (3 failures) with the wrappers removed, GREEN with them. --- plugins/platforms/telegram/adapter.py | 40 +++- .../test_telegram_start_polling_timeout.py | 209 +++++++++++------- 2 files changed, 166 insertions(+), 83 deletions(-) diff --git a/plugins/platforms/telegram/adapter.py b/plugins/platforms/telegram/adapter.py index a261d60fadb..fd5fc0177eb 100644 --- a/plugins/platforms/telegram/adapter.py +++ b/plugins/platforms/telegram/adapter.py @@ -1979,10 +1979,19 @@ class TelegramAdapter(BasePlatformAdapter): if not (self._app and self._app.updater): raise RuntimeError("Telegram application/updater not initialized") try: - await self._app.updater.start_polling( - allowed_updates=Update.ALL_TYPES, - drop_pending_updates=drop_pending_updates, - error_callback=error_callback, + # Same watchdog bound as the reconnect ladders: a wedged httpx + # connection pool can hang start_polling() forever at bootstrap + # too (#59614). A propagating TimeoutError is a builtins + # TimeoutError (OSError subclass), so the except below classifies + # it via _looks_like_network_error and schedules background + # recovery instead of blocking connect() indefinitely. + await asyncio.wait_for( + self._app.updater.start_polling( + allowed_updates=Update.ALL_TYPES, + drop_pending_updates=drop_pending_updates, + error_callback=error_callback, + ), + timeout=_UPDATER_START_TIMEOUT, ) return True except Exception as err: @@ -2488,11 +2497,24 @@ class TelegramAdapter(BasePlatformAdapter): try: if not app: raise RuntimeError("Telegram application was torn down during conflict reconnect") - await app.updater.start_polling( - allowed_updates=Update.ALL_TYPES, - drop_pending_updates=False, - error_callback=self._polling_error_callback_ref, - ) + # Same watchdog bound as the network-error ladder: an + # exhausted pool hangs start_polling() on the conflict path + # identically (#59614). Timeout converts to RuntimeError so + # the except below logs a readable message and schedules the + # next conflict attempt instead of wedging attempt N forever. + try: + await asyncio.wait_for( + app.updater.start_polling( + allowed_updates=Update.ALL_TYPES, + drop_pending_updates=False, + error_callback=self._polling_error_callback_ref, + ), + timeout=_UPDATER_START_TIMEOUT, + ) + except asyncio.TimeoutError: + raise RuntimeError( + "start_polling() timed out — connection pool may be wedged" + ) logger.info( "[%s] Telegram polling resumed after conflict retry %d/%d", self.name, self._polling_conflict_count, MAX_CONFLICT_RETRIES, diff --git a/tests/gateway/test_telegram_start_polling_timeout.py b/tests/gateway/test_telegram_start_polling_timeout.py index d89f587e0a7..12e4ef97b7d 100644 --- a/tests/gateway/test_telegram_start_polling_timeout.py +++ b/tests/gateway/test_telegram_start_polling_timeout.py @@ -1,96 +1,157 @@ -"""Test that start_polling() timeout prevents indefinite hanging. +"""Regression tests for #59614: start_polling() must be time-bounded. -This is a regression test for issue #59614 where start_polling() could hang -indefinitely when the connection pool is in a degraded state. +When both the primary Telegram API server and all fallback IPs are unreachable, +``await app.updater.start_polling(...)`` can block forever inside an exhausted +httpx connection pool — it neither returns nor raises. Unbounded, that wedges: + +1. the network-error reconnect ladder (stuck inside attempt 1, never advances), +2. the heartbeat loop (sees the recovery task as alive-but-wedged and skips), +3. the fatal-error escalation (never reached). + +The fix wraps every ``start_polling()`` await in ``asyncio.wait_for`` with +``_UPDATER_START_TIMEOUT`` so a hung call raises and feeds the existing retry +ladder. These tests patch the timeout down to keep the suite fast. """ import asyncio +import sys from unittest.mock import AsyncMock, MagicMock, patch + import pytest -# Only import TelegramAdapter if it exists -try: - from hermes.plugins.platforms.telegram import adapter - _UPDATER_START_TIMEOUT = adapter._UPDATER_START_TIMEOUT -except (ImportError, AttributeError): - pytest.skip("Telegram adapter not available", allow_module_level=True) + +def _ensure_telegram_mock(): + if "telegram" in sys.modules and hasattr(sys.modules["telegram"], "__file__"): + return + telegram_mod = MagicMock() + telegram_mod.ext.ContextTypes.DEFAULT_TYPE = type(None) + telegram_mod.constants.ParseMode.MARKDOWN_V2 = "MarkdownV2" + telegram_mod.constants.ChatType.GROUP = "group" + telegram_mod.constants.ChatType.SUPERGROUP = "supergroup" + telegram_mod.constants.ChatType.CHANNEL = "channel" + telegram_mod.constants.ChatType.PRIVATE = "private" + telegram_mod.error.NetworkError = type("NetworkError", (OSError,), {}) + telegram_mod.error.TimedOut = type("TimedOut", (OSError,), {}) + for name in ("telegram", "telegram.ext", "telegram.constants", "telegram.request"): + sys.modules.setdefault(name, telegram_mod) + sys.modules.setdefault("telegram.error", telegram_mod.error) -class TestStartPollingTimeout: - """Test that start_polling() timeout prevents indefinite hanging.""" +_ensure_telegram_mock() - @pytest.mark.asyncio - async def test_start_polling_timeout_raises_runtime_error(self): - """When start_polling() times out, it should raise RuntimeError.""" - from hermes.plugins.platforms.telegram.adapter import TelegramAdapter +from plugins.platforms.telegram import adapter as tg_adapter # noqa: E402 +from plugins.platforms.telegram.adapter import TelegramAdapter # noqa: E402 - # Mock the adapter's internal state - adapter = TelegramAdapter.__new__(TelegramAdapter) - adapter.name = "test_bot" - adapter.has_fatal_error = False - adapter._polling_network_error_count = 1 - adapter._polling_error_callback_ref = None - adapter._background_tasks = set() - adapter._send_path_degraded = True - # Mock the app and updater - mock_app = MagicMock() - mock_updater = AsyncMock() - mock_app.updater = mock_updater - adapter._app = mock_app +async def _hang_forever(**kwargs): + await asyncio.sleep(1000) - # Make start_polling() hang indefinitely (simulate the bug) - async def hanging_start_polling(**kwargs): - await asyncio.sleep(1000) # Hang for a long time - return None - mock_updater.start_polling = hanging_start_polling - mock_updater.running = True +def _bare_adapter(): + a = TelegramAdapter.__new__(TelegramAdapter) + # `name` / `has_fatal_error` are read-only base-class properties; set the + # backing fields they derive from instead. + from gateway.config import Platform - # Mock _drain_polling_connections to avoid actual connection cleanup - with patch.object(adapter, '_drain_polling_connections', new=AsyncMock()): - # Trigger the network error handler - task = asyncio.create_task(adapter._handle_polling_network_error(Exception("test"))) + a.platform = Platform.TELEGRAM + a._fatal_error_code = None + a._fatal_error_message = None + a._fatal_error_retryable = True + a._polling_network_error_count = 0 + a._polling_conflict_count = 0 + a._polling_error_callback_ref = None + a._background_tasks = set() + a._send_path_degraded = False + return a - # Wait for the timeout to trigger (start_polling_timeout is 30s) - try: - await asyncio.wait_for(task, timeout=_UPDATER_START_TIMEOUT + 5) - except asyncio.TimeoutError: - task.cancel() - pytest.fail("Network error handler did not complete within timeout") - # The task should have completed (either with success or error) - # The important part is that it didn't hang forever - assert task.done() +@pytest.mark.asyncio +async def test_network_ladder_start_polling_hang_does_not_wedge(monkeypatch): + """A hung start_polling() in _handle_polling_network_error must time out + and advance the ladder instead of blocking forever (#59614 core repro).""" + monkeypatch.setattr(tg_adapter, "_UPDATER_START_TIMEOUT", 0.2) + a = _bare_adapter() + a._polling_network_error_count = 0 # attempt 1 → 5s backoff before start_polling - @pytest.mark.asyncio - async def test_start_polling_success_returns_normally(self): - """When start_polling() succeeds quickly, it should return normally.""" - from hermes.plugins.platforms.telegram.adapter import TelegramAdapter + app = MagicMock() + app.updater = AsyncMock() + app.updater.start_polling = _hang_forever + app.updater.running = False + a._app = app - # Mock the adapter's internal state - adapter = TelegramAdapter.__new__(TelegramAdapter) - adapter.name = "test_bot" - adapter.has_fatal_error = False - adapter._polling_network_error_count = 1 - adapter._polling_error_callback_ref = None - adapter._background_tasks = set() - adapter._send_path_degraded = True + with patch.object(a, "_drain_polling_connections", new=AsyncMock()), \ + patch.object( + tg_adapter.asyncio, "ensure_future", + side_effect=lambda coro: (coro.close(), asyncio.get_event_loop().create_future())[1], + ): + # Unbounded, this await hangs past the 30s wait_for and fails the + # test; bounded, the handler waits its 5s backoff, times out the hung + # start_polling() in 0.2s, schedules the chained retry (captured by + # the ensure_future patch), and returns. + await asyncio.wait_for( + a._handle_polling_network_error(Exception("net down")), timeout=30 + ) - # Mock the app and updater - mock_app = MagicMock() - mock_updater = AsyncMock() - mock_app.updater = mock_updater - adapter._app = mock_app - # Make start_polling() succeed immediately - mock_updater.start_polling = AsyncMock(return_value=None) - mock_updater.running = True +@pytest.mark.asyncio +async def test_bootstrap_start_polling_hang_schedules_recovery(monkeypatch): + """_start_polling_resilient: a hung bootstrap start_polling() must raise + TimeoutError (an OSError → classified as network error) and schedule + background recovery instead of blocking connect() forever.""" + monkeypatch.setattr(tg_adapter, "_UPDATER_START_TIMEOUT", 0.2) + a = _bare_adapter() - # Mock _drain_polling_connections and _verify_polling_after_reconnect - with patch.object(adapter, '_drain_polling_connections', new=AsyncMock()), \ - patch.object(adapter, '_verify_polling_after_reconnect', new=AsyncMock()): - # Trigger the network error handler - await adapter._handle_polling_network_error(Exception("test")) + app = MagicMock() + app.updater = AsyncMock() + app.updater.start_polling = _hang_forever + a._app = app - # Verify that start_polling was called - mock_updater.start_polling.assert_called_once() \ No newline at end of file + scheduled = [] + monkeypatch.setattr( + a, "_schedule_polling_recovery", + lambda err, reason: scheduled.append((err, reason)), + raising=False, + ) + + ok = await asyncio.wait_for( + a._start_polling_resilient(drop_pending_updates=False, error_callback=None), + timeout=10, + ) + assert ok is False + assert len(scheduled) == 1 + assert isinstance(scheduled[0][0], (TimeoutError, asyncio.TimeoutError)) + + +@pytest.mark.asyncio +async def test_start_polling_success_path_unaffected(monkeypatch): + """Sanity: a fast start_polling() still returns True through the wrapper.""" + monkeypatch.setattr(tg_adapter, "_UPDATER_START_TIMEOUT", 5.0) + a = _bare_adapter() + + app = MagicMock() + app.updater = AsyncMock() + app.updater.start_polling = AsyncMock(return_value=None) + a._app = app + + ok = await a._start_polling_resilient(drop_pending_updates=False, error_callback=None) + assert ok is True + app.updater.start_polling.assert_awaited_once() + + +def test_every_start_polling_call_site_is_time_bounded(): + """Bug-class contract: every `updater.start_polling(` await in the adapter + must be wrapped in asyncio.wait_for. A new unbounded call site reintroduces + the #59614 wedge.""" + import inspect + import re + + src = inspect.getsource(tg_adapter) + # Find each start_polling( call and check an enclosing wait_for within the + # preceding 6 lines (the wrapper always sits directly above). + lines = src.splitlines() + unbounded = [] + for i, line in enumerate(lines): + if re.search(r"updater\.start_polling\(", line) and "def " not in line: + window = "\n".join(lines[max(0, i - 6):i + 1]) + if "wait_for" not in window: + unbounded.append((i + 1, line.strip())) + assert not unbounded, f"unbounded start_polling() call sites: {unbounded}"