diff --git a/plugins/platforms/telegram/adapter.py b/plugins/platforms/telegram/adapter.py index 92d3c9fbcc3..40e7cb06371 100644 --- a/plugins/platforms/telegram/adapter.py +++ b/plugins/platforms/telegram/adapter.py @@ -32,7 +32,7 @@ def _consume_abandoned_task(task: asyncio.Task) -> None: logger.debug("Abandoned Telegram init task failed after timeout", exc_info=True) -async def _await_with_thread_deadline(awaitable, timeout: float): +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 @@ -41,6 +41,14 @@ async def _await_with_thread_deadline(awaitable, timeout: float): 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() @@ -68,10 +76,69 @@ async def _await_with_thread_deadline(awaitable, timeout: float): 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: @@ -2980,6 +3047,13 @@ class TelegramAdapter(BasePlatformAdapter): 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: 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 index 0c7ecede103..ca697276e8b 100644 --- a/tests/gateway/test_telegram_init_deadline.py +++ b/tests/gateway/test_telegram_init_deadline.py @@ -54,7 +54,7 @@ async def test_connect_retries_when_initialize_wall_deadline_expires(monkeypatch deadline_calls = 0 - async def _fake_deadline(awaitable, timeout): + async def _fake_deadline(awaitable, timeout, *, on_abandon=None): nonlocal deadline_calls deadline_calls += 1 if deadline_calls == 1: @@ -79,3 +79,123 @@ async def test_connect_retries_when_initialize_wall_deadline_expires(monkeypatch 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