Telegram polling silent-failure detection and watchdogs

Fixes on record

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

• #52744 fix(telegram): persistent heartbeat loop to detect CLOSE-WAIT polling sockets (#48495)

  merged to main

  files: plugins/platforms/telegram/adapter.py, scripts/release.py, tests/gateway/test_telegram_conflict.py, tests/gateway/test_telegram_network_reconnect.py

  Show the diff

  • plugins/platforms/telegram/adapter.py

    @@ -537,6 +537,7 @@ def __init__(self, config: PlatformConfig):
             self._polling_conflict_count: int = 0
             self._polling_network_error_count: int = 0
             self._polling_error_callback_ref = None
    +        self._polling_heartbeat_task: Optional[asyncio.Task] = None
             # After sustained reconnect storms the PTB httpx pool can return
             # SendResult(success=True) for sends that never actually transmit.
             # _handle_polling_network_error sets this; _verify_polling_after_reconnect
    @@ -1714,6 +1715,63 @@ async def _handle_polling_network_error(self, error: Exception) -> None:
                     self._background_tasks.add(task)
                     task.add_done_callback(self._background_tasks.discard)
     
    +    async def _polling_heartbeat_loop(self) -> None:
    +        """Detect dead Telegram TCP sockets (CLOSE-WAIT) by periodic probing.
    +
    +        PTB's long-poll task blocks on epoll waiting for Telegram to push an
    +        update.  When the underlying TCP connection enters CLOSE-WAIT (the remote
    +        sent a FIN but the httpx pool has not yet noticed), epoll still reports
    +        the socket as readable and no exception is raised — so PTB's
    +        ``error_callback`` never fires and the gateway silently stops receiving
    +        messages.
    +
    +        This loop probes ``get_me()`` every ``HEARTBEAT_INTERVAL`` seconds on the
    +        *general* request path (not the getUpdates pool), so a healthy long-poll
    +        waiting for the 30-second Telegram window is never interrupted.  On any
    +        connect-level failure the loop hands off to
    +        ``_handle_polling_network_error`` — the same path triggered by PTB's own
    +        ``error_callback`` — which drains the dead pool and restarts polling.
    +
    +        Unlike ``_verify_polling_after_reconnect`` (a one-shot probe scheduled
    +        only after an explicit reconnect), this loop runs for the full lifetime
    +        of the polling connection, so it catches a socket that wedges during
    +        steady-state operation without any prior error event.
    +        """
    +        HEARTBEAT_INTERVAL = 90   # seconds between probes
    +        PROBE_TIMEOUT = 15        # seconds before declaring the path dead
    +
    +        while True:
    +            try:
    +                await asyncio.sleep(HEARTBEAT_INTERVAL)
    +                if self.has_fatal_error:
    +                    return
    +                bot = self._app.bot if self._app else None
    +                if bot is None:
    +                    continue
    +                # A real PTB Bot always exposes get_me(); if it's absent the
    +                # app isn't a live polling client (e.g. torn down or a test
    +                # double), so there is nothing to probe — exit rather than spin.
    +                if not callable(getattr(bot, "get_me", None)):
    +                    return
    +                await asyncio.wait_for(bot.get_me(), PROBE_TIMEOUT)
    +            except asyncio.CancelledError:
    +                return
    +            except (asyncio.TimeoutError, OSError) as probe_err:
    +                logger.warning(
    +                    "[%s] Polling heartbeat probe failed (%s); triggering reconnect",
    +                    self.name, probe_err,
    +                )
    +                if self._polling_error_task and not self._polling_error_task.done():
    +                    continue   # reconnect already in progress
    +                loop = asyncio.get_running_loop()
    +                self._polling_error_task = loop.create_task(
    +                    self._handle_polling_network_error(probe_err)
    +                )
    +            except Exception:
    +                # Non-connectivity errors (e.g. TelegramError 401) are not
    +                # CLOSE-WAIT symptoms — let PTB's own handlers surface them.
    +                pass
    +
         async def _verify_polling_after_reconnect(self) -> None:
             """Heartbeat probe scheduled after a successful reconnect.
     
    @@ -2505,6 +2563,16 @@ def _polling_error_callback(error: Exception) -> None:
                 mode = "webhook" if self._webhook_mode else "polling"
                 logger.info("[%s] Connected to Telegram (%s mode)", self.name, mode)
     
    +            # Start the persistent heartbeat loop in polling mode. Webhook mode
    +            # receives updates via incoming pushes — there is no long-poll
    +            # socket to wedge in CLOSE-WAIT, so the loop is not needed there.
    +            if not self._webhook_mode:
    +                if self._polling_heartbeat_task and not self._polling_heartbeat_task.done():
    +                    self._polling_heartbeat_task.cancel()
    +                self._polling_heartbeat_task = asyncio.ensure_future(
    +                    self._polling_heartbeat_loop()
    +                )
    +
                 # Surface the gateway as "Online" in the bot's short description
                 # (opt-in via extra.status_indicator). Non-fatal.
                 try:
    @@ -2563,6 +2631,16 @@ async def _set_status_indicator(self, online: bool) -> None:
     
         async def disconnect(self) -> None:
             """Stop polling/webhook, cancel pending album flushes, and disconnect."""
    +        # Cancel the heartbeat before tearing down the app so the probe task
    +        # cannot fire get_me() into a half-shutdown bot client.
    +        if self._polling_heartbeat_task and not self._polling_heartbeat_task.done():
    +            self._polling_heartbeat_task.cancel()
    +            try:
    +                await self._polling_heartbeat_task
    +            except asyncio.CancelledError:
    +                pass
    +        self._polling_heartbeat_task = None
    +
             # Mark the bot "Offline" in its short description while the bot's HTTP
             # client is still alive (before app shutdown closes it). Opt-in via
             # extra.status_indicator. Non-fatal. This is the clean-shutdown path;

  • scripts/release.py

    @@ -45,6 +45,7 @@
     
     # Auto-extracted from noreply emails + manual overrides
     AUTHOR_MAP = {
    +    "[redacted:email]": "agt-user",  # PR #48496 salvage (telegram CLOSE-WAIT polling heartbeat, #48495)
         "[redacted:email]": "herbalizer404",  # PR #49076 + #51835 salvage (auxiliary compression fallback: 403/session-usage payment errors + honor fallback chain when aux provider auth unavailable)
         "[redacted:email]": "pyxl-dev",  # PR #52230 salvage (include rate-limit in auxiliary capacity-error fallback gate; #52228)
         "[redacted:email]": "benbenlijie",  # PR #47205 salvage (named custom-provider extra_body + Z.AI Coding overload adaptive backoff; #50663)

  • tests/gateway/test_telegram_conflict.py

    @@ -47,6 +47,25 @@ async def _noop():
         monkeypatch.setattr("plugins.platforms.telegram.adapter.HTTPXRequest", lambda **kwargs: MagicMock())
     
     
    +async def _cancel_heartbeat(adapter):
    +    """Cancel the lifetime heartbeat task connect() starts in polling mode.
    +
    +    These tests call the real connect() but never disconnect(), so the
    +    _polling_heartbeat_loop task would otherwise outlive the test. With
    +    asyncio.sleep monkeypatched to instant, leaving it running busy-spins the
    +    event loop and starves the test (CI per-file timeout). disconnect() does
    +    this in production; tests that only connect() must do it themselves.
    +    """
    +    task = getattr(adapter, "_polling_heartbeat_task", None)
    +    if task and not task.done():
    +        task.cancel()
    +        try:
    +            await task
    +        except (asyncio.CancelledError, Exception):
    +            pass
    +    adapter._polling_heartbeat_task = None
    +
    +
     @pytest.mark.asyncio
     async def test_connect_rejects_same_host_token_lock(monkeypatch):
         adapter = TelegramAdapter(PlatformConfig(enabled=True, token="[redacted:secret-assignment]"))
    @@ -127,6 +146,11 @@ async def fake_start_polling(**kwargs):
         assert adapter.has_fatal_error is False, "First conflict should not be fatal"
         assert adapter._polling_conflict_count == 0, "Count should reset after successful retry"
     
    +    # connect() now starts a lifetime _polling_heartbeat_loop task. With
    +    # asyncio.sleep mocked to instant above, it must not be left running or it
    +    # busy-spins on the event loop and starves the test. Cancel it explicitly.
    +    await _cancel_heartbeat(adapter)
    +
     
     @pytest.mark.asyncio
     async def test_polling_conflict_becomes_fatal_after_retries(monkeypatch):
    @@ -205,6 +229,7 @@ async def failing_start_polling(**kwargs):
         )
         assert adapter.has_fatal_error is True
         fatal_handler.assert_awaited_once()
    +    await _cancel_heartbeat(adapter)
     
     
     @pytest.mark.asyncio
    @@ -285,6 +310,7 @@ async def test_connect_clears_webhook_before_polling(monkeypatch):
     
         assert ok is True
         bot.delete_webhook.assert_awaited_once_with(drop_pending_updates=False)
    +    await _cancel_heartbeat(adapter)
     
     
     @pytest.mark.asyncio
    @@ -398,3 +424,4 @@ def _boom():
             await adapter._polling_error_task
         except (asyncio.CancelledError, Exception):
             pass
    +    await _cancel_heartbeat(adapter)

  • tests/gateway/test_telegram_network_reconnect.py

    @@ -473,3 +473,211 @@ async def test_reconnect_schedules_heartbeat_probe_on_success():
                 await t
             except (asyncio.CancelledError, Exception):
                 pass
    +
    +
    +# ── Persistent heartbeat loop (_polling_heartbeat_loop) ──────────────────────
    +#
    +# These tests cover the continuous CLOSE-WAIT detection loop that fixes the bug
    +# (#48495) where a dead Telegram TCP socket caused the gateway to stop receiving
    +# messages silently. The _verify_polling_after_reconnect tests above cover the
    +# one-shot post-reconnect probe; these cover the background loop that runs for
    +# the gateway's full lifetime in polling mode.
    +#
    +# Loop structure: while True: sleep(INTERVAL) → fatal/app checks → get_me().
    +# So with cancel raised on the Nth patched sleep, get_me() fires (N-1) times.
    +
    +
    [redacted:email]
    +async def test_heartbeat_loop_exits_cleanly_on_cancel():
    +    """The heartbeat loop must exit without raising when cancelled (normal shutdown)."""
    +    adapter = _make_adapter()
    +
    +    mock_app = MagicMock()
    +    mock_app.bot.get_me = AsyncMock(return_value=MagicMock())
    +    adapter._app = mock_app
    +
    +    sleep_count = 0
    +
    +    async def fast_sleep(seconds):
    +        nonlocal sleep_count
    +        sleep_count += 1
    +        # sleep #1 → get_me, sleep #2 → get_me, sleep #3 → cancel.
    +        if sleep_count >= 3:
    +            raise asyncio.CancelledError()
    +
    +    with patch("asyncio.sleep", side_effect=fast_sleep):
    +        # Should not raise — CancelledError is swallowed internally.
    +        await adapter._polling_heartbeat_loop()
    +
    +    assert mock_app.bot.get_me.await_count == 2
    +
    +
    [redacted:email]
    +async def test_heartbeat_loop_triggers_reconnect_on_timeout():
    +    """A TimeoutError from get_me() must schedule a reconnect via _handle_polling_network_error."""
    +    adapter = _make_adapter()
    +    adapter._handle_polling_network_error = AsyncMock()
    +
    +    mock_app = MagicMock()
    +    adapter._app = mock_app
    +
    +    sleep_call = 0
    +
    +    async def fast_sleep(seconds):
    +        nonlocal sleep_call
    +        sleep_call += 1
    +        if sleep_call >= 3:
    +            raise asyncio.CancelledError()
    +
    +    async def fast_wait_for(coro, timeout):
    +        if asyncio.iscoroutine(coro):
    +            coro.close()
    +        raise asyncio.TimeoutError()
    +
    +    with patch("asyncio.sleep", side_effect=fast_sleep):
    +        with patch("plugins.platforms.telegram.adapter.asyncio.wait_for", side_effect=fast_wait_for):
    +            await adapter._polling_heartbeat_loop()
    +
    +    # A reconnect task must have been created.
    +    assert adapter._polling_error_task is not None
    +
    +
    [redacted:email]
    +async def test_heartbeat_loop_triggers_reconnect_on_os_error():
    +    """An OSError (e.g. connection reset) from get_me() must trigger a reconnect."""
    +    adapter = _make_adapter()
    +    adapter._handle_polling_network_error = AsyncMock()
    +
    +    mock_app = MagicMock()
    +    adapter._app = mock_app
    +
    +    sleep_call = 0
    +
    +    async def fast_sleep(seconds):
    +        nonlocal sleep_call
    +        sleep_call += 1
    +        if sleep_call >= 3:
    +            raise asyncio.CancelledError()
    +
    +    async def os_error_wait_for(coro, timeout):
    +        if asyncio.iscoroutine(coro):
    +            coro.close()
    +        raise OSError("Connection reset by peer")
    +
    +    with patch("asyncio.sleep", side_effect=fast_sleep):
    +        with patch("plugins.platforms.telegram.adapter.asyncio.wait_for", side_effect=os_error_wait_for):
    +            await adapter._polling_heartbeat_loop()
    +
    +    assert adapter._polling_error_task is not None
    +
    +
    [redacted:email]
    +async def test_heartbeat_loop_skips_reconnect_if_already_in_progress():
    +    """If a reconnect task is already running, the heartbeat must not spawn another."""
    +    adapter = _make_adapter()
    +
    +    # Simulate an already-running reconnect task.
    +    existing_task = asyncio.get_event_loop().create_task(asyncio.sleep(3600))
    +    adapter._polling_error_task = existing_task
    +    adapter._handle_polling_network_error = AsyncMock()
    +
    +    mock_app = MagicMock()
    +    adapter._app = mock_app
    +
    +    sleep_call = 0
    +
    +    async def fast_sleep(seconds):
    +        nonlocal sleep_call
    +        sleep_call += 1
    +        if sleep_call >= 3:
    +            raise asyncio.CancelledError()
    +
    +    async def timeout_wait_for(coro, timeout):
    +        if asyncio.iscoroutine(coro):
    +            coro.close()
    +        raise asyncio.TimeoutError()
    +
    +    with patch("asyncio.sleep", side_effect=fast_sleep):
    +        with patch("plugins.platforms.telegram.adapter.asyncio.wait_for", side_effect=timeout_wait_for):
    +            await adapter._polling_heartbeat_loop()
    +
    +    # _handle_polling_network_error must NOT have been called — existing task still running.
    +    adapter._handle_polling_network_error.assert_not_awaited()
    +
    +    existing_task.cancel()
    +    try:
    +        await existing_task
    +    except (asyncio.CancelledError, Exception):
    +        pass
    +
    +
    [redacted:email]
    +async def test_heartbeat_loop_ignores_non_connectivity_errors():
    +    """Errors that are not connectivity failures (e.g. TelegramError) must be swallowed."""
    +    adapter = _make_adapter()
    +    adapter._handle_polling_network_error = AsyncMock()
    +
    +    mock_app = MagicMock()
    +    adapter._app = mock_app
    +
    +    sleep_call = 0
    +
    +    async def fast_sleep(seconds):
    +        nonlocal sleep_call
    +        sleep_call += 1
    +        if sleep_call >= 3:
    +            raise asyncio.CancelledError()
    +
    +    async def telegram_error_wait_for(coro, timeout):
    +        if asyncio.iscoroutine(coro):
    +            coro.close()
    +        raise RuntimeError("TelegramError: Unauthorized")  # non-OSError, non-TimeoutError
    +
    +    with patch("asyncio.sleep", side_effect=fast_sleep):
    +        with patch("plugins.platforms.telegram.adapter.asyncio.wait_for", side_effect=telegram_error_wait_for):
    +            await adapter._polling_heartbeat_loop()
    +
    +    # No reconnect should have been triggered for a non-connectivity error.
    +    adapter._handle_polling_network_error.assert_not_awaited()
    +
    +
    [redacted:email]
    +async def test_heartbeat_loop_exits_on_fatal_error():
    +    """A fatal error short-circuits the loop before probing get_me()."""
    +    adapter = _make_adapter()
    +    adapter._set_fatal_error("telegram_network_error", "boom", retryable=True)
    +
    +    mock_app = MagicMock()
    +    mock_app.bot.get_me = AsyncMock(return_value=MagicMock())
    +    adapter._app = mock_app
    +
    +    async def fast_sleep(seconds):
    +        return None
    +
    +    with patch("asyncio.sleep", side_effect=fast_sleep):
    +        await adapter._polling_heartbeat_loop()
    +
    +    # Fatal error returns before the get_me() probe.
    +    mock_app.bot.get_me.assert_not_awaited()
    +
    +
    [redacted:email]
    +async def test_disconnect_cancels_heartbeat_task():
    +    """disconnect() must cancel the heartbeat task before shutting down the app."""
    +    adapter = _make_adapter()
    +
    +    # Simulate a running heartbeat.
    +    heartbeat_task = asyncio.get_event_loop().create_task(asyncio.sleep(3600))
    +    adapter._polling_heartbeat_task = heartbeat_task
    +
    +    mock_app = MagicMock()
    +    mock_app.updater = MagicMock()
    +    mock_app.updater.running = False
    +    mock_app.running = False
    +    mock_app.shutdown = AsyncMock()
    +    adapter._app = mock_app
    +
    +    await adapter.disconnect()
    +
    +    assert heartbeat_task.cancelled(), "Heartbeat task must be cancelled by disconnect()"
    +    assert adapter._polling_heartbeat_task is None