From d50aae0e355255312f4ca742c7f1f18600b32464 Mon Sep 17 00:00:00 2001 From: yoma Date: Sat, 4 Jul 2026 19:56:49 +0800 Subject: [PATCH] fix(telegram): use wall deadline for init timeout --- plugins/platforms/telegram/adapter.py | 57 +++++++++++++- tests/gateway/test_telegram_init_deadline.py | 81 ++++++++++++++++++++ 2 files changed, 137 insertions(+), 1 deletion(-) create mode 100644 tests/gateway/test_telegram_init_deadline.py diff --git a/plugins/platforms/telegram/adapter.py b/plugins/platforms/telegram/adapter.py index 59930c2eb49..92d3c9fbcc3 100644 --- a/plugins/platforms/telegram/adapter.py +++ b/plugins/platforms/telegram/adapter.py @@ -15,11 +15,63 @@ import logging import os import html as _html import re +import threading from datetime import datetime, timezone from typing import Dict, List, Optional, Set, Any logger = logging.getLogger(__name__) + +def _consume_abandoned_task(task: asyncio.Task) -> None: + """Observe a detached task's terminal exception to avoid noisy loop logs.""" + try: + task.exception() + except asyncio.CancelledError: + pass + except Exception: + logger.debug("Abandoned Telegram init task failed after timeout", exc_info=True) + + +async def _await_with_thread_deadline(awaitable, timeout: float): + """Await with a wall-clock deadline that does not depend on loop timers. + + ``asyncio.wait_for`` schedules its timeout on the event loop and then waits + for cancellation to propagate. PTB/httpcore initialization can sit inside + cancellation-shielded anyio scopes, so a timed-out initialize() may never + hand control back to the retry ladder under some supervisors. This helper + lets a daemon ``threading.Timer`` wake the loop and, on timeout, abandons + the shielded task instead of awaiting cancellation completion. + """ + task = asyncio.ensure_future(awaitable) + loop = asyncio.get_running_loop() + deadline = loop.create_future() + + def _mark_expired() -> None: + if not deadline.done(): + deadline.set_result(None) + + def _expire_from_thread() -> None: + loop.call_soon_threadsafe(_mark_expired) + + timer = threading.Timer(max(timeout, 0.0), _expire_from_thread) + timer.daemon = True + timer.start() + try: + done, _ = await asyncio.wait( + {task, deadline}, + return_when=asyncio.FIRST_COMPLETED, + ) + if task in done: + if not deadline.done(): + deadline.cancel() + return await task + + task.cancel() + task.add_done_callback(_consume_abandoned_task) + raise asyncio.TimeoutError() + finally: + timer.cancel() + try: from telegram import Update, Bot, Message, InlineKeyboardButton, InlineKeyboardMarkup try: @@ -2925,7 +2977,10 @@ class TelegramAdapter(BasePlatformAdapter): "[%s] Connecting to Telegram (attempt %d/%d)…", self.name, _attempt + 1, _max_connect, ) - await asyncio.wait_for(self._app.initialize(), timeout=_init_timeout) + await _await_with_thread_deadline( + self._app.initialize(), + timeout=_init_timeout, + ) break except asyncio.TimeoutError: if _attempt < _max_connect - 1: diff --git a/tests/gateway/test_telegram_init_deadline.py b/tests/gateway/test_telegram_init_deadline.py new file mode 100644 index 00000000000..0c7ecede103 --- /dev/null +++ b/tests/gateway/test_telegram_init_deadline.py @@ -0,0 +1,81 @@ +import sys +from unittest.mock import AsyncMock, MagicMock + +import pytest + +from gateway.config import PlatformConfig + + +def _ensure_telegram_mock(): + if "telegram" in sys.modules and hasattr(sys.modules["telegram"], "__file__"): + return + telegram_mod = MagicMock() + telegram_mod.ext.ContextTypes.DEFAULT_TYPE = type(None) + telegram_mod.constants.ParseMode.MARKDOWN_V2 = "MarkdownV2" + telegram_mod.constants.ChatType.GROUP = "group" + telegram_mod.constants.ChatType.SUPERGROUP = "supergroup" + telegram_mod.constants.ChatType.CHANNEL = "channel" + telegram_mod.constants.ChatType.PRIVATE = "private" + telegram_mod.error.NetworkError = type("NetworkError", (OSError,), {}) + telegram_mod.error.TimedOut = type("TimedOut", (OSError,), {}) + for name in ("telegram", "telegram.ext", "telegram.constants", "telegram.request"): + sys.modules.setdefault(name, telegram_mod) + sys.modules.setdefault("telegram.error", telegram_mod.error) + + +_ensure_telegram_mock() + +from plugins.platforms.telegram import adapter as tg_adapter # noqa: E402 +from plugins.platforms.telegram.adapter import TelegramAdapter # noqa: E402 + + +@pytest.mark.asyncio +async def test_connect_retries_when_initialize_wall_deadline_expires(monkeypatch): + """A wedged initialize() attempt must not trap startup on attempt 1/8.""" + fake_app = MagicMock() + fake_app.bot = MagicMock() + fake_app.initialize = AsyncMock(return_value=None) + fake_app.start = AsyncMock() + fake_app.add_handler = MagicMock() + + chainable = MagicMock() + chainable.token.return_value = chainable + chainable.request.return_value = chainable + chainable.get_updates_request.return_value = chainable + chainable.build.return_value = fake_app + + builder_root = MagicMock() + builder_root.builder.return_value = chainable + monkeypatch.setattr(tg_adapter, "Application", builder_root) + monkeypatch.setattr(tg_adapter, "HTTPXRequest", MagicMock) + monkeypatch.setattr(tg_adapter, "discover_fallback_ips", AsyncMock(return_value=[])) + monkeypatch.setattr(tg_adapter, "resolve_proxy_url", lambda *a, **k: None) + monkeypatch.setattr(tg_adapter.asyncio, "sleep", AsyncMock()) + + deadline_calls = 0 + + async def _fake_deadline(awaitable, timeout): + nonlocal deadline_calls + deadline_calls += 1 + if deadline_calls == 1: + awaitable.close() + raise tg_adapter.asyncio.TimeoutError() + return await awaitable + + monkeypatch.setattr(tg_adapter, "_await_with_thread_deadline", _fake_deadline) + + adapter = TelegramAdapter(PlatformConfig(enabled=True, token="test-token")) + monkeypatch.setattr(adapter, "_acquire_platform_lock", lambda *a, **k: True) + monkeypatch.setattr(adapter, "_fallback_ips", lambda: []) + monkeypatch.setattr(adapter, "_delete_webhook_best_effort", AsyncMock()) + monkeypatch.setattr(adapter, "_start_polling_resilient", AsyncMock(return_value=True)) + monkeypatch.setattr(adapter, "_polling_heartbeat_loop", AsyncMock(return_value=None)) + monkeypatch.setattr(adapter, "_start_post_connect_housekeeping", MagicMock()) + + assert await adapter.connect() is True + + assert fake_app.initialize.call_count == 2 + assert fake_app.initialize.await_count == 1 + assert deadline_calls == 2 + tg_adapter.asyncio.sleep.assert_awaited_once_with(1) + fake_app.start.assert_awaited_once()