Skip to content
Closed
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
62 changes: 62 additions & 0 deletions gateway/run.py
Original file line number Diff line number Diff line change
Expand Up @@ -3328,6 +3328,10 @@ def __init__(self, config: Optional[GatewayConfig] = None):
# 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] = {}
Expand Down Expand Up @@ -4378,7 +4382,65 @@ 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.

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
Expand Down
68 changes: 68 additions & 0 deletions tests/gateway/test_platform_reconnect.py
Original file line number Diff line number Diff line change
Expand Up @@ -949,3 +949,71 @@ async def _coro():
# _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