From 3c7cff843d45c02666dfeec60439210b4157b47f Mon Sep 17 00:00:00 2001 From: "Spark (DGX)" Date: Mon, 27 Jul 2026 00:37:12 -0400 Subject: [PATCH] fix(photon): dispatch fatal-error notification from a detached task Follow-up to #69112. That PR hardened the shared gateway dispatch path in gateway/run.py against the *caller* being cancelled. A second, self-referential cancellation specific to PhotonAdapter sits one layer underneath it and survived that fix. PhotonAdapter is the only platform adapter that awaits _notify_fatal_error() inline, on the same task that detected the fault. Both _monitor_sidecar_health and _supervise_sidecar run as self._sidecar_health_task / self._sidecar_supervisor_task, and the notification routes into GatewayRunner._handle_adapter_fatal_error_impl, which tears the adapter down via _safe_adapter_disconnect -> disconnect(). disconnect() then cancels self._sidecar_health_task and awaits it -- which, when the health task is what raised the notification, means disconnect() cancels its own caller several plain-await frames up. disconnect()'s `task is not asyncio.current_task()` guard does not catch this. The current task where that guard evaluates is the wrapper _await_adapter_cleanup_with_timeout creates around disconnect() via asyncio.ensure_future, not the health task further up the chain, so the guard passes and the cancel lands. CancelledError stopped subclassing Exception in Python 3.8, so the `except Exception` that wrapped the inline notify call never saw it. The health task died silently mid-handoff: no log line, no "exception never retrieved" warning (cancellation is normal asyncio), and no retry. The platform stayed stranded until the gateway was restarted by hand. Fix: dispatch the notification onto a new task, the same pattern DiscordAdapter._handle_bot_task_done already uses for this reason. disconnect() can then cancel the health/supervisor task freely without that cancellation reaching the code still running the handoff, so the handoff always reaches the reconnect queue. Both Photon fatal call sites are converted: the health-poll path (observed wedging) and the sidecar-crash path (same shape, not yet observed). gateway/run.py is untouched. Observed twice on a self-hosted gateway, ~4h38m and ~52min of silent inbound outage, both cleared only by a manual restart, both post-dating #69112's merge. In each case the fatal log line appears and `queued for background reconnection` never does. Tests: new tests/plugins/platforms/photon/test_fatal_notify_self_cancel.py covers the self-cancellation (fails with CancelledError without this change), that the dispatch does not block its caller, that a failing notification warns rather than raising, and a source guard against reintroducing either inline await. Two assertions in test_overflow_recovery.py that drove these coroutines directly now drain pending tasks before asserting delivery, since the notification is deliberately no longer awaited inline. Prepared with agent assistance and reviewed before submission. Co-Authored-By: Claude Opus 5 (1M context) --- plugins/platforms/photon/adapter.py | 51 +++++- .../photon/test_fatal_notify_self_cancel.py | 150 ++++++++++++++++++ .../photon/test_overflow_recovery.py | 26 +++ 3 files changed, 219 insertions(+), 8 deletions(-) create mode 100644 tests/plugins/platforms/photon/test_fatal_notify_self_cancel.py diff --git a/plugins/platforms/photon/adapter.py b/plugins/platforms/photon/adapter.py index b73bff8d5fc7..2cbe5c3a4b1b 100644 --- a/plugins/platforms/photon/adapter.py +++ b/plugins/platforms/photon/adapter.py @@ -659,6 +659,47 @@ class PhotonAdapter(BasePlatformAdapter): self._http_client = None self._mark_disconnected() + def _dispatch_fatal_notification(self) -> None: + """Notify the gateway of a fatal error from a detached task. + + ``_monitor_sidecar_health`` and ``_supervise_sidecar`` run as + ``self._sidecar_health_task`` / ``self._sidecar_supervisor_task`` and + used to call ``await self._notify_fatal_error()`` directly, on their + own call stack. That routes straight into + ``GatewayRunner._handle_adapter_fatal_error_impl``, which tears the + adapter down via ``_safe_adapter_disconnect`` -> ``disconnect()``. + ``disconnect()`` cancels ``self._sidecar_health_task`` and awaits it + to make sure it's really gone -- but when the *health task itself* is + what's driving this whole notification (the common case: the health + poll is what detected the fatal condition), that task IS the one + awaiting several frames up the stack, so ``disconnect()`` cancels and + awaits its own caller. ``disconnect()``'s + ``task is not asyncio.current_task()`` guard doesn't catch this, + because the wrapper task ``_await_adapter_cleanup_with_timeout`` + creates around ``disconnect()`` (via ``asyncio.ensure_future``) is + the current task there, not the health/supervisor task further up + the plain-await chain. The self-cancellation throws an unhandled + ``CancelledError`` through the health/supervisor task with nothing + to catch it -- ``CancelledError`` stopped subclassing ``Exception`` + in Python 3.8, so the ``except Exception`` around the notify call + never saw it -- and the task dies silently mid-handoff: no log line, + no retry, the platform stranded with no automated way back. + + Dispatching the notification onto a brand-new task instead -- the + same pattern ``DiscordAdapter._handle_bot_task_done`` already uses -- + means ``disconnect()`` can cancel the health/supervisor task freely + without that cancellation ever reaching back into the code that's + still running the handoff, so the handoff always reaches the + reconnect queue. + """ + asyncio.create_task(self._notify_fatal_error_logged()) + + async def _notify_fatal_error_logged(self) -> None: + try: + await self._notify_fatal_error() + except Exception as exc: # pragma: no cover - defensive + logger.warning("[photon] fatal-error notification failed: %s", exc) + # -- Inbound stream consumer ------------------------------------------ async def _inbound_loop(self) -> None: @@ -738,10 +779,7 @@ class PhotonAdapter(BasePlatformAdapter): message, retryable=True, ) - try: - await self._notify_fatal_error() - except Exception as exc: # pragma: no cover - defensive - logger.warning("[photon] fatal-error notification failed: %s", exc) + self._dispatch_fatal_notification() break async def _on_inbound_line(self, line: str) -> None: @@ -1250,10 +1288,7 @@ class PhotonAdapter(BasePlatformAdapter): f"Photon sidecar exited unexpectedly (code {exit_code})", retryable=True, ) - try: - await self._notify_fatal_error() - except Exception as exc: # pragma: no cover - defensive - logger.warning("[photon] fatal-error notification failed: %s", exc) + self._dispatch_fatal_notification() async def _stop_sidecar(self) -> None: proc = self._sidecar_proc diff --git a/tests/plugins/platforms/photon/test_fatal_notify_self_cancel.py b/tests/plugins/platforms/photon/test_fatal_notify_self_cancel.py new file mode 100644 index 000000000000..68c034a99d69 --- /dev/null +++ b/tests/plugins/platforms/photon/test_fatal_notify_self_cancel.py @@ -0,0 +1,150 @@ +"""Photon's fatal-error notification must not be cancelled by its own teardown. + +`_monitor_sidecar_health` and `_supervise_sidecar` run as long-lived tasks on +the adapter. When one of them detects a fatal condition, the gateway's handler +tears the adapter down, and `disconnect()` cancels `_sidecar_health_task` and +awaits it. If the notification is awaited inline on the health task's own call +stack, `disconnect()` ends up cancelling its own ultimate caller: the +`task is not asyncio.current_task()` guard in `disconnect()` compares against +the wrapper task the gateway created around `disconnect()`, not the health task +several plain-await frames further up, so the guard passes and the cancel lands. + +`CancelledError` stopped subclassing `Exception` in Python 3.8, so the +`except Exception` that used to wrap the inline notify call never saw it. The +health task died silently mid-handoff -- no log line, no retry -- and the +platform stayed stranded until someone restarted the gateway by hand. + +PR #69112 hardened the shared dispatch path in `gateway/run.py` against a +cancelled *caller*. These tests cover the Photon-specific self-cancellation one +layer down, which that PR's scope could not reach. +""" +from __future__ import annotations + +import asyncio +from typing import Any, Dict, List + +import pytest + +from gateway.config import PlatformConfig +from plugins.platforms.photon.adapter import PhotonAdapter + + +def _make_adapter(monkeypatch: pytest.MonkeyPatch) -> PhotonAdapter: + monkeypatch.setenv("PHOTON_PROJECT_ID", "test-project-id") + monkeypatch.setenv("PHOTON_PROJECT_SECRET", "test-project-secret") + return PhotonAdapter(PlatformConfig(enabled=True, token="", extra={})) + + +class TestFatalNotifyIsDetached: + """The notification must outlive cancellation of the task that raised it.""" + + @pytest.mark.asyncio + async def test_health_task_cancellation_does_not_kill_notification( + self, monkeypatch: pytest.MonkeyPatch + ) -> None: + """A fatal handler that cancels the health task (exactly what + ``disconnect()`` does) must still see the notification delivered.""" + adapter = _make_adapter(monkeypatch) + adapter._inbound_running = True + adapter._sidecar_health_interval = 0 + delivered = asyncio.Event() + + async def fake_notify() -> None: + # Mirror the real handler: tear the adapter down, cancelling the + # health task, then finish the handoff. + await adapter.disconnect() + delivered.set() + + monkeypatch.setattr(adapter, "_notify_fatal_error", fake_notify) + monkeypatch.setattr(adapter, "_stop_sidecar", lambda: _noop()) + + async def degraded(_path: str, _payload: Dict[str, Any]) -> Dict[str, Any]: + return {"stream": {"ok": False, "state": "degraded", "degradedForMs": 4000, + "lastIssue": "stream persistently failing"}} + + monkeypatch.setattr(adapter, "_sidecar_call", degraded) + + health = asyncio.create_task(adapter._monitor_sidecar_health()) + adapter._sidecar_health_task = health + + await asyncio.wait_for(delivered.wait(), timeout=5.0) + assert adapter.has_fatal_error + assert adapter.fatal_error_code == "UPSTREAM_STREAM_DEGRADED" + assert adapter.fatal_error_retryable + + # With the dispatch detached, the health task reaches its own `break` + # and returns cleanly instead of being cancelled out from under the + # handoff. Either way it must not die with an unhandled exception -- + # that was the silent failure that stranded the platform. + await asyncio.wait_for(asyncio.shield(health), timeout=2.0) + assert health.done() + if not health.cancelled(): + assert health.exception() is None + + @pytest.mark.asyncio + async def test_dispatch_does_not_await_on_caller_stack( + self, monkeypatch: pytest.MonkeyPatch + ) -> None: + """``_dispatch_fatal_notification`` must return without awaiting, so a + cancel aimed at the calling task cannot reach the handoff.""" + adapter = _make_adapter(monkeypatch) + started = asyncio.Event() + finished = asyncio.Event() + + async def slow_notify() -> None: + started.set() + await asyncio.sleep(0.05) + finished.set() + + monkeypatch.setattr(adapter, "_notify_fatal_error", slow_notify) + + async def caller() -> None: + adapter._dispatch_fatal_notification() # must not block + + task = asyncio.create_task(caller()) + await task # returns immediately even though notify sleeps + await asyncio.wait_for(started.wait(), timeout=2.0) + + task.cancel() # cancelling the caller must not touch the notification + await asyncio.wait_for(finished.wait(), timeout=2.0) + + @pytest.mark.asyncio + async def test_notification_failure_is_logged_not_raised( + self, monkeypatch: pytest.MonkeyPatch, caplog: pytest.LogCaptureFixture + ) -> None: + """A failing notification must warn rather than surface as an + unretrieved task exception.""" + adapter = _make_adapter(monkeypatch) + + async def boom() -> None: + raise RuntimeError("gateway unreachable") + + monkeypatch.setattr(adapter, "_notify_fatal_error", boom) + + with caplog.at_level("WARNING"): + await adapter._notify_fatal_error_logged() + + assert "fatal-error notification failed" in caplog.text + + +class TestBothCallSitesDetached: + """Neither fatal path may await the notification inline.""" + + def test_no_inline_notify_awaits_remain(self) -> None: + """Guard against a future edit reintroducing the inline await.""" + import inspect + + from plugins.platforms.photon import adapter as photon_adapter + + for name in ("_monitor_sidecar_health", "_supervise_sidecar"): + src = inspect.getsource(getattr(photon_adapter.PhotonAdapter, name)) + assert "await self._notify_fatal_error()" not in src, ( + f"{name} awaits _notify_fatal_error inline; use " + f"_dispatch_fatal_notification() so disconnect() cannot cancel " + f"its own caller" + ) + assert "_dispatch_fatal_notification()" in src + + +async def _noop() -> None: + return None diff --git a/tests/plugins/platforms/photon/test_overflow_recovery.py b/tests/plugins/platforms/photon/test_overflow_recovery.py index 75d9af59d83d..25216f8f7f86 100644 --- a/tests/plugins/platforms/photon/test_overflow_recovery.py +++ b/tests/plugins/platforms/photon/test_overflow_recovery.py @@ -19,6 +19,7 @@ No Node sidecar is spawned and no ports are bound. """ from __future__ import annotations +import asyncio from typing import Any, Dict import pytest @@ -174,6 +175,10 @@ async def test_unexpected_sidecar_exit_raises_retryable_fatal( # than crashing the whole gateway. assert adapter.fatal_error_retryable is True assert adapter._running is False + # The notification is dispatched onto its own task rather than awaited on + # the supervisor's stack, so that disconnect() cancelling the supervisor + # cannot kill the handoff. Let that task run before asserting delivery. + await _drain_pending_tasks() assert notified == [True] @@ -233,4 +238,25 @@ async def test_degraded_stream_health_raises_retryable_fatal( assert adapter.has_fatal_error is True assert adapter.fatal_error_code == "UPSTREAM_STREAM_DEGRADED" assert adapter.fatal_error_retryable is True + # Dispatched detached (see _dispatch_fatal_notification) so the health + # task's own teardown cannot cancel the handoff; drain before asserting. + await _drain_pending_tasks() assert notified == [True] + + +async def _drain_pending_tasks(limit: int = 50) -> None: + """Let detached fatal-notification tasks finish before asserting on them. + + ``_dispatch_fatal_notification`` deliberately does not await the + notification (that is what kept ``disconnect()`` from cancelling its own + caller), so a test that drives ``_monitor_sidecar_health`` / + ``_supervise_sidecar`` directly returns before the notification has run. + """ + for _ in range(limit): + pending = [ + t for t in asyncio.all_tasks() + if t is not asyncio.current_task() and not t.done() + ] + if not pending: + return + await asyncio.wait(pending, timeout=1.0)