diff --git a/gateway/platforms/base.py b/gateway/platforms/base.py index f3a022c81b21d..6edd1688a99fb 100644 --- a/gateway/platforms/base.py +++ b/gateway/platforms/base.py @@ -27,6 +27,23 @@ from utils import normalize_proxy_url logger = logging.getLogger(__name__) + +def _consume_detached_handler_exception(task: "asyncio.Task") -> None: + """Done-callback retrieving a detached fatal-error handler's exception. + + Prevents "Task exception was never retrieved" warnings for handler tasks + we deliberately let finish in the background after their awaiting + (carrier) task was cancelled — see ``_notify_fatal_error``. + """ + if task.cancelled(): + return + exc = task.exception() + if exc is not None: + logger.error( + "Detached fatal-error handler task failed: %s", exc, exc_info=exc + ) + + # Audio file extensions Hermes recognizes for native audio delivery. # Keep Telegram's narrower attachment/voice sets below separate: formats such # as MPEG-2 Layer II are audio to Hermes but unsupported by sendAudio/sendVoice. @@ -3444,7 +3461,27 @@ class BasePlatformAdapter(ABC): return result = handler(self) if asyncio.iscoroutine(result): - await result + # Run the handler as a detached, shielded task. The notification + # is frequently awaited from inside an adapter-owned task (e.g. + # the Telegram ``_polling_error_task``), and the gateway's fatal + # handler tears the adapter down via ``disconnect()`` — which + # cancels that very task. Without the shield the cancellation + # killed the handler mid-flight: the adapter was already popped + # from the gateway's adapter map but never queued for background + # reconnection, leaving a zombie gateway with no platforms and no + # pending retries (#81335). + task = asyncio.ensure_future(result) + try: + await asyncio.shield(task) + except asyncio.CancelledError: + # The carrier task was cancelled (typically by our own + # teardown running inside the handler). Let the handler + # finish detached so reconnect queueing / the shutdown + # decision completes, and consume its eventual exception to + # avoid "Task exception was never retrieved" noise. + if not task.done(): + task.add_done_callback(_consume_detached_handler_exception) + raise def _acquire_platform_lock(self, scope: str, identity: str, resource_desc: str) -> bool: """Acquire a scoped lock for this adapter. Returns True on success. diff --git a/tests/gateway/test_notify_fatal_error_shield.py b/tests/gateway/test_notify_fatal_error_shield.py new file mode 100644 index 0000000000000..09c6fd36d27e2 --- /dev/null +++ b/tests/gateway/test_notify_fatal_error_shield.py @@ -0,0 +1,120 @@ +"""Regression test for #81335 — fatal-error handler must survive cancellation +of the task that awaits ``_notify_fatal_error``. + +The Telegram adapter escalates exhausted polling retries from inside its own +``_polling_error_task``. The gateway's fatal handler tears the adapter down +via ``disconnect()``, which cancels that very task. The handler used to be +killed mid-flight by the propagating ``CancelledError``: the adapter was +already popped from the gateway's adapter map, but the platform was never +queued for background reconnection — a zombie gateway. + +These tests model that carrier-cancellation race directly against +``BasePlatformAdapter._notify_fatal_error``. +""" + +import asyncio + +import pytest + +from gateway.platforms.base import BasePlatformAdapter + + +class _FakeAdapter: + """Minimal stand-in exposing only what ``_notify_fatal_error`` touches.""" + + _notify_fatal_error = BasePlatformAdapter._notify_fatal_error + + def __init__(self): + self._fatal_error_handler = None + self.handler_completed = False + + +@pytest.mark.asyncio +async def test_handler_survives_carrier_cancellation(): + """Handler must run to completion even when the awaiting task is + cancelled from inside the handler (the disconnect() self-cancel race).""" + adapter = _FakeAdapter() + carrier_task = None + + async def gateway_handler(a): + # Step 1: teardown — cancels the carrier task (what the real + # handler does indirectly via adapter.disconnect()). + carrier_task.cancel() + # Yield so the cancellation is delivered while we're still running. + await asyncio.sleep(0.05) + # Step 2: the part that never ran before the fix — queueing the + # platform for background reconnection. + a.handler_completed = True + + adapter._fatal_error_handler = gateway_handler + + async def carrier(): + await adapter._notify_fatal_error() + + carrier_task = asyncio.create_task(carrier()) + with pytest.raises(asyncio.CancelledError): + await carrier_task + + # Let the detached, shielded handler finish. + await asyncio.sleep(0.2) + + assert carrier_task.cancelled() + assert adapter.handler_completed, ( + "fatal-error handler was killed by carrier cancellation — platform " + "would never be queued for reconnection (zombie gateway, #81335)" + ) + + +@pytest.mark.asyncio +async def test_carrier_cancellation_still_propagates(): + """The carrier task itself must still observe CancelledError (teardown + semantics unchanged) — only the handler is shielded.""" + adapter = _FakeAdapter() + carrier_task = None + + async def gateway_handler(a): + carrier_task.cancel() + await asyncio.sleep(0.05) + a.handler_completed = True + + adapter._fatal_error_handler = gateway_handler + + async def carrier(): + await adapter._notify_fatal_error() + + carrier_task = asyncio.create_task(carrier()) + with pytest.raises(asyncio.CancelledError): + await carrier_task + assert carrier_task.cancelled() + + +@pytest.mark.asyncio +async def test_uncancelled_path_unchanged(): + """Normal path (no cancellation) behaves exactly as before.""" + adapter = _FakeAdapter() + + async def gateway_handler(a): + a.handler_completed = True + + adapter._fatal_error_handler = gateway_handler + await adapter._notify_fatal_error() + assert adapter.handler_completed + + +@pytest.mark.asyncio +async def test_sync_handler_still_supported(): + """Synchronous handlers (non-coroutine return) keep working.""" + adapter = _FakeAdapter() + + def gateway_handler(a): + a.handler_completed = True + + adapter._fatal_error_handler = gateway_handler + await adapter._notify_fatal_error() + assert adapter.handler_completed + + +@pytest.mark.asyncio +async def test_no_handler_is_noop(): + adapter = _FakeAdapter() + await adapter._notify_fatal_error() # must not raise