From 80bacecd2e8fd30dbce3cac23942d0f9e3931f10 Mon Sep 17 00:00:00 2001 From: joaomarcos Date: Wed, 1 Jul 2026 00:29:06 -0300 Subject: [PATCH 1/2] fix(telegram): close reconnect races that leave adapter half-destroyed _handle_polling_network_error's chained retry never updated self._polling_error_task, so the reentrancy guard shared with the heartbeat loop and the pending-updates probe went stale mid-recovery, letting more than one recovery attempt run concurrently against the same adapter. Combined with a TOCTOU window in _handle_adapter_fatal_error (the adapter was only removed from self.adapters in a finally block after awaiting disconnect()), two concurrent fatal notifications for the same adapter could both pass the "still installed" check and call disconnect() twice, which is where the reported "'NoneType' object has no attribute 'updater'" originates once self._app is cleared by the first call. - Reassign the chained retry task to self._polling_error_task so the guard reflects an in-flight recovery. - Capture self._app in a local variable across the stop/start_polling sequence instead of re-reading self._app between awaits. - Claim (pop) the adapter from self.adapters before awaiting disconnect() in _handle_adapter_fatal_error, not after, closing the TOCTOU window for a concurrent notification on the same adapter. --- gateway/run.py | 13 +++-- plugins/platforms/telegram/adapter.py | 20 ++++++- tests/gateway/test_runner_fatal_adapter.py | 55 +++++++++++++++++++ .../test_telegram_network_reconnect.py | 37 +++++++++++++ 4 files changed, 117 insertions(+), 8 deletions(-) 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 @@ async def _handle_adapter_fatal_error(self, adapter: BasePlatformAdapter) -> Non 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 @@ 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. 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(): """ From a30e4478904d4a529a64652c7d39672ec6cca5f6 Mon Sep 17 00:00:00 2001 From: joaomarcos Date: Wed, 1 Jul 2026 00:48:04 -0300 Subject: [PATCH 2/2] fix(gateway): ignore stale fatal-error notifications from superseded adapters A delayed fatal-error notification from an adapter instance that has already been replaced by a successful reconnect (a different adapter object now owns the platform slot) was still processed: it overwrote the platform's runtime status back to retrying/fatal and could re-queue an already-healthy platform for reconnection. Snapshot the current owner of the platform slot at the top of _handle_adapter_fatal_error and bail out before any side effect when it belongs to a different, already-installed adapter. --- gateway/run.py | 18 +++++++++- tests/gateway/test_runner_fatal_adapter.py | 39 ++++++++++++++++++++++ 2 files changed, 56 insertions(+), 1 deletion(-) diff --git a/gateway/run.py b/gateway/run.py index 049706fda9b5..4c4a4b107b2f 100644 --- a/gateway/run.py +++ b/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,7 +3694,6 @@ 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: # Claim this adapter for teardown before awaiting disconnect() — # a second fatal-error notification for the same adapter (e.g. diff --git a/tests/gateway/test_runner_fatal_adapter.py b/tests/gateway/test_runner_fatal_adapter.py index dc1462235732..7fce3841fde5 100644 --- a/tests/gateway/test_runner_fatal_adapter.py +++ b/tests/gateway/test_runner_fatal_adapter.py @@ -153,3 +153,42 @@ async def slow_disconnect(): ) assert disconnect_calls == 1 + + +@pytest.mark.asyncio +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()