diff --git a/plugins/platforms/telegram/adapter.py b/plugins/platforms/telegram/adapter.py index 60e2d7bb9ee1..5405728ed3a5 100644 --- a/plugins/platforms/telegram/adapter.py +++ b/plugins/platforms/telegram/adapter.py @@ -545,6 +545,10 @@ _UPDATER_STOP_TIMEOUT = 15.0 # reconnect ladder from stalling indefinitely and allows the heartbeat loop to # trigger its own recovery path. Refs: NousResearch/hermes-agent#59614 _UPDATER_START_TIMEOUT = 30.0 +# Initial connect is not healthy until the dedicated getUpdates request completes +# one successful round trip. Unlike reconnect, initial bootstrap must fail closed +# so GatewayRunner disposes the partial adapter and retries with a fresh PTB app. +_INITIAL_POLLING_PROGRESS_TIMEOUT = 60.0 # shutdown()/initialize() on the getUpdates httpx request close and rebuild the # connection pool. When a connection is wedged on a stale CLOSE-WAIT socket that # close can block forever, hanging _drain_polling_connections() and freezing the @@ -2129,6 +2133,7 @@ class TelegramAdapter(BasePlatformAdapter): *, drop_pending_updates: bool, error_callback, + abandon_app_on_timeout: bool = False, ) -> None: """Start one generation and verify real getUpdates progress.""" if getattr(self, "_polling_teardown_started", False): @@ -2151,13 +2156,22 @@ class TelegramAdapter(BasePlatformAdapter): context_token = _POLLING_GENERATION_CONTEXT.set(generation) try: - await asyncio.wait_for( + # asyncio.wait_for can wait forever for cancellation to escape + # httpcore/AnyIO shielded scopes (#58236/#67498). Reuse the + # proven wall-deadline helper and abandon the partial updater; + # caller recovery will dispose/rebuild the whole adapter. + await _await_with_thread_deadline( app.updater.start_polling( allowed_updates=Update.ALL_TYPES, drop_pending_updates=drop_pending_updates, error_callback=_generation_error_callback, ), timeout=_UPDATER_START_TIMEOUT, + on_abandon=( + (lambda app=app: _shutdown_abandoned_app(app)) + if abandon_app_on_timeout + else None + ), ) finally: _POLLING_GENERATION_CONTEXT.reset(context_token) @@ -2264,12 +2278,14 @@ class TelegramAdapter(BasePlatformAdapter): self._background_tasks.add(self._polling_error_task) self._polling_error_task.add_done_callback(self._background_tasks.discard) - async def _delete_webhook_best_effort(self) -> bool: - """Clear any stale webhook, but never fail polling on a network error. + async def _delete_webhook_best_effort( + self, *, require_success: bool = False + ) -> bool: + """Clear stale webhook, optionally failing closed on initial connect. - Returns True when the webhook was cleared (or there was nothing to do) - and False when a transient network error was swallowed so bootstrap can - continue to polling; the reconnect ladder recovers from there. + Reconnect can recover a transient error in background. Cold startup uses + ``require_success`` so GatewayRunner disposes the partial adapter and + retries with a fresh PTB Application instead of publishing degraded state. """ if not self._bot: return False @@ -2277,10 +2293,19 @@ class TelegramAdapter(BasePlatformAdapter): if not callable(delete_webhook): return True try: - await delete_webhook(drop_pending_updates=False) + # Same shielded-cancellation class as initialize/start_polling: + # never let a wedged duplicate deleteWebhook pin initial connect. + await _await_with_thread_deadline( + delete_webhook(drop_pending_updates=False), + timeout=_UPDATER_START_TIMEOUT, + ) return True except Exception as err: if self._looks_like_network_error(err): + if require_success: + raise OSError( + "Telegram deleteWebhook did not complete during initial connect" + ) from err logger.warning( "[%s] deleteWebhook failed with a recoverable network error; " "continuing to polling so getUpdates/retry can recover: %s", @@ -2290,12 +2315,19 @@ class TelegramAdapter(BasePlatformAdapter): return False raise - async def _start_polling_resilient(self, *, drop_pending_updates: bool, error_callback) -> bool: - """Start PTB polling; on a transient bootstrap failure, recover in background. + async def _start_polling_resilient( + self, + *, + drop_pending_updates: bool, + error_callback, + require_progress: bool = False, + ) -> bool: + """Start PTB polling and optionally require real getUpdates readiness. - Returns True when polling started, False when a transient conflict or - network error was scheduled for background recovery instead of raising - (keeping the gateway process alive). + Reconnects may recover in background. Initial connect sets + ``require_progress`` so a bootstrap failure or missing first successful + getUpdates response raises; GatewayRunner then disposes this partial + adapter and retries with a fresh PTB Application. """ if getattr(self, "_polling_teardown_started", False): return False @@ -2312,13 +2344,30 @@ class TelegramAdapter(BasePlatformAdapter): self._app, drop_pending_updates=drop_pending_updates, error_callback=error_callback, + abandon_app_on_timeout=require_progress, ) + if require_progress: + try: + await _await_with_thread_deadline( + self._polling_progress_event.wait(), + timeout=_INITIAL_POLLING_PROGRESS_TIMEOUT, + ) + except asyncio.TimeoutError as exc: + raise OSError( + "Telegram getUpdates made no progress during initial connect" + ) from exc + if not self._polling_progress_event.is_set(): + raise OSError( + "Telegram getUpdates did not become ready during initial connect" + ) return True except _PollingLifecycleAbort: return False except Exception as err: if getattr(self, "_polling_teardown_started", False): return False + if require_progress: + raise if self._looks_like_polling_conflict(err): logger.warning( "[%s] Telegram polling bootstrap conflict; gateway stays alive " @@ -3815,7 +3864,9 @@ class TelegramAdapter(BasePlatformAdapter): # updates. Best-effort: a transient Bot API network error here # must not fail gateway startup — degrade to background polling # recovery instead. - await self._delete_webhook_best_effort() + await self._delete_webhook_best_effort( + require_success=not is_reconnect + ) loop = asyncio.get_running_loop() @@ -3853,6 +3904,7 @@ class TelegramAdapter(BasePlatformAdapter): # sent while the bot was offline are delivered (#46621). drop_pending_updates=not is_reconnect, error_callback=_polling_error_callback, + require_progress=not is_reconnect, ) if not polling_started: logger.warning( diff --git a/tests/gateway/test_telegram_polling_progress.py b/tests/gateway/test_telegram_polling_progress.py index fdf00e17c99b..9f075c189cd9 100644 --- a/tests/gateway/test_telegram_polling_progress.py +++ b/tests/gateway/test_telegram_polling_progress.py @@ -138,13 +138,20 @@ async def test_polling_disconnect_webhook_reconnect_heals_webhook_send_path(monk adapter = _make_adapter() polling_app = _lifecycle_app() webhook_app = _lifecycle_app() + + async def start_polling_with_progress(**_kwargs): + adapter._record_polling_progress(adapter._polling_generation) + + polling_app.updater.start_polling = AsyncMock( + side_effect=start_polling_with_progress + ) _configure_lifecycle_connect(monkeypatch, adapter, [polling_app, webhook_app]) monkeypatch.delenv("TELEGRAM_WEBHOOK_URL", raising=False) monkeypatch.delenv("TELEGRAM_WEBHOOK_SECRET", raising=False) assert await adapter.connect() is True assert adapter._webhook_mode is False - assert adapter._send_path_degraded is True + assert adapter._send_path_degraded is False await adapter.disconnect() monkeypatch.setenv("TELEGRAM_WEBHOOK_URL", "https://example.test/telegram") diff --git a/tests/gateway/test_telegram_start_polling_timeout.py b/tests/gateway/test_telegram_start_polling_timeout.py index 12e4ef97b7df..27e3fc712b01 100644 --- a/tests/gateway/test_telegram_start_polling_timeout.py +++ b/tests/gateway/test_telegram_start_polling_timeout.py @@ -8,9 +8,10 @@ httpx connection pool — it neither returns nor raises. Unbounded, that wedges: 2. the heartbeat loop (sees the recovery task as alive-but-wedged and skips), 3. the fatal-error escalation (never reached). -The fix wraps every ``start_polling()`` await in ``asyncio.wait_for`` with -``_UPDATER_START_TIMEOUT`` so a hung call raises and feeds the existing retry -ladder. These tests patch the timeout down to keep the suite fast. +The fix wraps every ``start_polling()`` await in the wall-deadline helper with +``_UPDATER_START_TIMEOUT`` so a cancellation-shielded hung call still raises and +feeds the existing retry ladder. These tests patch the timeout down to keep the +suite fast. """ import asyncio import sys @@ -138,20 +139,34 @@ async def test_start_polling_success_path_unaffected(monkeypatch): def test_every_start_polling_call_site_is_time_bounded(): - """Bug-class contract: every `updater.start_polling(` await in the adapter - must be wrapped in asyncio.wait_for. A new unbounded call site reintroduces - the #59614 wedge.""" + """Every updater.start_polling call must use the wall-deadline helper.""" import inspect import re src = inspect.getsource(tg_adapter) - # Find each start_polling( call and check an enclosing wait_for within the - # preceding 6 lines (the wrapper always sits directly above). lines = src.splitlines() unbounded = [] for i, line in enumerate(lines): if re.search(r"updater\.start_polling\(", line) and "def " not in line: - window = "\n".join(lines[max(0, i - 6):i + 1]) - if "wait_for" not in window: + window = "\n".join(lines[max(0, i - 8):i + 1]) + if "_await_with_thread_deadline" not in window: unbounded.append((i + 1, line.strip())) assert not unbounded, f"unbounded start_polling() call sites: {unbounded}" + + +@pytest.mark.asyncio +async def test_initial_connect_requires_get_updates_progress(monkeypatch): + """Cold connect must not claim success before real getUpdates I/O.""" + monkeypatch.setattr(tg_adapter, "_INITIAL_POLLING_PROGRESS_TIMEOUT", 0.05) + a = _bare_adapter() + app = MagicMock() + app.updater = AsyncMock() + app.updater.start_polling = AsyncMock(return_value=None) + a._app = app + + with pytest.raises(OSError, match="getUpdates made no progress"): + await a._start_polling_resilient( + drop_pending_updates=True, + error_callback=None, + require_progress=True, + )