Telegram polling reconnect hang and wedge fixes

Fixes on record

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

• #3268 fix(telegram): self-reschedule reconnect when start_polling fails after 502

  merged to main

  files: gateway/platforms/telegram.py, tests/gateway/test_telegram_network_reconnect.py

• #18751 fix(telegram): probe polling liveness after reconnect to detect wedged Updater (salvage #18088)

  merged to main

  files: gateway/platforms/telegram.py, tests/gateway/test_telegram_network_reconnect.py

  Show the diff

  • gateway/platforms/telegram.py

    @@ -512,6 +512,17 @@ async def _handle_polling_network_error(self, error: Exception) -> None:
                     self.name, attempt,
                 )
                 self._polling_network_error_count = 0
    +            # start_polling() returning is necessary but not sufficient:
    +            # PTB's Updater can be left in a state where `running` is True
    +            # but the underlying long-poll task is wedged on a stale httpx
    +            # connection and never makes progress. No error_callback fires
    +            # in that state, so the reconnect ladder won't advance on its
    +            # own. Schedule a deferred probe to detect the wedge and
    +            # re-enter the ladder if needed.
    +            if not self.has_fatal_error:
    +                probe = asyncio.ensure_future(self._verify_polling_after_reconnect())
    +                self._background_tasks.add(probe)
    +                probe.add_done_callback(self._background_tasks.discard)
             except Exception as retry_err:
                 logger.warning("[%s] Telegram polling reconnect failed: %s", self.name, retry_err)
                 # start_polling failed — polling is dead and no further error
    @@ -523,6 +534,50 @@ 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 _verify_polling_after_reconnect(self) -> None:
    +        """Heartbeat probe scheduled after a successful reconnect.
    +
    +        PTB's Updater can survive a botched stop()+start_polling() cycle
    +        with `running=True` but a wedged consumer task. No error callback
    +        fires, so the reconnect ladder doesn't advance on its own. This
    +        probe detects the wedge by:
    +
    +        1. Sleeping HEARTBEAT_PROBE_DELAY so a healthy long-poll has time
    +           to complete at least one cycle.
    +        2. Verifying `Updater.running` is still True.
    +        3. Probing the bot endpoint with a tight asyncio timeout. A
    +           wedged httpx pool fails this probe; a healthy one returns
    +           well under the timeout.
    +
    +        On any failure, re-enter the reconnect ladder so the existing
    +        MAX_NETWORK_RETRIES path can ultimately escalate to fatal-error.
    +        """
    +        HEARTBEAT_PROBE_DELAY = 60
    +        PROBE_TIMEOUT = 10
    +
    +        await asyncio.sleep(HEARTBEAT_PROBE_DELAY)
    +
    +        if self.has_fatal_error:
    +            return
    +        if not (self._app and self._app.updater and self._app.updater.running):
    +            logger.warning(
    +                "[%s] Updater not running %ds after reconnect — treating as wedged",
    +                self.name, HEARTBEAT_PROBE_DELAY,
    +            )
    +            await self._handle_polling_network_error(
    +                RuntimeError("Updater not running after reconnect heartbeat")
    +            )
    +            return
    +
    +        try:
    +            await asyncio.wait_for(self._app.bot.get_me(), PROBE_TIMEOUT)
    +        except Exception as probe_err:
    +            logger.warning(
    +                "[%s] Polling heartbeat probe failed %ds after reconnect: %s",
    +                self.name, HEARTBEAT_PROBE_DELAY, probe_err,
    +            )
    +            await self._handle_polling_network_error(probe_err)
    +
         async def _handle_polling_conflict(self, error: Exception) -> None:
             if self.has_fatal_error and self.fatal_error_code == "telegram_polling_conflict":
                 return

  • tests/gateway/test_telegram_network_reconnect.py

    @@ -132,6 +132,7 @@ async def test_reconnect_success_resets_error_count():
     
         mock_app = MagicMock()
         mock_app.updater = mock_updater
    +    mock_app.bot.get_me = AsyncMock(return_value=MagicMock())  # heartbeat probe path
         adapter._app = mock_app
     
         with patch("asyncio.sleep", new_callable=AsyncMock):
    @@ -139,6 +140,15 @@ async def test_reconnect_success_resets_error_count():
     
         assert adapter._polling_network_error_count == 0
     
    +    # Clean up the heartbeat-probe task scheduled after a successful reconnect.
    +    pending = [t for t in adapter._background_tasks if not t.done()]
    +    for t in pending:
    +        t.cancel()
    +        try:
    +            await t
    +        except (asyncio.CancelledError, Exception):
    +            pass
    +
     
     @pytest.mark.asyncio
     async def test_reconnect_triggers_fatal_after_max_retries():
    @@ -284,3 +294,182 @@ async def test_drain_helper_noop_without_app():
         adapter._app = None
         # Should not raise
         await adapter._drain_polling_connections()
    +
    +
    +# ── Heartbeat probe ──────────────────────────────────────────────────────
    +
    +
    [redacted:email]
    +async def test_heartbeat_probe_no_op_when_polling_healthy():
    +    """
    +    Probe scheduled after a successful reconnect: Updater.running=True and
    +    bot.get_me() returns quickly → recovery confirmed, no further action.
    +    """
    +    adapter = _make_adapter()
    +
    +    mock_updater = MagicMock()
    +    mock_updater.running = True
    +
    +    mock_app = MagicMock()
    +    mock_app.updater = mock_updater
    +    mock_app.bot.get_me = AsyncMock(return_value=MagicMock())
    +    adapter._app = mock_app
    +
    +    adapter._handle_polling_network_error = AsyncMock()
    +
    +    with patch("asyncio.sleep", new_callable=AsyncMock):
    +        await adapter._verify_polling_after_reconnect()
    +
    +    mock_app.bot.get_me.assert_awaited_once()
    +    adapter._handle_polling_network_error.assert_not_awaited()
    +
    +
    [redacted:email]
    +async def test_heartbeat_probe_reenters_ladder_when_updater_not_running():
    +    """
    +    If Updater.running has flipped to False by the heartbeat delay, treat
    +    as wedged: re-enter the reconnect ladder.
    +    """
    +    adapter = _make_adapter()
    +
    +    mock_updater = MagicMock()
    +    mock_updater.running = False
    +
    +    mock_app = MagicMock()
    +    mock_app.updater = mock_updater
    +    mock_app.bot.get_me = AsyncMock()
    +    adapter._app = mock_app
    +
    +    adapter._handle_polling_network_error = AsyncMock()
    +
    +    with patch("asyncio.sleep", new_callable=AsyncMock):
    +        await adapter._verify_polling_after_reconnect()
    +
    +    mock_app.bot.get_me.assert_not_called()
    +    adapter._handle_polling_network_error.assert_awaited_once()
    +    err = adapter._handle_polling_network_error.await_args.args[0]
    +    assert isinstance(err, RuntimeError)
    +    assert "not running" in str(err).lower()
    +
    +
    [redacted:email]
    +async def test_heartbeat_probe_reenters_ladder_when_get_me_times_out():
    +    """
    +    If bot.get_me() hangs longer than PROBE_TIMEOUT, treat as wedged.
    +    Simulates the connection-pool wedge that motivated this fix.
    +    """
    +    adapter = _make_adapter()
    +
    +    mock_updater = MagicMock()
    +    mock_updater.running = True
    +
    +    async def hang_forever(*args, **kwargs):
    +        await asyncio.sleep(3600)
    +
    +    mock_app = MagicMock()
    +    mock_app.updater = mock_updater
    +    mock_app.bot.get_me = AsyncMock(side_effect=hang_forever)
    +    adapter._app = mock_app
    +
    +    adapter._handle_polling_network_error = AsyncMock()
    +
    +    async def fast_wait_for(coro, timeout):
    +        if asyncio.iscoroutine(coro):
    +            coro.close()
    +        raise asyncio.TimeoutError()
    +
    +    with patch("asyncio.sleep", new_callable=AsyncMock):
    +        with patch("gateway.platforms.telegram.asyncio.wait_for", new=fast_wait_for):
    +            await adapter._verify_polling_after_reconnect()
    +
    +    adapter._handle_polling_network_error.assert_awaited_once()
    +
    +
    [redacted:email]
    +async def test_heartbeat_probe_reenters_ladder_on_get_me_network_error():
    +    """
    +    Any exception raised by bot.get_me() (NetworkError, ConnectionError, etc.)
    +    should re-enter the reconnect ladder with the original exception.
    +    """
    +    adapter = _make_adapter()
    +
    +    mock_updater = MagicMock()
    +    mock_updater.running = True
    +
    +    mock_app = MagicMock()
    +    mock_app.updater = mock_updater
    +    mock_app.bot.get_me = AsyncMock(side_effect=ConnectionError("pool wedged"))
    +    adapter._app = mock_app
    +
    +    adapter._handle_polling_network_error = AsyncMock()
    +
    +    with patch("asyncio.sleep", new_callable=AsyncMock):
    +        await adapter._verify_polling_after_reconnect()
    +
    +    adapter._handle_polling_network_error.assert_awaited_once()
    +    assert isinstance(
    +        adapter._handle_polling_network_error.await_args.args[0], ConnectionError
    +    )
    +
    +
    [redacted:email]
    +async def test_heartbeat_probe_skips_when_already_fatal():
    +    """
    +    If the adapter is already in fatal-error state by the time the probe
    +    delay elapses, the probe should bail without further action.
    +    """
    +    adapter = _make_adapter()
    +    adapter._set_fatal_error("telegram_polling_conflict", "already fatal", retryable=False)
    +
    +    mock_app = MagicMock()
    +    mock_app.bot.get_me = AsyncMock()
    +    adapter._app = mock_app
    +
    +    adapter._handle_polling_network_error = AsyncMock()
    +
    +    with patch("asyncio.sleep", new_callable=AsyncMock):
    +        await adapter._verify_polling_after_reconnect()
    +
    +    mock_app.bot.get_me.assert_not_called()
    +    adapter._handle_polling_network_error.assert_not_awaited()
    +
    +
    [redacted:email]
    +async def test_reconnect_schedules_heartbeat_probe_on_success():
    +    """
    +    After a successful start_polling() in the reconnect path, a probe task
    +    must be added to _background_tasks. Without it, a wedged Updater would
    +    sit silent indefinitely with no further error_callback to advance the
    +    reconnect ladder.
    +    """
    +    adapter = _make_adapter()
    +    adapter._polling_network_error_count = 1
    +
    +    mock_updater = MagicMock()
    +    mock_updater.running = True
    +    mock_updater.stop = AsyncMock()
    +    mock_updater.start_polling = AsyncMock()  # succeeds
    +
    +    mock_app = MagicMock()
    +    mock_app.updater = mock_updater
    +    mock_app.bot.get_me = AsyncMock(return_value=MagicMock())
    +    adapter._app = mock_app
    +
    +    initial_count = len(adapter._background_tasks)
    +
    +    with patch("asyncio.sleep", new_callable=AsyncMock):
    +        await adapter._handle_polling_network_error(Exception("Bad Gateway"))
    +
    +    assert len(adapter._background_tasks) > initial_count, (
    +        "Expected a heartbeat probe task to be scheduled after a successful "
    +        "reconnect's start_polling()"
    +    )
    +
    +    # Clean up.
    +    pending = [t for t in adapter._background_tasks if not t.done()]
    +    for t in pending:
    +        t.cancel()
    +        try:
    +            await t
    +        except (asyncio.CancelledError, Exception):
    +            pass

• #56200 fix(telegram): close reconnect races that leave the adapter half-destroyed (#55992)

  merged to main

  files: gateway/run.py, plugins/platforms/telegram/adapter.py, tests/gateway/test_runner_fatal_adapter.py, tests/gateway/test_telegram_network_reconnect.py

  Show the diff

  • gateway/run.py

    @@ -3654,6 +3654,23 @@ async def _handle_adapter_fatal_error(self, adapter: BasePlatformAdapter) -> Non
             If the error is retryable (e.g. network blip, DNS failure), queue the
             platform for background reconnection instead of giving up permanently.
             """
    +        # Snapshot the current owner of this platform slot before doing
    +        # anything else. If it's neither this adapter nor empty, a different
    +        # adapter has already taken over (e.g. this is a delayed notification
    +        # from a background retry chain that raced with, and lost to, a
    +        # reconnect that already succeeded). Acting on a stale notification
    +        # would overwrite an already-healthy platform's runtime status and
    +        # incorrectly re-queue it for reconnection, so bail out before any of
    +        # that happens.
    +        existing = self.adapters.get(adapter.platform)
    +        if existing is not None and existing is not adapter:
    +            logger.debug(
    +                "Ignoring stale fatal error from a superseded %s adapter instance: %s",
    +                adapter.platform.value,
    +                adapter.fatal_error_code or "unknown",
    +            )
    +            return
    +
             logger.error(
                 "Fatal %s adapter error (%s): %s",
                 adapter.platform.value,
    @@ -3677,13 +3694,15 @@ async def _handle_adapter_fatal_error(self, adapter: BasePlatformAdapter) -> Non
                 error_message=adapter.fatal_error_message,
             )
     
    -        existing = self.adapters.get(adapter.platform)
             if existing is adapter:
    -            try:
    -                await adapter.disconnect()
    -            finally:
    -                self.adapters.pop(adapter.platform, None)
    -                self.delivery_router.adapters = self.adapters
    +            # Claim this adapter for teardown before awaiting disconnect() —
    +            # a second fatal-error notification for the same adapter (e.g.
    +            # from a concurrent recovery path) would otherwise still see
    +            # itself as "existing" during the await below and disconnect()
    +            # the same object twice.
    +            self.adapters.pop(adapter.platform, None)
    +            self.delivery_router.adapters = self.adapters
    +            await adapter.disconnect()
     
             # Queue retryable failures for background reconnection
             if adapter.fatal_error_retryable:

  • plugins/platforms/telegram/adapter.py

    @@ -1773,16 +1773,24 @@ async def _handle_polling_network_error(self, error: Exception) -> None:
             )
             await asyncio.sleep(delay)
     
    +        # Capture a stable local reference: self._app can be reassigned to None
    +        # by a concurrent disconnect() while we're suspended across the awaits
    +        # below, and re-reading self._app after that point would silently swap
    +        # in None mid-sequence instead of failing fast in one place.
    +        app = self._app
    +
             try:
    -            if self._app and self._app.updater and self._app.updater.running:
    -                await self._app.updater.stop()
    +            if app and app.updater and app.updater.running:
    +                await app.updater.stop()
             except Exception:
                 pass
     
             await self._drain_polling_connections()
     
             try:
    -            await self._app.updater.start_polling(
    +            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,
    @@ -1824,6 +1832,12 @@ async def _handle_polling_network_error(self, error: Exception) -> None:
                     )
                     self._background_tasks.add(task)
                     task.add_done_callback(self._background_tasks.discard)
    +                # This chained retry IS the in-flight recovery attempt — it
    +                # must replace the reentrancy guard, otherwise the heartbeat
    +                # loop, the pending-updates probe, and the PTB error callback
    +                # all see _polling_error_task as "done" and can each start a
    +                # second, concurrent recovery for the same outage.
    +                self._polling_error_task = task
     
         async def _polling_heartbeat_loop(self) -> None:
             """Detect dead Telegram TCP sockets (CLOSE-WAIT) by periodic probing.
    @@ -2151,8 +2165,17 @@ async def _handle_polling_conflict(self, error: Exception) -> None:
                 await asyncio.sleep(RETRY_DELAY)
                 await self._drain_polling_connections()
     
    +            # Capture a stable local reference: self._app can be reassigned to
    +            # None by a concurrent disconnect() while we're suspended across
    +            # the awaits above (same race #55992 fixed on the network path).
    +            # Re-reading self._app after that point would raise
    +            # AttributeError deep inside start_polling instead of failing fast
    +            # here, where the except below reschedules or escalates to fatal.
    +            app = self._app
                 try:
    -                await self._app.updater.start_polling(
    +                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,

  • tests/gateway/test_runner_fatal_adapter.py

    @@ -1,3 +1,4 @@
    +import asyncio
     from unittest.mock import AsyncMock
     
     import pytest
    @@ -98,3 +99,96 @@ async def test_runner_queues_retryable_runtime_fatal_for_reconnection(monkeypatc
         assert runner._exit_with_failure is False
         assert Platform.WHATSAPP in runner._failed_platforms
         assert runner._failed_platforms[Platform.WHATSAPP]["attempts"] == 0
    +
    +
    [redacted:email]
    +async def test_concurrent_fatal_notifications_disconnect_same_adapter_once(monkeypatch, tmp_path):
    +    """
    +    Two fatal-error notifications for the same still-installed adapter (e.g.
    +    from two concurrent recovery paths racing on the same underlying outage)
    +    must result in exactly one disconnect() call.
    +
    +    Regression test for the TOCTOU race in _handle_adapter_fatal_error: the
    +    old code only removed the adapter from self.adapters in a `finally` block
    +    *after* awaiting disconnect(), so a second concurrent call could still see
    +    itself as "existing" and disconnect() the same object twice — the
    +    concrete origin of the "'NoneType' object has no attribute 'updater'"
    +    crash when the adapter's own teardown code re-reads self._app afterwards.
    +    """
    +    config = GatewayConfig(
    +        platforms={
    +            Platform.WHATSAPP: PlatformConfig(enabled=True, token="token")
    +        },
    +        sessions_dir=tmp_path / "sessions",
    +    )
    +    runner = GatewayRunner(config)
    +    adapter = _RuntimeRetryableAdapter()
    +    adapter._set_fatal_error(
    +        "whatsapp_bridge_exited",
    +        "WhatsApp bridge process exited unexpectedly (code 1).",
    +        retryable=True,
    +    )
    +
    +    runner.adapters = {Platform.WHATSAPP: adapter}
    +    runner.delivery_router.adapters = runner.adapters
    +    runner.stop = AsyncMock()
    +
    +    disconnect_calls = 0
    +    release_second_call = asyncio.Event()
    +
    +    async def slow_disconnect():
    +        nonlocal disconnect_calls
    +        disconnect_calls += 1
    +        # Yield control so the second concurrent notification can run its
    +        # "existing is adapter" check before this call finishes tearing down.
    +        release_second_call.set()
    +        await asyncio.sleep(0)
    +        adapter._mark_disconnected()
    +
    +    monkeypatch.setattr(adapter, "disconnect", slow_disconnect)
    +
    +    await asyncio.gather(
    +        runner._handle_adapter_fatal_error(adapter),
    +        runner._handle_adapter_fatal_error(adapter),
    +    )
    +
    +    assert disconnect_calls == 1
    +
    +
    [redacted:email]
    +async def test_stale_fatal_notification_from_superseded_adapter_is_ignored(monkeypatch, tmp_path):
    +    """
    +    A delayed fatal-error notification from an adapter instance that has
    +    since been replaced by a different, already-installed adapter (e.g. a
    +    background retry chain on the old instance finally giving up after a
    +    reconnect on a new instance already succeeded) must be ignored: it must
    +    not disconnect the new adapter, must not re-queue an already-healthy
    +    platform for reconnection, and must not shut the gateway down.
    +    """
    +    config = GatewayConfig(
    +        platforms={
    +            Platform.WHATSAPP: PlatformConfig(enabled=True, token="token")
    +        },
    +        sessions_dir=tmp_path / "sessions",
    +    )
    +    runner = GatewayRunner(config)
    +
    +    old_adapter = _RuntimeRetryableAdapter()
    +    old_adapter._set_fatal_error(
    +        "whatsapp_bridge_exited",
    +        "stale failure from a superseded adapter instance",
    +        retryable=True,
    +    )
    +
    +    new_adapter = _RuntimeRetryableAdapter()
    +    new_adapter.disconnect = AsyncMock()
    +    runner.adapters = {Platform.WHATSAPP: new_adapter}
    +    runner.delivery_router.adapters = runner.adapters
    +    runner.stop = AsyncMock()
    +
    +    await runner._handle_adapter_fatal_error(old_adapter)
    +
    +    new_adapter.disconnect.assert_not_awaited()
    +    assert runner.adapters[Platform.WHATSAPP] is new_adapter
    +    assert Platform.WHATSAPP not in runner._failed_platforms
    +    runner.stop.assert_not_awaited()

  • tests/gateway/test_telegram_network_reconnect.py

    @@ -117,6 +117,43 @@ async def test_reconnect_does_not_self_schedule_when_fatal_error_set():
         )
     
     
    [redacted:email]
    +async def test_reconnect_chained_retry_updates_polling_error_task():
    +    """
    +    When start_polling() fails and the handler self-schedules a retry, that
    +    retry task must become the new `_polling_error_task` — otherwise the
    +    reentrancy guard used by the heartbeat loop, the pending-updates probe,
    +    and the PTB error callback goes stale while a recovery is still in
    +    flight, letting a second concurrent recovery start for the same outage.
    +
    +    Regression test for the race behind the "half-destroyed adapter" bug
    +    (gateway reports connected but silently stops processing messages).
    +    """
    +    adapter = _make_adapter()
    +    adapter._polling_network_error_count = 1
    +
    +    mock_updater = MagicMock()
    +    mock_updater.running = True
    +    mock_updater.stop = AsyncMock()
    +    mock_updater.start_polling = AsyncMock(side_effect=Exception("Timed out"))
    +
    +    mock_app = MagicMock()
    +    mock_app.updater = mock_updater
    +    adapter._app = mock_app
    +
    +    with patch("asyncio.sleep", new_callable=AsyncMock):
    +        await adapter._handle_polling_network_error(Exception("Bad Gateway"))
    +
    +    assert adapter._polling_error_task is not None
    +    assert not adapter._polling_error_task.done()
    +
    +    adapter._polling_error_task.cancel()
    +    try:
    +        await adapter._polling_error_task
    +    except (asyncio.CancelledError, Exception):
    +        pass
    +
    +
     @pytest.mark.asyncio
     async def test_reconnect_success_resets_error_count():
         """

• #58342 fix(telegram): bound all updater.stop() calls to prevent CLOSE-WAIT reconnect hang (salvage #58272, closes #58270)

  merged to main

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

  Show the diff

  • 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.
    @@ -1990,7 +1998,25 @@ async def _handle_polling_network_error(self, error: Exception) -> None:
     
             try:
                 if app and app.updater and app.updater.running:
    -                await app.updater.stop()
    +                try:
    +                    # Guard stop() with a timeout: when the underlying TCP
    +                    # connection is in CLOSE-WAIT the PTB polling task is
    +                    # blocked on epoll on the dead socket and never wakes up,
    +                    # so an unguarded stop() hangs indefinitely.  The result
    +                    # is that _polling_error_task stays alive-but-blocked
    +                    # forever, every subsequent heartbeat probe sees it as
    +                    # "in-flight" and skips triggering a new reconnect, and
    +                    # 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=_UPDATER_STOP_TIMEOUT)
    +                except asyncio.TimeoutError:
    +                    logger.warning(
    +                        "[%s] updater.stop() timed out during network-error "
    +                        "reconnect (likely CLOSE-WAIT socket); forcing drain "
    +                        "and restart without clean stop",
    +                        self.name,
    +                    )
             except Exception:
                 pass
     
    @@ -2364,10 +2390,19 @@ async def _handle_polling_conflict(self, error: Exception) -> None:
                 )
                 # 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
     
    @@ -2435,16 +2470,33 @@ async def _handle_polling_conflict(self, error: Exception) -> None:
                 "[%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,
    @@ -3334,9 +3386,20 @@ async def disconnect(self) -> None:
     
             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()

  • tests/gateway/test_telegram_conflict.py

    @@ -220,6 +220,20 @@ async def failing_start_polling(**kwargs):
                 conflict("Conflict: terminated by other getUpdates request")
             )
     
    +    # Retries 1-4 each schedule a background recovery task via
    +    # loop.create_task(self._handle_polling_conflict(...)) that this test
    +    # never awaits.  Cancel the last one so a leaked task can't get a
    +    # scheduler turn under load and re-drive the counter into the fatal
    +    # branch a second time — which would fire _notify_fatal_error twice and
    +    # break assert_awaited_once() non-deterministically.
    +    leaked = adapter._polling_error_task
    +    if leaked is not None and not leaked.done():
    +        leaked.cancel()
    +        try:
    +            await leaked
    +        except (asyncio.CancelledError, Exception):
    +            pass
    +
         # After 5 failed retries (count 1-5 each enter the retry branch but
         # start_polling raises), the 6th conflict pushes count to 6 which
         # exceeds MAX_CONFLICT_RETRIES (5), entering the fatal branch.

  • tests/gateway/test_telegram_network_reconnect.py

    @@ -812,3 +812,59 @@ async def test_schedule_polling_recovery_tracks_background_task():
         await adapter._polling_error_task
         adapter._handle_polling_network_error.assert_awaited_once()
     
    +
    [redacted:email]
    +async def test_handle_polling_network_error_updater_stop_timeout():
    +    """updater.stop() hanging (CLOSE-WAIT) must not block the reconnect ladder.
    +
    +    When the underlying TCP connection is in CLOSE-WAIT, PTB's polling task is
    +    blocked on epoll on the dead socket.  updater.stop() awaits that task and
    +    therefore hangs indefinitely.  The fix wraps stop() in asyncio.wait_for()
    +    with a 15-second timeout so the reconnect always advances.
    +
    +    This test simulates the hang by making stop() sleep forever and verifies
    +    that _drain_polling_connections() and start_polling() are still called
    +    after the timeout fires.
    +    Refs: NousResearch/hermes-agent#58270
    +    """
    +    adapter = _make_adapter()
    +    adapter._polling_network_error_count = 0
    +
    +    # Build a fake app whose updater.stop() hangs forever.
    +    app = MagicMock()
    +    app.updater = MagicMock()
    +    app.updater.running = True
    +
    +    async def _hanging_stop():
    +        await asyncio.sleep(9999)  # simulate CLOSE-WAIT block
    +
    +    app.updater.stop = _hanging_stop
    +    app.updater.start_polling = AsyncMock()
    +    adapter._app = app
    +
    +    drain_called = []
    +
    +    async def _fake_drain():
    +        drain_called.append(True)
    +
    +    adapter._drain_polling_connections = _fake_drain
    +
    +    start_polling_called = []
    +
    +    async def _fake_start_polling(**kwargs):
    +        start_polling_called.append(True)
    +
    +    app.updater.start_polling = AsyncMock(side_effect=_fake_start_polling)
    +
    +    # 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
    +
    +    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().
    +    assert drain_called, "_drain_polling_connections was not called after stop() timeout"
    +    assert start_polling_called, "start_polling was not called after stop() timeout"
    +

• #67241 fix(telegram): cause-agnostic wedged-recovery watchdog + bounded drain so the reconnect ladder can't freeze silently (#66377, supersedes #66492)

  merged to main

  files: contributors/emails/[redacted:email], plugins/platforms/telegram/adapter.py, tests/gateway/test_telegram_network_reconnect.py

  Show the diff

  • contributors/emails/[redacted:email]

    @@ -0,0 +1,2 @@
    +MaheshBhushan
    +# PR #66492 salvage (#66377)

  • plugins/platforms/telegram/adapter.py

    @@ -17,6 +17,7 @@
     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
    @@ -543,6 +544,24 @@ def _rich_normalize_linebreaks(text: str) -> str:
     # 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
    +# shutdown()/initialize() on the getUpdates httpx request close and rebuild the
    +# connection pool. When a connection is wedged on a stale CLOSE-WAIT socket that
    +# close can block forever, hanging _drain_polling_connections() and freezing the
    +# whole reconnect ladder (the tracked _polling_error_task never completes, so
    +# every escalation path stays gated behind its in-flight guard). Bound the drain
    +# 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
    @@ -1987,20 +2006,22 @@ async def _drain_polling_connections(self) -> None:
             except Exception:
                 return
             try:
    -            await polling_req.shutdown()
    +            # Bounded: a wedged CLOSE-WAIT socket can make this close hang
    +            # forever and freeze the reconnect ladder (#66377).
    +            await asyncio.wait_for(polling_req.shutdown(), timeout=_DRAIN_TIMEOUT)
             except Exception:
                 logger.debug(
    -                "[%s] Polling request shutdown failed (non-fatal)",
    +                "[%s] Polling request shutdown failed/timed out (non-fatal)",
                     self.name, exc_info=True,
                 )
             try:
    -            await polling_req.initialize()
    +            await asyncio.wait_for(polling_req.initialize(), timeout=_DRAIN_TIMEOUT)
                 logger.debug(
                     "[%s] Polling request pool drained before reconnect", self.name
                 )
             except Exception:
                 logger.debug(
    -                "[%s] Polling request re-initialize failed (non-fatal)",
    +                "[%s] Polling request re-initialize failed/timed out (non-fatal)",
                     self.name, exc_info=True,
                 )
     
    @@ -2454,6 +2475,16 @@ async def _polling_heartbeat_loop(self) -> None:
             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)
    @@ -2461,6 +2492,42 @@ async def _polling_heartbeat_loop(self) -> None:
                         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

  • tests/gateway/test_telegram_network_reconnect.py

    @@ -325,6 +325,104 @@ async def test_initialize_still_runs_when_shutdown_fails():
         mock_app.updater.start_polling.assert_called_once()
     
     
    [redacted:email]
    +async def test_reconnect_continues_if_drain_hangs(monkeypatch):
    +    """If the polling request drain HANGS (wedged httpx pool close on a
    +    CLOSE-WAIT socket), the reconnect ladder must still advance rather than
    +    freezing the tracked _polling_error_task forever.
    +
    +    Regression test for #66377: an unbounded ``shutdown()`` /
    +    ``initialize()`` in ``_drain_polling_connections`` leaves the handler
    +    task pending, which gates every escalation path and silently kills the
    +    gateway. The drain awaits are bounded by ``_DRAIN_TIMEOUT``, so the
    +    handler must complete and reach ``start_polling`` within a hard bound.
    +    """
    +    adapter = _make_adapter()
    +    adapter._polling_network_error_count = 1
    +
    +    mock_app, mock_polling_req = _make_mock_app()
    +
    +    async def _hang(*args, **kwargs):
    +        await asyncio.Event().wait()  # never returns
    +
    +    # Both drain awaits wedge indefinitely.
    +    mock_polling_req.shutdown = AsyncMock(side_effect=_hang)
    +    mock_polling_req.initialize = AsyncMock(side_effect=_hang)
    +    adapter._app = mock_app
    +
    +    # Keep the drain timeout tiny so the test stays fast; the real default
    +    # is generous enough not to truncate healthy closes.
    +    monkeypatch.setattr(tg_adapter, "_DRAIN_TIMEOUT", 0.01, raising=False)
    +
    +    with patch("asyncio.sleep", new_callable=AsyncMock):
    +        # Hard outer bound: on unfixed code the drain hangs forever and this
    +        # trips; with the fix the inner wait_for releases well before it.
    +        await asyncio.wait_for(
    +            adapter._handle_polling_network_error(Exception("Timed out")),
    +            timeout=5,
    +        )
    +
    +    # Ladder advanced past the wedged drain despite it never returning.
    +    mock_app.updater.start_polling.assert_called_once()
    +    assert adapter._polling_network_error_count == 2
    +    # The tracked task must not be stuck pending — otherwise every
    +    # escalation path stays gated behind an in-flight guard.
    +    assert (
    +        adapter._polling_error_task is None
    +        or adapter._polling_error_task.done()
    +    )
    +
    +
    [redacted:email]
    +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."""