diff --git a/gateway/run.py b/gateway/run.py index 053f95f8793d..049706fda9b5 100644 --- a/gateway/run.py +++ b/gateway/run.py @@ -3679,11 +3679,14 @@ class GatewayRunner(GatewayAuthorizationMixin, GatewayKanbanWatchersMixin, Gatew 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: diff --git a/plugins/platforms/telegram/adapter.py b/plugins/platforms/telegram/adapter.py index 3b8302723b2c..4a8f1c679caa 100644 --- a/plugins/platforms/telegram/adapter.py +++ b/plugins/platforms/telegram/adapter.py @@ -1773,16 +1773,24 @@ class TelegramAdapter(BasePlatformAdapter): ) 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 @@ class TelegramAdapter(BasePlatformAdapter): ) 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. diff --git a/tests/gateway/test_runner_fatal_adapter.py b/tests/gateway/test_runner_fatal_adapter.py index 7e7739582d16..dc1462235732 100644 --- a/tests/gateway/test_runner_fatal_adapter.py +++ b/tests/gateway/test_runner_fatal_adapter.py @@ -1,3 +1,4 @@ +import asyncio from unittest.mock import AsyncMock import pytest @@ -98,3 +99,57 @@ 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 + + +@pytest.mark.asyncio +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 diff --git a/tests/gateway/test_telegram_network_reconnect.py b/tests/gateway/test_telegram_network_reconnect.py index 8c0dc6a563f2..c970adc8ff24 100644 --- a/tests/gateway/test_telegram_network_reconnect.py +++ b/tests/gateway/test_telegram_network_reconnect.py @@ -117,6 +117,43 @@ async def test_reconnect_does_not_self_schedule_when_fatal_error_set(): ) +@pytest.mark.asyncio +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(): """