diff --git a/plugins/platforms/telegram/adapter.py b/plugins/platforms/telegram/adapter.py index 59930c2eb49..40e7cb06371 100644 --- a/plugins/platforms/telegram/adapter.py +++ b/plugins/platforms/telegram/adapter.py @@ -15,11 +15,130 @@ 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, *, on_abandon=None): + """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. + + ``on_abandon`` (optional) is a zero-arg callable returning an awaitable that + is scheduled as a detached best-effort cleanup when the task is abandoned on + timeout. The abandoned initialize() may leave a half-built httpx client / + connection pool open (it never completed and we do not await its + cancellation), so the caller uses this to shut that state down and avoid + leaking a pool per retry attempt. Cleanup runs detached and its own errors + are swallowed, so it can never re-block the retry ladder. + """ + 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) + if on_abandon is not None: + # Detached best-effort cleanup: close the half-built app's httpx + # client/pool so an abandoned attempt can't leak sockets across the + # retry ladder. Detached + exception-observed so it never re-blocks + # or re-hangs the ladder we are trying to advance. + cleanup = asyncio.ensure_future(_run_abandon_cleanup(on_abandon)) + cleanup.add_done_callback(_consume_abandoned_task) + raise asyncio.TimeoutError() + finally: + timer.cancel() + + +async def _run_abandon_cleanup(on_abandon) -> None: + """Run the abandonment cleanup coroutine, swallowing any failure. + + Wrapped so a cleanup that itself hangs or raises cannot surface as an + unhandled task error or block anything — it is fully fire-and-forget. + """ + try: + result = on_abandon() + if asyncio.iscoroutine(result) or asyncio.isfuture(result): + await result + except Exception: + logger.debug("Abandoned Telegram init cleanup failed", exc_info=True) + + +async def _shutdown_abandoned_app(app) -> None: + """Release a half-built PTB app's httpx transports after init was abandoned. + + ``Application.shutdown()`` / ``Bot.shutdown()`` are gated on the app's + ``_initialized`` / ``_requests_initialized`` flags, which a wedged + ``initialize()`` (the case this whole path exists for) may never have set — + so calling only ``app.shutdown()`` no-ops and leaks the connection pool it + was meant to close. ``HTTPXRequest`` builds its ``httpx.AsyncClient`` + eagerly in its constructor and its ``shutdown()`` gates only on + ``client.is_closed``, so closing the request transports directly releases + the pool regardless of PTB init state. We try the clean path first, then + fall back to the transports. All best-effort and swallowed. + """ + if app is None: + return + try: + await app.shutdown() + except Exception: + logger.debug("Abandoned Telegram app.shutdown() failed", exc_info=True) + # Directly close the underlying request transports (bypasses PTB's + # init-gated shutdown so the eagerly-built httpx pool is released even when + # the abandoned initialize() never flipped _initialized). + bot = getattr(app, "bot", None) + requests = getattr(bot, "_request", None) if bot is not None else None + if not requests: + return + for request in requests: + shutdown = getattr(request, "shutdown", None) + if shutdown is None: + continue + try: + result = shutdown() + if asyncio.iscoroutine(result) or asyncio.isfuture(result): + await result + except Exception: + logger.debug("Abandoned Telegram request shutdown failed", exc_info=True) + try: from telegram import Update, Bot, Message, InlineKeyboardButton, InlineKeyboardMarkup try: @@ -2925,7 +3044,17 @@ 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, + # On timeout the initialize() task is abandoned without + # awaiting its cancellation (it may be wedged in a + # shielded scope). Best-effort release the half-built + # app's httpx client/connection pool so it isn't leaked + # across the retry ladder (mirrors the client-close-on- + # timeout pattern in agent/auxiliary_client.py). + on_abandon=lambda app=self._app: _shutdown_abandoned_app(app), + ) break except asyncio.TimeoutError: if _attempt < _max_connect - 1: diff --git a/scripts/release.py b/scripts/release.py index ffebac6b3d2..0ab5cc2aabe 100755 --- a/scripts/release.py +++ b/scripts/release.py @@ -45,6 +45,7 @@ ACP_REGISTRY_MANIFEST = REPO_ROOT / "acp_registry" / "agent.json" # Auto-extracted from noreply emails + manual overrides AUTHOR_MAP = { + "yingwaizhiying@gmail.com": "msh01", # PR #58250 salvage (telegram: wall-clock init timeout via daemon-thread deadline + abandon the shielded initialize task on timeout so the retry ladder advances instead of hanging on attempt 1/8 under s6 supervision; #58236) "huanshan5195@users.noreply.github.com": "huanshan5195", # PR #57601 salvage (custom-provider: emit reasoning_effort at the live CustomProfile path so GLM-5.2/ARK/vLLM/Ollama endpoints receive it; + "max" reasoning level) "infinitycrew39@gmail.com": "infinitycrew39", # PR #56431 salvage (honor live vLLM context limits on local endpoints) "jonathan.kovacs999@gmail.com": "CocaKova", # PR #57692 salvage (cron: run jobs under the profile secret scope so get_secret does not fail-close with UnscopedSecretError under profile isolation) diff --git a/tests/gateway/test_telegram_init_deadline.py b/tests/gateway/test_telegram_init_deadline.py new file mode 100644 index 00000000000..ca697276e8b --- /dev/null +++ b/tests/gateway/test_telegram_init_deadline.py @@ -0,0 +1,201 @@ +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, *, on_abandon=None): + 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() + + +@pytest.mark.asyncio +async def test_await_with_thread_deadline_returns_value_on_happy_path(): + """The real helper returns the awaited result and raises no timeout.""" + async def _ok(): + return 42 + + result = await tg_adapter._await_with_thread_deadline(_ok(), timeout=5.0) + assert result == 42 + + +@pytest.mark.asyncio +async def test_await_with_thread_deadline_abandons_and_runs_cleanup_on_timeout(): + """A wedged awaitable must raise TimeoutError promptly AND trigger the + best-effort on_abandon cleanup (the httpx-pool-leak guard). + + This exercises the REAL _await_with_thread_deadline (not a monkeypatched + stub), covering the abandonment + cleanup mechanism directly. + """ + import asyncio as _asyncio + import time as _time + + cleanup_ran = _asyncio.Event() + + async def _wedged(): + # Swallows cancellation for a bounded window — long enough that the + # helper must return control BEFORE this finishes (proving it doesn't + # await cancellation, the #58236 shielded-scope behavior), but bounded + # so the abandoned task can't outlive the test and wedge teardown. + for _ in range(20): + try: + await _asyncio.sleep(0.05) + except _asyncio.CancelledError: + # Keep going despite cancellation, like the shielded scope. + pass + + async def _cleanup(): + cleanup_ran.set() + + started = _time.monotonic() + with pytest.raises(_asyncio.TimeoutError): + await tg_adapter._await_with_thread_deadline( + _wedged(), timeout=0.2, on_abandon=_cleanup + ) + elapsed = _time.monotonic() - started + + # Returned control promptly — well before the wedged coroutine's ~1s span. + assert elapsed < 0.8 + # The detached cleanup was scheduled; give the loop a tick to run it. + await _asyncio.wait_for(cleanup_ran.wait(), timeout=2.0) + assert cleanup_ran.is_set() + + +@pytest.mark.asyncio +async def test_await_with_thread_deadline_cleanup_error_is_swallowed(): + """A cleanup that raises must not surface as an unhandled task error.""" + import asyncio as _asyncio + + async def _wedged(): + for _ in range(20): + try: + await _asyncio.sleep(0.05) + except _asyncio.CancelledError: + pass + + def _boom(): + raise RuntimeError("cleanup blew up") + + # Must still raise TimeoutError (not the cleanup error) and not crash. + with pytest.raises(_asyncio.TimeoutError): + await tg_adapter._await_with_thread_deadline( + _wedged(), timeout=0.2, on_abandon=_boom + ) + # Let the detached cleanup task run and be observed (no unraised error). + await _asyncio.sleep(0.05) + + +@pytest.mark.asyncio +async def test_shutdown_abandoned_app_closes_request_transports_when_uninitialized(): + """The leak fix must release the httpx transports even when PTB's own + Application.shutdown()/Bot.shutdown() no-op because the wedged initialize() + never flipped _initialized. _shutdown_abandoned_app falls back to closing + each bot._request transport directly (HTTPXRequest.shutdown gates only on + client.is_closed, not on an init flag).""" + from unittest.mock import AsyncMock, MagicMock + + # A half-built app: shutdown() is a no-op (uninitialized), but the request + # transports still hold open httpx clients that must be closed. + req0 = MagicMock() + req0.shutdown = AsyncMock() + req1 = MagicMock() + req1.shutdown = AsyncMock() + bot = MagicMock() + bot._request = (req0, req1) + app = MagicMock() + app.bot = bot + app.shutdown = AsyncMock(return_value=None) # PTB no-op on uninitialized app + + await tg_adapter._shutdown_abandoned_app(app) + + app.shutdown.assert_awaited_once() + # Fell back to closing the transports directly — the actual leak fix. + req0.shutdown.assert_awaited_once() + req1.shutdown.assert_awaited_once() + + +@pytest.mark.asyncio +async def test_shutdown_abandoned_app_handles_none_and_missing_requests(): + """Robust against app=None and an app whose bot/_request aren't present.""" + from unittest.mock import AsyncMock, MagicMock + + # None app -> no-op, no crash. + await tg_adapter._shutdown_abandoned_app(None) + + # app.shutdown() raising must be swallowed, and missing _request tolerated. + app = MagicMock() + app.shutdown = AsyncMock(side_effect=RuntimeError("still running")) + app.bot = None + await tg_adapter._shutdown_abandoned_app(app) # must not raise