Telegram fatal recovery handoff during teardown or retry exhaustion

Fixes on record

Merged to main. A stamp counts the issues a fix is linked to.

• #68447 fix(telegram): preserve fatal recovery handoff from self-cancelling teardown

  merged to main

  files: plugins/platforms/telegram/adapter.py, tests/gateway/test_telegram_conflict.py, tests/gateway/test_telegram_network_reconnect.py

  Show the diff

  • plugins/platforms/telegram/adapter.py

    @@ -2357,11 +2357,11 @@ async def _handle_polling_network_error(self, error: Exception) -> None:
             if attempt > MAX_NETWORK_RETRIES:
                 message = (
                     "Telegram polling could not reconnect after %d network error retries. "
    -                "Restarting gateway." % MAX_NETWORK_RETRIES
    +                "Escalating to gateway recovery." % MAX_NETWORK_RETRIES
                 )
                 logger.error("[%s] %s Last error: %s", self.name, message, _redact_telegram_error_text(error))
                 self._set_fatal_error("telegram_network_error", message, retryable=True)
    -            await self._notify_fatal_error()
    +            await self._handoff_polling_fatal_error()
                 return
     
             delay = min(BASE_DELAY * (2 ** (attempt - 1)), MAX_DELAY)
    @@ -2523,7 +2523,7 @@ async def _polling_heartbeat_loop(self) -> None:
                                 "gateway reconnect." % stuck_for,
                                 retryable=True,
                             )
    -                        await self._notify_fatal_error()
    +                        await self._handoff_polling_fatal_error()
                             return
                     else:
                         stuck_task_ref = None
    @@ -2982,7 +2982,24 @@ async def _handle_polling_conflict(self, error: Exception) -> None:
                     self.name, stop_error, exc_info=True,
                 )
             if not _already_fatal:
    -            await self._notify_fatal_error()
    +            await self._handoff_polling_fatal_error()
    +
    +    async def _handoff_polling_fatal_error(self) -> None:
    +        """Notify the runner without letting child teardown cancel this owner.
    +
    +        The runner bounds adapter cleanup in a child task.  ``disconnect()``
    +        cancels the tracked polling-recovery task and the heartbeat task, so
    +        retaining the current notifier in either field would cancel the fatal
    +        callback before the runner can finish its reconnect or shutdown
    +        decision.  Release only the current owner from whichever field tracks
    +        it; unrelated tasks remain under teardown control.
    +        """
    +        current_task = asyncio.current_task()
    +        if self._polling_error_task is current_task:
    +            self._polling_error_task = None
    +        if getattr(self, "_polling_heartbeat_task", None) is current_task:
    +            self._polling_heartbeat_task = None
    +        await self._notify_fatal_error()
     
         async def _create_dm_topic(
             self,

  • tests/gateway/test_telegram_conflict.py

    @@ -293,6 +293,32 @@ async def failing_start_polling(**kwargs):
         await _cancel_heartbeat(adapter)
     
     
    [redacted:email]
    +async def test_conflict_exhaustion_hands_off_before_child_disconnect():
    +    """The conflict recovery owner must survive its fatal callback handoff."""
    +    adapter = TelegramAdapter(PlatformConfig(enabled=True, token="***"))
    +    adapter._polling_conflict_count = 5  # MAX_CONFLICT_RETRIES
    +    disconnect_tasks = []
    +
    +    async def fatal_handler(failed_adapter):
    +        disconnect_task = asyncio.create_task(failed_adapter.disconnect())
    +        disconnect_tasks.append(disconnect_task)
    +        await asyncio.wait({disconnect_task})
    +
    +    adapter.set_fatal_error_handler(fatal_handler)
    +
    +    conflict = type("Conflict", (Exception,), {})
    +    recovery_task = asyncio.create_task(
    +        adapter._handle_polling_conflict(conflict("getUpdates conflict"))
    +    )
    +    adapter._polling_error_task = recovery_task
    +    result = await asyncio.gather(recovery_task, return_exceptions=True)
    +    await asyncio.gather(*disconnect_tasks, return_exceptions=True)
    +
    +    assert result == [None]
    +    assert adapter._polling_error_task is None
    +
    +
     @pytest.mark.asyncio
     async def test_connect_marks_retryable_fatal_error_for_startup_network_failure(monkeypatch):
         adapter = TelegramAdapter(PlatformConfig(enabled=True, token="***"))
    @@ -730,4 +756,3 @@ async def fake_start_polling(**kwargs):
         for _ in range(10):
             await asyncio.sleep(0)
         await _cancel_heartbeat(adapter)
    -

  • tests/gateway/test_telegram_network_reconnect.py

    @@ -14,7 +14,7 @@
     
     import pytest
     
    -from gateway.config import PlatformConfig
    +from gateway.config import GatewayConfig, Platform, PlatformConfig
     
     
     def _ensure_telegram_mock():
    @@ -37,6 +37,7 @@ def _ensure_telegram_mock():
     
     from plugins.platforms.telegram import adapter as tg_adapter  # noqa: E402
     from plugins.platforms.telegram.adapter import TelegramAdapter  # noqa: E402
    +from gateway.run import GatewayRunner  # noqa: E402
     
     
     @pytest.fixture(autouse=True)
    @@ -224,6 +225,76 @@ async def test_reconnect_triggers_fatal_after_max_retries():
         fatal_handler.assert_called_once()
     
     
    [redacted:email]
    +async def test_retry_exhaustion_queues_reconnect_before_child_disconnect(tmp_path):
    +    """Fatal teardown must not cancel the gateway's reconnect handoff.
    +
    +    The gateway runs ``disconnect()`` in a bounded child task.  If the current
    +    polling-recovery owner remains in ``_polling_error_task``, Telegram teardown
    +    cancels that parent while it is still awaiting the fatal handler, so the
    +    handler never gets to queue background reconnection.
    +    """
    +    config = GatewayConfig(
    +        platforms={
    +            Platform.TELEGRAM: PlatformConfig(enabled=True, token="[redacted:secret-assignment]")
    +        },
    +        sessions_dir=tmp_path / "sessions",
    +    )
    +    runner = GatewayRunner(config)
    +    adapter = _make_adapter()
    +    adapter._polling_network_error_count = 10  # MAX_NETWORK_RETRIES
    +    adapter.set_fatal_error_handler(runner._handle_adapter_fatal_error)
    +    runner.adapters = {Platform.TELEGRAM: adapter}
    +    runner.delivery_router.adapters = runner.adapters
    +
    +    recovery_task = asyncio.create_task(
    +        adapter._handle_polling_network_error(Exception("still failing"))
    +    )
    +    adapter._polling_error_task = recovery_task
    +    result = await asyncio.gather(recovery_task, return_exceptions=True)
    +
    +    assert result == [None]
    +    assert runner.adapters == {}
    +    assert Platform.TELEGRAM in runner._failed_platforms
    +    assert runner._failed_platforms[Platform.TELEGRAM]["attempts"] == 0
    +
    +
    [redacted:email]
    +async def test_heartbeat_watchdog_handoff_survives_child_disconnect(tmp_path):
    +    """The wedged-recovery heartbeat watchdog must survive its fatal callback.
    +
    +    The heartbeat loop force-escalates a stuck polling-recovery task.  Like
    +    the network/conflict terminal paths, the heartbeat task itself is the
    +    owner that ``disconnect()`` cancels, so the fatal callback must release
    +    ``_polling_heartbeat_task`` before notifying the runner.
    +    """
    +    config = GatewayConfig(
    +        platforms={
    +            Platform.TELEGRAM: PlatformConfig(enabled=True, token="[redacted:secret-assignment]")
    +        },
    +        sessions_dir=tmp_path / "sessions",
    +    )
    +    runner = GatewayRunner(config)
    +    adapter = _make_adapter()
    +    adapter.set_fatal_error_handler(runner._handle_adapter_fatal_error)
    +    runner.adapters = {Platform.TELEGRAM: adapter}
    +    runner.delivery_router.adapters = runner.adapters
    +
    +    # Simulate the heartbeat watchdog's fatal-escalation path directly.
    +    adapter._set_fatal_error(
    +        "telegram_network_error",
    +        "Telegram reconnect task wedged; forcing gateway reconnect.",
    +        retryable=True,
    +    )
    +    heartbeat_task = asyncio.create_task(adapter._handoff_polling_fatal_error())
    +    adapter._polling_heartbeat_task = heartbeat_task
    +    result = await asyncio.gather(heartbeat_task, return_exceptions=True)
    +
    +    assert result == [None]
    +    assert runner.adapters == {}
    +    assert Platform.TELEGRAM in runner._failed_platforms
    +
    +
     # ---------------------------------------------------------------------------
     # Connection pool drain tests (PR #16466 salvage)
     # ---------------------------------------------------------------------------