From c8ff720508863babb4cefb548f363ed3a7002323 Mon Sep 17 00:00:00 2001 From: teknium1 <127238744+teknium1@users.noreply.github.com> Date: Fri, 24 Jul 2026 16:55:25 -0700 Subject: [PATCH] fix(telegram): bind strict cold-start readiness to its own polling generation MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Follow-up hardening for the salvaged #69240 readiness gate (#67498): - _start_polling_once now returns its (generation, progress_event) pair so the strict cold-start gate binds to exactly the generation it started, instead of re-reading self._polling_progress_event which a concurrent recovery task may have replaced with a newer generation's event (the G1/G2 race flagged in the #69240 review). - Strict cold start no longer schedules background polling recovery: a polling error during the readiness wait is captured by a strict callback and fails the connect attempt immediately with a loud OSError, so GatewayRunner disposes the partial adapter and retries with a fresh one — no more waiting out the full readiness deadline on a generation that already errored, and no G2-on-partial-app healing. - After readiness is proven the strict callback delegates every later polling error to the real background-recovery callback, preserving the existing degraded/reconnect semantics for the polling lifetime. - The readiness-timeout error message now states the deadline and that the gateway will retry with a fresh adapter (loud failure, not a silent wait). - Regression tests: current-generation progress connects; a polling error during strict cold start fails fast without scheduling background recovery (the #67498 idle-threads shape); stale-generation progress is rejected. Progresses #67498 --- .../mannnrachman@users.noreply.github.com | 1 + plugins/platforms/telegram/adapter.py | 98 ++++++++++++++-- .../test_telegram_start_polling_timeout.py | 107 ++++++++++++++++++ 3 files changed, 198 insertions(+), 8 deletions(-) create mode 100644 contributors/emails/mannnrachman@users.noreply.github.com diff --git a/contributors/emails/mannnrachman@users.noreply.github.com b/contributors/emails/mannnrachman@users.noreply.github.com new file mode 100644 index 000000000000..bed3dbde7b5d --- /dev/null +++ b/contributors/emails/mannnrachman@users.noreply.github.com @@ -0,0 +1 @@ +mannnrachman diff --git a/plugins/platforms/telegram/adapter.py b/plugins/platforms/telegram/adapter.py index 5405728ed3a5..64708ff6882c 100644 --- a/plugins/platforms/telegram/adapter.py +++ b/plugins/platforms/telegram/adapter.py @@ -155,6 +155,16 @@ async def _await_with_thread_deadline(awaitable, timeout: float, *, on_abandon=N loop_processed_expiry.set() +async def _first_completed(*futures: "asyncio.Future") -> None: + """Return when the first of ``futures`` completes. + + Used by the strict cold-start readiness gate to wait on "progress OR + polling error", whichever fires first (#67498). Does not cancel the + losers — the caller owns their lifecycle. + """ + await asyncio.wait(set(futures), return_when=asyncio.FIRST_COMPLETED) + + async def _run_abandon_cleanup(on_abandon) -> None: """Run the abandonment cleanup coroutine, swallowing any failure. @@ -2134,8 +2144,16 @@ class TelegramAdapter(BasePlatformAdapter): drop_pending_updates: bool, error_callback, abandon_app_on_timeout: bool = False, - ) -> None: - """Start one generation and verify real getUpdates progress.""" + schedule_verifier: bool = True, + ) -> tuple[int, asyncio.Event]: + """Start one generation and verify real getUpdates progress. + + Returns the ``(generation, progress_event)`` pair created for this + polling generation so callers that must gate on readiness (strict + cold start, #67498) can bind to exactly this generation instead of + re-reading ``self._polling_progress_event`` — which a concurrent + recovery task may have replaced with a newer generation's event. + """ if getattr(self, "_polling_teardown_started", False): raise _PollingLifecycleAbort("Telegram polling teardown started") generation, progress = self._begin_polling_generation() @@ -2179,7 +2197,9 @@ class TelegramAdapter(BasePlatformAdapter): self._polling_progress_accepting = False self._send_path_degraded = True raise _PollingLifecycleAbort("Telegram polling teardown started") - self._schedule_polling_progress_verifier(generation, progress) + if schedule_verifier: + self._schedule_polling_progress_verifier(generation, progress) + return generation, progress def _schedule_polling_progress_verifier( self, generation: int, progress: asyncio.Event @@ -2333,6 +2353,39 @@ class TelegramAdapter(BasePlatformAdapter): return False if not (self._app and self._app.updater): raise RuntimeError("Telegram application/updater not initialized") + + # Strict cold start (#67498): background recovery must not run while + # the readiness gate is waiting. A G1 polling error would otherwise + # schedule _handle_polling_network_error(), which starts generation + # G2 on the same partial application while this coroutine still waits + # on G1's event — the cold connect then either times out on G1 despite + # G2 succeeding, or G2 "heals" the partial app so GatewayRunner never + # disposes it and retries fresh. Instead, capture the first polling + # error and fail the cold attempt immediately; GatewayRunner owns + # disposal and retry with a fresh adapter. + strict_error: list[BaseException] = [] + strict_error_event = asyncio.Event() + strict_gate_open = True + effective_callback = error_callback + if require_progress: + loop = asyncio.get_running_loop() + + def _strict_error_callback(error: Exception) -> None: + # PTB registers this callback for the whole polling + # generation. After the readiness gate closes (success), + # delegate to the real callback so ongoing polling errors + # keep flowing into background recovery. + if not strict_gate_open: + if error_callback is not None: + error_callback(error) + return + if not strict_error: + strict_error.append(error) + # PTB invokes error callbacks from the polling task; the + # event must be set on the loop to wake the strict waiter. + loop.call_soon_threadsafe(strict_error_event.set) + + effective_callback = _strict_error_callback try: # Same watchdog bound as the reconnect ladders: a wedged httpx # connection pool can hang start_polling() forever at bootstrap @@ -2340,26 +2393,55 @@ class TelegramAdapter(BasePlatformAdapter): # TimeoutError (OSError subclass), so the except below classifies # it via _looks_like_network_error and schedules background # recovery instead of blocking connect() indefinitely. - await self._start_polling_once( + generation, progress = await self._start_polling_once( self._app, drop_pending_updates=drop_pending_updates, - error_callback=error_callback, + error_callback=effective_callback, abandon_app_on_timeout=require_progress, + # The strict gate below IS the cold-start verifier; the + # background verifier would only race it on the partial app. + schedule_verifier=not require_progress, ) if require_progress: + # Bind to THIS generation's progress event (returned above), + # not self._polling_progress_event — a concurrent task could + # have replaced it with a later generation's event. + progress_wait = asyncio.ensure_future(progress.wait()) + error_wait = asyncio.ensure_future(strict_error_event.wait()) try: await _await_with_thread_deadline( - self._polling_progress_event.wait(), + _first_completed(progress_wait, error_wait), timeout=_INITIAL_POLLING_PROGRESS_TIMEOUT, ) except asyncio.TimeoutError as exc: raise OSError( - "Telegram getUpdates made no progress during initial connect" + "Telegram getUpdates made no progress within " + f"{_INITIAL_POLLING_PROGRESS_TIMEOUT:.0f}s during initial " + "connect — failing startup so the gateway retries with a " + "fresh adapter (#67498)" ) from exc - if not self._polling_progress_event.is_set(): + finally: + for fut in (progress_wait, error_wait): + if not fut.done(): + fut.cancel() + await asyncio.gather( + progress_wait, error_wait, return_exceptions=True + ) + if strict_error and not progress.is_set(): + raise OSError( + "Telegram polling errored before first getUpdates " + "success during initial connect: " + f"{_redact_telegram_error_text(strict_error[0])}" + ) from strict_error[0] + if not progress.is_set(): raise OSError( "Telegram getUpdates did not become ready during initial connect" ) + # Readiness proven — close the strict gate so any later + # polling error flows to the real background-recovery + # callback instead of the (now finished) cold-start gate. + strict_gate_open = False + self._polling_error_callback_ref = error_callback return True except _PollingLifecycleAbort: return False diff --git a/tests/gateway/test_telegram_start_polling_timeout.py b/tests/gateway/test_telegram_start_polling_timeout.py index 27e3fc712b01..b7e3f7bd69a7 100644 --- a/tests/gateway/test_telegram_start_polling_timeout.py +++ b/tests/gateway/test_telegram_start_polling_timeout.py @@ -170,3 +170,110 @@ async def test_initial_connect_requires_get_updates_progress(monkeypatch): error_callback=None, require_progress=True, ) + + +@pytest.mark.asyncio +async def test_initial_connect_succeeds_on_current_generation_progress(monkeypatch): + """Strict cold start returns True once THIS generation records progress.""" + monkeypatch.setattr(tg_adapter, "_INITIAL_POLLING_PROGRESS_TIMEOUT", 5.0) + a = _bare_adapter() + app = MagicMock() + app.updater = AsyncMock() + + async def start_polling_with_progress(**_kwargs): + # PTB's instrumented getUpdates request records progress for the + # generation started by this call. + a._record_polling_progress(a._polling_generation) + + app.updater.start_polling = AsyncMock(side_effect=start_polling_with_progress) + a._app = app + + ok = await asyncio.wait_for( + a._start_polling_resilient( + drop_pending_updates=True, + error_callback=None, + require_progress=True, + ), + timeout=10, + ) + assert ok is True + assert a._send_path_degraded is False + # Strict mode owns readiness itself; the background verifier must not be + # racing it on the cold-start application. + assert getattr(a, "_polling_progress_verifier_task", None) is None + + +@pytest.mark.asyncio +async def test_initial_connect_polling_error_fails_fast_not_background(monkeypatch): + """#67498 idle-threads shape: a polling error during strict cold start + must surface immediately as a connect failure — NOT be swallowed into a + background recovery task that restarts polling on the partial app while + the readiness gate waits out its full deadline (the G1/G2 race from the + #69240 review).""" + monkeypatch.setattr(tg_adapter, "_INITIAL_POLLING_PROGRESS_TIMEOUT", 30.0) + a = _bare_adapter() + app = MagicMock() + app.updater = AsyncMock() + + captured_callbacks = [] + + async def start_polling_capture(**kwargs): + captured_callbacks.append(kwargs.get("error_callback")) + + app.updater.start_polling = AsyncMock(side_effect=start_polling_capture) + a._app = app + + recovery_scheduled = [] + a._schedule_polling_recovery = lambda err, reason: recovery_scheduled.append(err) + + async def run_connect(): + return await a._start_polling_resilient( + drop_pending_updates=True, + error_callback=lambda e: recovery_scheduled.append(e), + require_progress=True, + ) + + task = asyncio.ensure_future(run_connect()) + # Let start_polling run and the strict gate begin waiting. + for _ in range(10): + await asyncio.sleep(0) + if captured_callbacks: + break + assert captured_callbacks and captured_callbacks[0] is not None + + # PTB reports a network error on the first getUpdates (G1 error). + captured_callbacks[0](OSError("getUpdates failed")) + + # The cold attempt must fail promptly (well before the 30s readiness + # deadline) with a loud OSError, and must NOT have scheduled background + # recovery (which would start G2 on the same partial application). + with pytest.raises(OSError, match="errored before first getUpdates"): + await asyncio.wait_for(task, timeout=5) + assert recovery_scheduled == [] + + +@pytest.mark.asyncio +async def test_initial_connect_ignores_stale_generation_progress(monkeypatch): + """Progress recorded for a stale generation must not satisfy readiness.""" + monkeypatch.setattr(tg_adapter, "_INITIAL_POLLING_PROGRESS_TIMEOUT", 0.1) + a = _bare_adapter() + app = MagicMock() + app.updater = AsyncMock() + + async def start_polling_stale_progress(**_kwargs): + # Progress arrives tagged with a PREVIOUS generation (e.g. a late + # response from an abandoned attempt) — must be rejected. + a._record_polling_progress(a._polling_generation - 1) + + app.updater.start_polling = AsyncMock(side_effect=start_polling_stale_progress) + a._app = app + + with pytest.raises(OSError, match="getUpdates made no progress"): + await asyncio.wait_for( + a._start_polling_resilient( + drop_pending_updates=True, + error_callback=None, + require_progress=True, + ), + timeout=10, + )