Telegram adapter reconnect problems after network errors

Fixes on record

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

• #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():
         """