diff --git a/gateway/run.py b/gateway/run.py index dbfc3326cd053..a22b7965bcd9b 100644 --- a/gateway/run.py +++ b/gateway/run.py @@ -3328,6 +3328,10 @@ class GatewayRunner(GatewayAuthorizationMixin, GatewayKanbanWatchersMixin, Gatew # Key: Platform enum, Value: {"config": platform_config, "attempts": int, "next_retry": float} self._failed_platforms: Dict[Platform, Dict[str, Any]] = {} + # Strong refs to detached fatal-error handler tasks (see + # _handle_adapter_fatal_error) so the event loop can't GC them mid-run. + self._fatal_handler_tasks: set = set() + # Track pending /update prompt responses per session. # Key: session_key, Value: True when a prompt is waiting for user input. self._update_prompt_pending: Dict[str, bool] = {} @@ -4378,7 +4382,65 @@ class GatewayRunner(GatewayAuthorizationMixin, GatewayKanbanWatchersMixin, Gatew If the error is retryable (e.g. network blip, DNS failure), queue the platform for background reconnection instead of giving up permanently. + + The notification arrives on the failing adapter's own polling task, + and the disconnect inside the handler can cancel that task mid-flight: + disconnect()'s current-task guard misses it because + _safe_adapter_disconnect runs the close in a wrapper task. A cancelled + handler dies between the fatal log and the reconnect queue, silently + stranding the platform (observed 2026-07-21: telegram popped from + adapters but never queued after a travel network outage). Run the real + work in a detached task that adapter teardown cannot cancel. """ + tasks = getattr(self, "_fatal_handler_tasks", None) + if tasks is None: + tasks = self._fatal_handler_tasks = set() + task = asyncio.create_task(self._handle_adapter_fatal_error_detached(adapter)) + tasks.add(task) + task.add_done_callback(tasks.discard) + # Await so callers that expect completion still get it — but through + # shield(): Task.cancel() on the caller also cancels the future it is + # awaiting (_fut_waiter), so a plain `await task` would tunnel the + # cancellation straight into the "detached" task. shield() absorbs + # it: the caller sees CancelledError, the handler runs to completion. + await asyncio.shield(task) + + async def _handle_adapter_fatal_error_detached( + self, adapter: BasePlatformAdapter + ) -> None: + """Run the fatal handler; if the platform still ends up stranded + (not reconnected, not queued, not intentionally disabled), exit the + gateway with failure so the service manager restarts it instead of + leaving a silent partial outage.""" + try: + await self._handle_adapter_fatal_error_impl(adapter) + except Exception: + logger.exception( + "Fatal-error handling for %s raised unexpectedly", + adapter.platform.value, + ) + finally: + platform = adapter.platform + shutdown_event = getattr(self, "_shutdown_event", None) + stranded = ( + adapter.fatal_error_retryable + and platform not in self.adapters + and platform not in getattr(self, "_failed_platforms", {}) + and not (shutdown_event is not None and shutdown_event.is_set()) + ) + if stranded: + logger.error( + "%s adapter was lost without entering the reconnection " + "queue; exiting gateway so the service manager restarts it.", + platform.value, + ) + self._exit_reason = ( + f"{platform.value} adapter lost without reconnection queue" + ) + self._exit_with_failure = True + await self.stop() + + async def _handle_adapter_fatal_error_impl(self, adapter: BasePlatformAdapter) -> None: # 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 diff --git a/tests/gateway/test_platform_reconnect.py b/tests/gateway/test_platform_reconnect.py index 4c5cbc93ea82a..fe3526a9a1a93 100644 --- a/tests/gateway/test_platform_reconnect.py +++ b/tests/gateway/test_platform_reconnect.py @@ -949,3 +949,71 @@ class TestSpawnSupervised: # _MAX_SUPERVISED_RESTARTS + 1; the reset lets it run to completion. assert calls["n"] == target assert calls["n"] > runner._MAX_SUPERVISED_RESTARTS + 1 + + +class TestFatalHandoffCancellationProof: + """The fatal-error handoff must survive cancellation of the notifying + task, and a retryable platform must never be silently stranded.""" + + @pytest.mark.asyncio + async def test_caller_cancellation_does_not_strand_platform(self): + """The fatal notification arrives on the failing adapter's own + polling task, and adapter.disconnect() inside the handler can cancel + that task mid-teardown. The platform must still reach the reconnect + queue (previously the CancelledError killed the handler between the + fatal log and the queue, stranding the platform until a manual + restart).""" + runner = _make_runner() + runner.stop = AsyncMock() + + adapter = StubAdapter(succeed=True) + adapter._set_fatal_error("network_error", "DNS failure", retryable=True) + runner.adapters[Platform.TELEGRAM] = adapter + + release = asyncio.Event() + + async def slow_disconnect(): + await release.wait() + + adapter.disconnect = slow_disconnect # hold the handler mid-teardown + + caller = asyncio.create_task(runner._handle_adapter_fatal_error(adapter)) + for _ in range(5): + await asyncio.sleep(0) # let the handler reach the disconnect await + caller.cancel() # what disconnect() does to the notifying task + with pytest.raises(asyncio.CancelledError): + await caller + release.set() # teardown completes after the caller has died + + for _ in range(200): + if Platform.TELEGRAM in runner._failed_platforms: + break + await asyncio.sleep(0.01) + assert Platform.TELEGRAM in runner._failed_platforms + + @pytest.mark.asyncio + async def test_stranded_retryable_platform_exits_for_supervisor_restart(self): + """If a retryable platform ends up neither reconnected nor queued + (e.g. its config entry is gone so queueing is skipped), the gateway + must exit with failure so launchd/systemd KeepAlive restarts it, + instead of running indefinitely with a dead platform while healthy + peers mask the loss (#68693).""" + runner = _make_runner() + + async def _stop(): + runner._shutdown_event.set() + + runner.stop = AsyncMock(side_effect=_stop) + runner.config = GatewayConfig(platforms={}) # queueing impossible + + adapter = StubAdapter(succeed=True) + adapter._set_fatal_error("network_error", "DNS failure", retryable=True) + runner.adapters[Platform.TELEGRAM] = adapter + # A healthy peer keeps self.adapters non-empty, so the existing + # "no platforms remain" shutdown branches do not fire. + runner.adapters[Platform.FEISHU] = StubAdapter(platform=Platform.FEISHU) + + await runner._handle_adapter_fatal_error(adapter) + + assert runner._exit_with_failure is True + assert runner.stop.await_count == 1