diff --git a/gateway/run.py b/gateway/run.py index d06512c043a0..836ed14f4d78 100644 --- a/gateway/run.py +++ b/gateway/run.py @@ -8710,11 +8710,20 @@ class GatewayRunner(GatewayAuthorizationMixin, GatewayKanbanWatchersMixin, Gatew ", ".join(p.value for p in self._failed_platforms), ) # Track the reconnect watcher task so _ensure_reconnect_watcher_running - # can detect if it dies and respawn it (#70344). - self._reconnect_watcher_task = asyncio.create_task( - self._platform_reconnect_watcher() + # can detect if it dies and respawn it (#70344). Spawned via + # _spawn_supervised (not a bare asyncio.create_task) so an exception + # escaping the watcher's OUTER while-loop -- not just the per-platform + # inner try/except -- is caught, logged, and auto-restarted with + # backoff instead of silently killing the watcher forever. Without + # this, a platform already queued in _failed_platforms when the + # watcher dies stays stranded indefinitely: _ensure_reconnect_watcher_running() + # only gets called from a NEW fatal-error arrival, so if no other + # platform ever fails afterward, nothing ever notices the watcher is + # dead (#71758 -- reported as 17.5h of silent downtime for a platform + # whose transient upstream outage had long since recovered). + self._reconnect_watcher_task = self._spawn_supervised( + self._platform_reconnect_watcher, "platform_reconnect_watcher" ) - self._background_tasks.add(self._reconnect_watcher_task) # Start background handoff watcher — picks up CLI sessions marked # handoff_state='pending' in state.db and re-binds them to the @@ -9297,11 +9306,9 @@ class GatewayRunner(GatewayAuthorizationMixin, GatewayKanbanWatchersMixin, Gatew "Reconnect watcher task is dead (done=%s) — respawning", task.done() if task is not None else "N/A", ) - self._reconnect_watcher_task = asyncio.create_task( - self._platform_reconnect_watcher() + self._reconnect_watcher_task = self._spawn_supervised( + self._platform_reconnect_watcher, "platform_reconnect_watcher" ) - if getattr(self, "_background_tasks", None) is not None: - self._background_tasks.add(self._reconnect_watcher_task) async def _platform_reconnect_watcher(self) -> None: """Background task that periodically retries connecting failed platforms. @@ -9333,7 +9340,13 @@ class GatewayRunner(GatewayAuthorizationMixin, GatewayKanbanWatchersMixin, Gatew for platform in list(self._failed_platforms.keys()): if not self._running: return - info = self._failed_platforms[platform] + info = self._failed_platforms.get(platform) + if info is None: + # Removed concurrently (e.g. a manual /platform resume, + # or a reconnect that succeeded via a different path) + # between the snapshot above and this lookup. Not an + # error -- just nothing to do for it this pass. + continue # Skip paused platforms entirely — they need explicit # /platform resume to come back. if info.get("paused"): diff --git a/tests/gateway/test_platform_reconnect.py b/tests/gateway/test_platform_reconnect.py index 7f526e70bfe5..3980160f21e9 100644 --- a/tests/gateway/test_platform_reconnect.py +++ b/tests/gateway/test_platform_reconnect.py @@ -1126,7 +1126,224 @@ class TestEnsureReconnectWatcherRunning: # ── _handle_adapter_fatal_error calls _ensure_reconnect_watcher ──────── -class TestFatalErrorCallsEnsureWatcher: +class TestReconnectWatcherSelfHeals: + """Regression tests for issue #71758: a platform already queued in + _failed_platforms when the reconnect watcher task dies from an + uncaught exception stayed stranded forever, because + _ensure_reconnect_watcher_running() is only called from a NEW + fatal-error arrival -- if no other platform ever fails afterward, + nothing notices the watcher is dead. The watcher must now be spawned + via _spawn_supervised (like other long-lived background tasks), so an + exception escaping its OUTER while-loop is caught, logged, and + auto-restarted with backoff -- independent of any new fatal-error + event. + """ + + @pytest.mark.asyncio + async def test_startup_spawns_watcher_via_spawn_supervised(self, monkeypatch): + """The initial watcher spawn at gateway startup must go through + _spawn_supervised, not a bare asyncio.create_task.""" + runner = _make_runner() + runner._background_tasks = set() + calls = [] + + def fake_spawn_supervised(coro_factory, name, **kw): + calls.append(name) + task = asyncio.create_task(coro_factory()) + runner._background_tasks.add(task) + return task + + async def _noop_watcher(): + await asyncio.sleep(3600) + + monkeypatch.setattr(runner, "_spawn_supervised", fake_spawn_supervised) + monkeypatch.setattr(runner, "_platform_reconnect_watcher", _noop_watcher) + + # Mirror the startup snippet: spawn via _spawn_supervised. + runner._reconnect_watcher_task = runner._spawn_supervised( + runner._platform_reconnect_watcher, "platform_reconnect_watcher" + ) + + assert "platform_reconnect_watcher" in calls + runner._reconnect_watcher_task.cancel() + try: + await runner._reconnect_watcher_task + except asyncio.CancelledError: + pass + + @pytest.mark.asyncio + async def test_ensure_reconnect_watcher_running_uses_spawn_supervised(self): + """The manual-respawn path in _ensure_reconnect_watcher_running must + also use _spawn_supervised, so a respawned watcher gets the same + auto-restart protection going forward.""" + runner = _make_runner() + runner._running = True + runner._background_tasks = set() + runner._reconnect_watcher_task = asyncio.create_task(asyncio.sleep(0)) + await runner._reconnect_watcher_task # let it finish (dead) + + spawn_calls = [] + original_spawn = runner._spawn_supervised + + def spy_spawn_supervised(coro_factory, name, **kw): + spawn_calls.append(name) + return original_spawn(coro_factory, name, **kw) + + runner._spawn_supervised = spy_spawn_supervised + runner._ensure_reconnect_watcher_running() + + assert spawn_calls == ["platform_reconnect_watcher"] + runner._reconnect_watcher_task.cancel() + try: + await runner._reconnect_watcher_task + except asyncio.CancelledError: + pass + + @pytest.mark.asyncio + async def test_watcher_self_heals_after_uncaught_exception_with_no_new_fatal_error(self): + """The core #71758 regression: a platform sits queued in + _failed_platforms. The watcher task dies from an uncaught + exception (simulating the KeyError race / any other bug in the + outer loop). WITHOUT any new fatal-error event for a different + platform, the watcher must still come back on its own via + _spawn_supervised's crash-detection callback -- the exact gap + that stranded the platform for 17.5h in the reported bug. + """ + runner = _make_runner() + runner._running = True + runner._background_tasks = set() + runner._SUPERVISED_HEALTHY_SECS = GatewayRunner._SUPERVISED_HEALTHY_SECS + runner._MAX_SUPERVISED_RESTARTS = GatewayRunner._MAX_SUPERVISED_RESTARTS + runner._spawn_supervised = GatewayRunner._spawn_supervised.__get__(runner) + + attempt_count = {"n": 0} + + async def _flaky_watcher(): + attempt_count["n"] += 1 + if attempt_count["n"] == 1: + # Simulate the watcher's outer loop raising -- e.g. the + # KeyError race this same fix also hardens against, or any + # other bug in code outside the per-platform try/except. + raise RuntimeError("simulated watcher crash") + await asyncio.sleep(3600) # second run: stays "alive" + + runner._reconnect_watcher_task = runner._spawn_supervised( + _flaky_watcher, "platform_reconnect_watcher" + ) + + # Let the first (crashing) attempt run and die. + for _ in range(50): + await asyncio.sleep(0) + if attempt_count["n"] >= 1 and runner._reconnect_watcher_task.done(): + break + + assert attempt_count["n"] == 1 + assert runner._reconnect_watcher_task.done() + + # The supervised _done callback schedules a respawn after a short + # backoff (2**0 = 1s at attempt 0) -- wait for it without a new + # fatal-error event ever firing. + for _ in range(30): + await asyncio.sleep(0.1) + if attempt_count["n"] >= 2: + break + + assert attempt_count["n"] >= 2, ( + "Watcher must self-heal via _spawn_supervised without any new " + "fatal-error event -- this is the exact gap that stranded a " + "platform in the reported bug" + ) + + # Cleanup: cancel whatever task is currently tracked. + for task in list(runner._background_tasks): + task.cancel() + await asyncio.sleep(0) + + +class TestReconnectWatcherRaceGuard: + """Regression: a platform removed from _failed_platforms concurrently + (e.g. a manual /platform resume racing with the watcher's own + snapshot-then-lookup) must not raise KeyError and kill the loop + iteration -- it should just be skipped for that pass.""" + + @pytest.mark.asyncio + async def test_watcher_survives_platform_removed_mid_pass(self, monkeypatch): + """Two platforms are due for retry. Reconnecting the first one, as + a side effect, removes the second from _failed_platforms (stand-in + for any concurrent path -- a manual /platform resume, a reconnect + that succeeded elsewhere, etc.). The watcher must finish its pass + without raising, and must still be alive afterward.""" + runner = _make_runner() + runner._running = True + runner._background_tasks = set() + runner.session_store = MagicMock() + runner._busy_text_mode = "interrupt" + + cfg = PlatformConfig(enabled=True, token="test") + runner._failed_platforms = { + Platform.TELEGRAM: {"config": cfg, "attempts": 0, "next_retry": 0.0}, + Platform.DISCORD: {"config": cfg, "attempts": 0, "next_retry": 0.0}, + } + runner._platform_connect_timeout_secs = MagicMock(return_value=5) + runner._sync_voice_mode_state_to_adapter = MagicMock() + runner._schedule_resume_pending_sessions = MagicMock() + runner._make_adapter_auth_check = MagicMock(return_value=lambda *a, **kw: True) + runner._recover_telegram_topic_thread_id = MagicMock() + runner._handle_message = MagicMock() + runner._handle_active_session_busy_message = MagicMock() + runner._handle_reaction_event = MagicMock() + runner._update_platform_runtime_status = MagicMock() + + def fake_create_adapter(platform, platform_config): + adapter = MagicMock() + adapter.platform = platform + adapter.has_fatal_error = False + adapter.set_message_handler = MagicMock() + adapter.set_fatal_error_handler = MagicMock() + adapter.set_session_store = MagicMock() + adapter.set_busy_session_handler = MagicMock() + adapter.set_reaction_handler = MagicMock() + adapter.set_topic_recovery_fn = MagicMock() + adapter.set_authorization_check = MagicMock() + if platform == Platform.TELEGRAM: + # Side effect: concurrently "resolves" Discord's entry via + # some other path (e.g. a manual /platform resume), racing + # with the watcher's own snapshot-then-lookup for it. + runner._failed_platforms.pop(Platform.DISCORD, None) + return adapter + + runner._create_adapter = MagicMock(side_effect=fake_create_adapter) + + async def fake_connect(adapter, platform, is_reconnect=False): + return True + + runner._connect_adapter_with_timeout = fake_connect + + async def _one_pass(): + now = time.monotonic() + for platform in list(runner._failed_platforms.keys()): + info = runner._failed_platforms.get(platform) + if info is None: + continue + if now < info["next_retry"]: + continue + adapter = runner._create_adapter(platform, info["config"]) + success = await runner._connect_adapter_with_timeout( + adapter, platform, is_reconnect=True + ) + if success: + runner.adapters[platform] = adapter + runner._failed_platforms.pop(platform, None) + + # Must not raise, even though Discord vanishes from the dict as a + # side effect of processing Telegram. + await _one_pass() + + assert Platform.TELEGRAM in runner.adapters + assert Platform.DISCORD not in runner._failed_platforms + + + """Verify _handle_adapter_fatal_error calls _ensure_reconnect_watcher_running.""" @pytest.mark.asyncio