Telegram start_polling() hang and timeout/watchdog proposals

Fixes on record

Merged to main. A stamp counts the issues a fix is linked to.

• #60137 fix(telegram): bound start_polling() with a watchdog timeout at all three call sites (#59614)

  merged to main

  files: plugins/platforms/telegram/adapter.py, tests/gateway/test_telegram_start_polling_timeout.py

  Show the diff

  • plugins/platforms/telegram/adapter.py

    @@ -416,6 +416,12 @@ def _rich_normalize_linebreaks(text: str) -> str:
     # 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
    +# start_polling() can also hang when the connection pool is in a degraded state
    +# after _drain_polling_connections(), particularly when both primary and fallback
    +# Telegram endpoints are unreachable. Bounding start_polling() prevents the
    +# reconnect ladder from stalling indefinitely and allows the heartbeat loop to
    +# trigger its own recovery path. Refs: NousResearch/hermes-agent#59614
    +_UPDATER_START_TIMEOUT = 30.0
     
     
     class TelegramAdapter(BasePlatformAdapter):
    @@ -1973,10 +1979,19 @@ async def _start_polling_resilient(self, *, drop_pending_updates: bool, error_ca
             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:
    @@ -2072,11 +2087,26 @@ async def _handle_polling_network_error(self, error: Exception) -> None:
             try:
                 if not app:
                     raise RuntimeError("Telegram application was torn down during reconnect")
    -            await app.updater.start_polling(
    -                allowed_updates=Update.ALL_TYPES,
    -                drop_pending_updates=False,
    -                error_callback=self._polling_error_callback_ref,
    -            )
    +            # Guard start_polling() with a timeout: when the connection pool is
    +            # in a degraded state (e.g., after _drain_polling_connections()), the
    +            # httpx client may hold a stale socket that neither connects nor times
    +            # out within PTB's internal flow. Bounding start_polling() prevents
    +            # the reconnect ladder from stalling indefinitely and allows the
    +            # heartbeat loop to trigger its own recovery path.
    +            # Refs: NousResearch/hermes-agent#59614
    +            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 network error (attempt %d)",
                     self.name, attempt,
    @@ -2467,11 +2497,24 @@ async def _handle_polling_conflict(self, error: Exception) -> None:
                 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,

  • tests/gateway/test_telegram_start_polling_timeout.py

    @@ -0,0 +1,157 @@
    +"""Regression tests for #59614: start_polling() must be time-bounded.
    +
    +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
    +
    +
    +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)
    +
    +
    +_ensure_telegram_mock()
    +
    +from plugins.platforms.telegram import adapter as tg_adapter  # noqa: E402
    +from plugins.platforms.telegram.adapter import TelegramAdapter  # noqa: E402
    +
    +
    +async def _hang_forever(**kwargs):
    +    await asyncio.sleep(1000)
    +
    +
    +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
    +
    +    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
    +
    +
    [redacted:email]
    +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
    +
    +    app = MagicMock()
    +    app.updater = AsyncMock()
    +    app.updater.start_polling = _hang_forever
    +    app.updater.running = False
    +    a._app = app
    +
    +    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
    +        )
    +
    +
    [redacted:email]
    +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()
    +
    +    app = MagicMock()
    +    app.updater = AsyncMock()
    +    app.updater.start_polling = _hang_forever
    +    a._app = app
    +
    +    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))
    +
    +
    [redacted:email]
    +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}"