diff --git a/plugins/platforms/telegram/adapter.py b/plugins/platforms/telegram/adapter.py index e617b7b0bd23..3a986283239c 100644 --- a/plugins/platforms/telegram/adapter.py +++ b/plugins/platforms/telegram/adapter.py @@ -17,6 +17,7 @@ import os import html as _html import re import threading +from contextvars import ContextVar from datetime import datetime, timezone from typing import Dict, List, Optional, Set, Any @@ -474,6 +475,16 @@ _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 +# A generation is not healthy until the dedicated getUpdates request returns +# successfully. This exceeds a normal long-poll cycle for healthy idle bots. +_POLLING_PROGRESS_TIMEOUT = 60.0 +_POLLING_GENERATION_CONTEXT: ContextVar[Optional[int]] = ContextVar( + "telegram_polling_generation", default=None +) + + +class _PollingLifecycleAbort(RuntimeError): + """Internal control flow for polling startup fenced by teardown.""" class TelegramAdapter(BasePlatformAdapter): @@ -637,6 +648,7 @@ class TelegramAdapter(BasePlatformAdapter): self._polling_progress_event = asyncio.Event() self._polling_progress_accepting: bool = False self._polling_progress_verifier_task: Optional[asyncio.Task] = None + self._polling_teardown_started: bool = False self._polling_error_callback_ref = None self._polling_heartbeat_task: Optional[asyncio.Task] = None # Consecutive heartbeat probes that saw queued updates the running @@ -652,10 +664,9 @@ class TelegramAdapter(BasePlatformAdapter): # error_callback ever fires and the gateway silently stops receiving # messages with the process still alive (#55769). self._polling_not_running_count: int = 0 - # After sustained reconnect storms the PTB httpx pool can return - # SendResult(success=True) for sends that never actually transmit. - # _handle_polling_network_error sets this; _verify_polling_after_reconnect - # clears it once getMe() confirms the Bot client is healthy. + # A polling generation stays degraded until the dedicated getUpdates + # request makes successful progress. start_polling() return and getMe() + # success on the general request path are not polling-health signals. # While True, send() short-circuits to a failure so callers # (cron live-adapter branch) fall through to standalone delivery. self._send_path_degraded: bool = False @@ -1950,11 +1961,20 @@ class TelegramAdapter(BasePlatformAdapter): def _begin_polling_generation(self) -> tuple[int, asyncio.Event]: """Start accepting progress for a new getUpdates polling generation.""" - verifier = self._polling_progress_verifier_task + if getattr(self, "_polling_teardown_started", False): + self._polling_progress_accepting = False + self._send_path_degraded = True + progress = getattr(self, "_polling_progress_event", None) + if progress is None: + progress = asyncio.Event() + self._polling_progress_event = progress + return getattr(self, "_polling_generation", 0), progress + + verifier = getattr(self, "_polling_progress_verifier_task", None) if verifier is not None and not verifier.done(): verifier.cancel() self._polling_progress_verifier_task = None - self._polling_generation += 1 + self._polling_generation = getattr(self, "_polling_generation", 0) + 1 self._polling_progress_event = asyncio.Event() self._polling_progress_accepting = True self._send_path_degraded = True @@ -1962,12 +1982,15 @@ class TelegramAdapter(BasePlatformAdapter): def _record_polling_progress(self, generation: int) -> None: """Record successful getUpdates I/O for the current generation only.""" + if getattr(self, "_polling_teardown_started", False): + return if not self._polling_progress_accepting: return if generation != self._polling_generation: return self._polling_progress_event.set() self._polling_network_error_count = 0 + self._polling_conflict_count = 0 self._send_path_degraded = False def _instrument_polling_request(self, request): @@ -1975,16 +1998,100 @@ class TelegramAdapter(BasePlatformAdapter): do_request = request.do_request async def _do_request(*args, **kwargs): - generation = self._polling_generation + generation = _POLLING_GENERATION_CONTEXT.get() result = await do_request(*args, **kwargs) - status_code, _ = result - if 200 <= status_code < 300: - self._record_polling_progress(generation) + status_code, payload = result + if generation is not None and 200 <= status_code < 300: + try: + # Use the request's own parser so health observation agrees + # exactly with PTB's authoritative response handling (e.g. + # UTF-8 replacement decoding and BOM rejection). + envelope = request.parse_json_payload(payload) + except Exception: + # Instrumentation is observational: PTB still parses the + # untouched payload and owns the resulting exception. + pass + else: + if ( + isinstance(envelope, dict) + and envelope.get("ok") is True + and "result" in envelope + ): + self._record_polling_progress(generation) return result request.do_request = _do_request return request + async def _start_polling_once( + self, + app, + *, + drop_pending_updates: bool, + error_callback, + ) -> None: + """Start one generation and verify real getUpdates progress.""" + if getattr(self, "_polling_teardown_started", False): + raise _PollingLifecycleAbort("Telegram polling teardown started") + generation, progress = self._begin_polling_generation() + if not self._polling_progress_accepting: + raise _PollingLifecycleAbort("Telegram polling teardown started") + + def _generation_error_callback(error: Exception) -> None: + if getattr(self, "_polling_teardown_started", False): + return + if generation != self._polling_generation: + return + if error_callback is not None: + callback_context_token = _POLLING_GENERATION_CONTEXT.set(None) + try: + error_callback(error) + finally: + _POLLING_GENERATION_CONTEXT.reset(callback_context_token) + + context_token = _POLLING_GENERATION_CONTEXT.set(generation) + try: + await asyncio.wait_for( + app.updater.start_polling( + allowed_updates=Update.ALL_TYPES, + drop_pending_updates=drop_pending_updates, + error_callback=_generation_error_callback, + ), + timeout=_UPDATER_START_TIMEOUT, + ) + finally: + _POLLING_GENERATION_CONTEXT.reset(context_token) + if getattr(self, "_polling_teardown_started", False): + self._polling_progress_accepting = False + self._send_path_degraded = True + raise _PollingLifecycleAbort("Telegram polling teardown started") + self._schedule_polling_progress_verifier(generation, progress) + + def _schedule_polling_progress_verifier( + self, generation: int, progress: asyncio.Event + ) -> None: + """Own exactly one tracked verifier for the current generation.""" + if getattr(self, "_polling_teardown_started", False): + self._polling_progress_accepting = False + self._send_path_degraded = True + return + previous = getattr(self, "_polling_progress_verifier_task", None) + if previous is not None and not previous.done(): + previous.cancel() + + task = asyncio.get_running_loop().create_task( + self._verify_polling_after_reconnect(generation, progress) + ) + self._polling_progress_verifier_task = task + self._background_tasks.add(task) + + def _clear_finished_verifier(finished: asyncio.Task) -> None: + self._background_tasks.discard(finished) + if self._polling_progress_verifier_task is finished: + self._polling_progress_verifier_task = None + + task.add_done_callback(_clear_finished_verifier) + def _get_general_request_drain_lock(self) -> asyncio.Lock: lock = getattr(self, "_general_request_drain_lock", None) if lock is None: @@ -2037,6 +2144,8 @@ class TelegramAdapter(BasePlatformAdapter): adapter: the gateway process stays alive and the existing reconnect ladder (``_handle_polling_network_error``) recovers in the background. """ + if getattr(self, "_polling_teardown_started", False): + return if self.has_fatal_error: return if self._polling_error_task and not self._polling_error_task.done(): @@ -2088,6 +2197,8 @@ class TelegramAdapter(BasePlatformAdapter): network error was scheduled for background recovery instead of raising (keeping the gateway process alive). """ + if getattr(self, "_polling_teardown_started", False): + return False if not (self._app and self._app.updater): raise RuntimeError("Telegram application/updater not initialized") try: @@ -2097,16 +2208,17 @@ 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 asyncio.wait_for( - self._app.updater.start_polling( - allowed_updates=Update.ALL_TYPES, - drop_pending_updates=drop_pending_updates, - error_callback=error_callback, - ), - timeout=_UPDATER_START_TIMEOUT, + await self._start_polling_once( + self._app, + drop_pending_updates=drop_pending_updates, + error_callback=error_callback, ) return True + except _PollingLifecycleAbort: + return False except Exception as err: + if getattr(self, "_polling_teardown_started", False): + return False if self._looks_like_polling_conflict(err): logger.warning( "[%s] Telegram polling bootstrap conflict; gateway stays alive " @@ -2135,6 +2247,8 @@ class TelegramAdapter(BasePlatformAdapter): MAX_NETWORK_RETRIES attempts, then mark the adapter retryable-fatal so the supervisor restarts the gateway process. """ + if getattr(self, "_polling_teardown_started", False): + return if self.has_fatal_error: return @@ -2164,6 +2278,9 @@ class TelegramAdapter(BasePlatformAdapter): ) await asyncio.sleep(delay) + if getattr(self, "_polling_teardown_started", False): + return + # Capture a stable local reference: self._app can be reassigned to None # by a concurrent disconnect() while we're suspended across the awaits # below, and re-reading self._app after that point would silently swap @@ -2194,64 +2311,39 @@ class TelegramAdapter(BasePlatformAdapter): except Exception: pass + if getattr(self, "_polling_teardown_started", False): + return await self._drain_polling_connections() + if getattr(self, "_polling_teardown_started", False): + return + try: if not app: raise RuntimeError("Telegram application was torn down during reconnect") - # Guard start_polling() with a timeout: when the connection pool is - # in a degraded state (e.g., after _drain_polling_connections()), the - # httpx client may hold a stale socket that neither connects nor times - # out within PTB's internal flow. Bounding start_polling() prevents - # the reconnect ladder from stalling indefinitely and allows the - # heartbeat loop to trigger its own recovery path. - # Refs: NousResearch/hermes-agent#59614 - try: - await asyncio.wait_for( - app.updater.start_polling( - allowed_updates=Update.ALL_TYPES, - drop_pending_updates=False, - error_callback=self._polling_error_callback_ref, - ), - timeout=_UPDATER_START_TIMEOUT, - ) - except asyncio.TimeoutError: - raise RuntimeError( - "start_polling() timed out — connection pool may be wedged" - ) + await self._start_polling_once( + app, + drop_pending_updates=False, + error_callback=self._polling_error_callback_ref, + ) logger.info( - "[%s] Telegram polling resumed after network error (attempt %d)", + "[%s] Telegram polling restarted after network error (attempt %d); " + "health pending getUpdates progress", self.name, attempt, ) - self._polling_network_error_count = 0 - # start_polling() succeeding IS the recovery signal: the long-poll - # connection is live again, so clear the degraded flag immediately - # rather than blocking all outbound sends for the full - # HEARTBEAT_PROBE_DELAY window. The deferred probe below is a - # defensive re-check — if it later detects a silent wedge (PTB - # running=True but consumer task dead) it re-enters the ladder, - # which re-sets _send_path_degraded. Without this clear here, a - # clean reconnect leaves the flag stuck True until the 60s probe - # (or forever, if the probe is never scheduled), blocking the send - # path even though the bot has fully recovered. See #35205. - self._send_path_degraded = False - # start_polling() returning is necessary but not sufficient: - # PTB's Updater can be left in a state where `running` is True - # but the underlying long-poll task is wedged on a stale httpx - # connection and never makes progress. No error_callback fires - # in that state, so the reconnect ladder won't advance on its - # own. Schedule a deferred probe to detect the wedge and - # re-enter the ladder if needed. - if not self.has_fatal_error: - probe = asyncio.ensure_future(self._verify_polling_after_reconnect()) - self._background_tasks.add(probe) - probe.add_done_callback(self._background_tasks.discard) + except _PollingLifecycleAbort: + return except Exception as retry_err: + if getattr(self, "_polling_teardown_started", False): + return safe_retry_error = _redact_telegram_error_text(retry_err) logger.warning("[%s] Telegram polling reconnect failed: %s", self.name, safe_retry_error) # start_polling failed — polling is dead and no further error # callbacks will fire, so schedule the next retry ourselves. - if not self.has_fatal_error: + if ( + not self.has_fatal_error + and not getattr(self, "_polling_teardown_started", False) + ): task = asyncio.ensure_future( self._handle_polling_network_error(retry_err) ) @@ -2281,9 +2373,9 @@ class TelegramAdapter(BasePlatformAdapter): ``_handle_polling_network_error`` — the same path triggered by PTB's own ``error_callback`` — which drains the dead pool and restarts polling. - Unlike ``_verify_polling_after_reconnect`` (a one-shot probe scheduled - only after an explicit reconnect), this loop runs for the full lifetime - of the polling connection, so it catches a socket that wedges during + Unlike the generation verifier (a one-shot progress deadline after + every polling start), this loop runs for the full lifetime of the + polling connection, so it catches a socket that wedges later during steady-state operation without any prior error event. """ HEARTBEAT_INTERVAL = 90 # seconds between probes @@ -2292,6 +2384,8 @@ class TelegramAdapter(BasePlatformAdapter): while True: try: await asyncio.sleep(HEARTBEAT_INTERVAL) + if getattr(self, "_polling_teardown_started", False): + return if self.has_fatal_error: return bot = self._app.bot if self._app else None @@ -2348,6 +2442,8 @@ class TelegramAdapter(BasePlatformAdapter): report a queue against a live consumer. We detect the stopped updater directly and feed the same ladder (#55769). """ + if getattr(self, "_polling_teardown_started", False): + return # Only meaningful in polling mode; in webhook mode Telegram pushes # updates and holds no server-side queue. if self._webhook_mode: @@ -2381,6 +2477,8 @@ class TelegramAdapter(BasePlatformAdapter): ) if self._polling_not_running_count >= 2: self._polling_not_running_count = 0 + if getattr(self, "_polling_teardown_started", False): + return logger.warning( "[%s] Telegram updater is not running (long-poll task " "gone); triggering polling restart", @@ -2415,6 +2513,8 @@ class TelegramAdapter(BasePlatformAdapter): ) if self._polling_pending_stuck_count >= 2: self._polling_pending_stuck_count = 0 + if getattr(self, "_polling_teardown_started", False): + return logger.warning( "[%s] getUpdates consumer appears wedged (queue not draining); " "triggering polling restart", @@ -2427,66 +2527,96 @@ class TelegramAdapter(BasePlatformAdapter): ) ) - async def _verify_polling_after_reconnect(self) -> None: - """Heartbeat probe scheduled after a successful reconnect. + async def _verify_polling_after_reconnect( + self, + generation: Optional[int] = None, + progress: Optional[asyncio.Event] = None, + ) -> None: + """Require getUpdates progress, using getMe only to classify failure. - PTB's Updater can survive a botched stop()+start_polling() cycle - with `running=True` but a wedged consumer task. No error callback - fires, so the reconnect ladder doesn't advance on its own. This - probe detects the wedge by: - - 1. Sleeping HEARTBEAT_PROBE_DELAY so a healthy long-poll has time - to complete at least one cycle. - 2. Verifying `Updater.running` is still True. - 3. Probing the bot endpoint with a tight asyncio timeout. A - wedged httpx pool fails this probe; a healthy one returns - well under the timeout. - - On connectivity failure, re-enter the reconnect ladder (via - ``_schedule_polling_recovery`` so the in-flight guard and task - bookkeeping apply) and let the existing MAX_NETWORK_RETRIES path - ultimately escalate to fatal-error. Auth/validation failures - (``InvalidToken``, ``BadRequest``, ...) are not connectivity symptoms - and must not trigger reconnect churn — same policy as the heartbeat - loop and the pending-update probe (#62098, #63243). + The generation-bound event is set only by a successful response on the + dedicated getUpdates request. A general-path getMe success can classify + connectivity, but cannot heal polling health. Connectivity failures + enter the guarded recovery ladder; auth/validation errors do not churn. """ - HEARTBEAT_PROBE_DELAY = 60 PROBE_TIMEOUT = 10 - - await asyncio.sleep(HEARTBEAT_PROBE_DELAY) - - if self.has_fatal_error: + if getattr(self, "_polling_teardown_started", False): return - if not (self._app and self._app.updater and self._app.updater.running): + if generation is None: + generation = self._polling_generation + if progress is None: + progress = self._polling_progress_event + + try: + await asyncio.wait_for( + progress.wait(), timeout=_POLLING_PROGRESS_TIMEOUT + ) + except asyncio.TimeoutError: + pass + + if getattr(self, "_polling_teardown_started", False): + return + if progress.is_set() or self.has_fatal_error: + return + if not self._polling_progress_accepting: + return + if generation != self._polling_generation: + return + if progress is not self._polling_progress_event: + return + + app = self._app + if not (app and app.updater and app.updater.running): logger.warning( - "[%s] Updater not running %ds after reconnect — treating as wedged", - self.name, HEARTBEAT_PROBE_DELAY, + "[%s] Updater made no getUpdates progress and is not running", + self.name, ) self._schedule_polling_recovery( - RuntimeError("Updater not running after reconnect heartbeat"), - reason="post-reconnect probe: updater not running", + RuntimeError("Updater not running after polling progress deadline"), + reason="polling progress verifier: updater not running", ) return try: - await asyncio.wait_for(self._app.bot.get_me(), PROBE_TIMEOUT) - self._send_path_degraded = False + await asyncio.wait_for(app.bot.get_me(), PROBE_TIMEOUT) except Exception as probe_err: + if getattr(self, "_polling_teardown_started", False): + return + if self.has_fatal_error or not self._polling_progress_accepting: + return + if generation != self._polling_generation: + return + if progress is not self._polling_progress_event or progress.is_set(): + return if not self._looks_like_network_error(probe_err): logger.warning( - "[%s] Post-reconnect probe hit a non-connectivity error" + "[%s] Polling progress verifier hit a non-connectivity error" " (not retrying): %s", self.name, _redact_telegram_error_text(probe_err), ) return logger.warning( - "[%s] Polling heartbeat probe failed %ds after reconnect: %s", - self.name, HEARTBEAT_PROBE_DELAY, - _redact_telegram_error_text(probe_err), + "[%s] Polling progress verifier connectivity probe failed: %s", + self.name, _redact_telegram_error_text(probe_err), ) self._schedule_polling_recovery( - probe_err, reason="post-reconnect probe failure" + probe_err, + reason="polling progress verifier connectivity failure", ) + return + + if getattr(self, "_polling_teardown_started", False): + return + if self.has_fatal_error or not self._polling_progress_accepting: + return + if generation != self._polling_generation: + return + if progress is not self._polling_progress_event or progress.is_set(): + return + self._schedule_polling_recovery( + RuntimeError("getUpdates made no progress before verifier deadline"), + reason="polling progress verifier: general path healthy but getUpdates stalled", + ) def _disarm_ptb_retry_loop(self) -> None: """Synchronously stop PTB's internal polling retry loop. @@ -2552,6 +2682,8 @@ class TelegramAdapter(BasePlatformAdapter): ) async def _handle_polling_conflict(self, error: Exception) -> None: + if getattr(self, "_polling_teardown_started", False): + return if self.has_fatal_error and self.fatal_error_code == "telegram_polling_conflict": return # Transient 409 Conflict errors arise when the previous gateway process @@ -2607,7 +2739,11 @@ class TelegramAdapter(BasePlatformAdapter): pass await asyncio.sleep(RETRY_DELAY) + if getattr(self, "_polling_teardown_started", False): + return await self._drain_polling_connections() + if getattr(self, "_polling_teardown_started", False): + return # Capture a stable local reference: self._app can be reassigned to # None by a concurrent disconnect() while we're suspended across @@ -2619,31 +2755,22 @@ class TelegramAdapter(BasePlatformAdapter): try: if not app: raise RuntimeError("Telegram application was torn down during conflict reconnect") - # Same watchdog bound as the network-error ladder: an - # exhausted pool hangs start_polling() on the conflict path - # identically (#59614). Timeout converts to RuntimeError so - # the except below logs a readable message and schedules the - # next conflict attempt instead of wedging attempt N forever. - try: - await asyncio.wait_for( - app.updater.start_polling( - allowed_updates=Update.ALL_TYPES, - drop_pending_updates=False, - error_callback=self._polling_error_callback_ref, - ), - timeout=_UPDATER_START_TIMEOUT, - ) - except asyncio.TimeoutError: - raise RuntimeError( - "start_polling() timed out — connection pool may be wedged" - ) + await self._start_polling_once( + app, + drop_pending_updates=False, + error_callback=self._polling_error_callback_ref, + ) logger.info( - "[%s] Telegram polling resumed after conflict retry %d/%d", + "[%s] Telegram polling restarted after conflict retry %d/%d; " + "health pending getUpdates progress", self.name, self._polling_conflict_count, MAX_CONFLICT_RETRIES, ) - self._polling_conflict_count = 0 # reset counter on success + return + except _PollingLifecycleAbort: return except Exception as retry_err: + if getattr(self, "_polling_teardown_started", False): + return logger.warning( "[%s] Telegram polling retry %d/%d failed: %s. " "Scheduling next attempt.", @@ -2655,7 +2782,10 @@ class TelegramAdapter(BasePlatformAdapter): # a fatal error leaves the adapter in a limbo state: the # gateway process is alive and reports "connected" but # no messages are received or sent. - if self._polling_conflict_count < MAX_CONFLICT_RETRIES: + if ( + self._polling_conflict_count < MAX_CONFLICT_RETRIES + and not getattr(self, "_polling_teardown_started", False) + ): # We are inside a running coroutine, so the running loop is # guaranteed to exist. asyncio.get_event_loop() is deprecated # and raises "RuntimeError: There is no current event loop in @@ -2669,6 +2799,9 @@ class TelegramAdapter(BasePlatformAdapter): return # Fall through to fatal on the last retry. + if getattr(self, "_polling_teardown_started", False): + return + # Exhausted all retries — declare a fatal error so the gateway # runner can surface this clearly and the user knows to act. message = ( @@ -3125,6 +3258,14 @@ class TelegramAdapter(BasePlatformAdapter): TELEGRAM_WEBHOOK_PORT Local listen port (default 8443) TELEGRAM_WEBHOOK_SECRET Secret token for update verification """ + # Explicit connect() is the only operation allowed to reopen polling + # after a completed, serialized teardown. Background recovery never + # clears this fence. + self._polling_teardown_started = False + # Mode selection is re-evaluated on every explicit connection. Keep + # webhook state false unless this connection starts its webhook. + self._webhook_mode = False + if not TELEGRAM_AVAILABLE: logger.error( "[%s] python-telegram-bot not installed. Run: pip install python-telegram-bot", @@ -3420,6 +3561,8 @@ class TelegramAdapter(BasePlatformAdapter): drop_pending_updates=not is_reconnect, ) self._webhook_mode = True + self._polling_progress_accepting = False + self._send_path_degraded = False logger.info( "[%s] Webhook server listening on 0.0.0.0:%d%s", self.name, webhook_port, webhook_path, @@ -3436,6 +3579,8 @@ class TelegramAdapter(BasePlatformAdapter): loop = asyncio.get_running_loop() def _polling_error_callback(error: Exception) -> None: + if getattr(self, "_polling_teardown_started", False): + return if self._polling_error_task and not self._polling_error_task.done(): return if self._looks_like_polling_conflict(error): @@ -3569,7 +3714,8 @@ class TelegramAdapter(BasePlatformAdapter): collect(task) for task in list(self._pending_text_batch_tasks.values()): collect(task) - collect(self._polling_error_task) + collect(getattr(self, "_polling_error_task", None)) + collect(getattr(self, "_polling_progress_verifier_task", None)) for task in pending_tasks: task.cancel() @@ -3582,8 +3728,10 @@ class TelegramAdapter(BasePlatformAdapter): self._pending_photo_batches.clear() self._pending_text_batch_tasks.clear() self._pending_text_batches.clear() - if self._polling_error_task is not current_task: + if getattr(self, "_polling_error_task", None) is not current_task: self._polling_error_task = None + if getattr(self, "_polling_progress_verifier_task", None) is not current_task: + self._polling_progress_verifier_task = None async def disconnect(self) -> None: """Stop polling/webhook, cancel pending delayed deliveries, and disconnect.""" @@ -3591,6 +3739,42 @@ class TelegramAdapter(BasePlatformAdapter): # that wins the race against teardown and prevents new delayed tasks # from being scheduled by late update handlers. self._mark_disconnected() + self._polling_teardown_started = True + self._polling_progress_accepting = False + self._polling_generation = getattr(self, "_polling_generation", 0) + 1 + self._polling_progress_event = asyncio.Event() + self._send_path_degraded = True + + # Recovery can be suspended in stop/drain/start while disconnect begins. + # Cancel and await both polling lifecycle owners immediately after the + # fence, before any other teardown await lets them start a new generation. + current_task = asyncio.current_task() + lifecycle_tasks: list[asyncio.Task] = [] + lifecycle_seen: set[int] = set() + for task in ( + getattr(self, "_polling_error_task", None), + getattr(self, "_polling_progress_verifier_task", None), + ): + if not task or task.done() or task is current_task: + continue + marker = id(task) + if marker in lifecycle_seen: + continue + lifecycle_seen.add(marker) + task.cancel() + if asyncio.isfuture(task) or asyncio.iscoroutine(task): + lifecycle_tasks.append(task) + if lifecycle_tasks: + await asyncio.gather(*lifecycle_tasks, return_exceptions=True) + if getattr(self, "_polling_error_task", None) is not current_task: + self._polling_error_task = None + if getattr(self, "_polling_progress_verifier_task", None) is not current_task: + self._polling_progress_verifier_task = None + + # Cancellation callbacks may have run while awaited; the teardown fence + # remains authoritative regardless of their finalizers. + self._polling_progress_accepting = False + self._send_path_degraded = True # Cancel deferred post-connect housekeeping (command-menu / DM-topic / # status-indicator Bot API calls) so it cannot fire into a half-torn-down @@ -3604,10 +3788,11 @@ class TelegramAdapter(BasePlatformAdapter): # Cancel the heartbeat before tearing down the app so the probe task # cannot fire get_me() into a half-shutdown bot client. - if self._polling_heartbeat_task and not self._polling_heartbeat_task.done(): - self._polling_heartbeat_task.cancel() + polling_heartbeat_task = getattr(self, "_polling_heartbeat_task", None) + if polling_heartbeat_task and not polling_heartbeat_task.done(): + polling_heartbeat_task.cancel() try: - await self._polling_heartbeat_task + await polling_heartbeat_task except asyncio.CancelledError: pass self._polling_heartbeat_task = None diff --git a/tests/gateway/test_telegram_conflict.py b/tests/gateway/test_telegram_conflict.py index 5868a0c56ff5..e00a0f1d33c6 100644 --- a/tests/gateway/test_telegram_conflict.py +++ b/tests/gateway/test_telegram_conflict.py @@ -137,14 +137,13 @@ async def test_polling_conflict_retries_before_fatal(monkeypatch): # First conflict: should retry, NOT be fatal captured["error_callback"](conflict("Conflict: terminated by other getUpdates request")) - await asyncio.sleep(0) - await asyncio.sleep(0) - # Give the scheduled task a chance to run - for _ in range(10): - await asyncio.sleep(0) + await adapter._polling_error_task assert adapter.has_fatal_error is False, "First conflict should not be fatal" - assert adapter._polling_conflict_count == 0, "Count should reset after successful retry" + assert adapter._polling_conflict_count == 1, ( + "Count must remain until the retried generation makes getUpdates progress" + ) + assert adapter._send_path_degraded is True # connect() now starts a lifetime _polling_heartbeat_loop task. With # asyncio.sleep mocked to instant above, it must not be left running or it @@ -152,6 +151,54 @@ async def test_polling_conflict_retries_before_fatal(monkeypatch): await _cancel_heartbeat(adapter) +@pytest.mark.asyncio +async def test_current_generation_conflicts_accumulate_after_start_returns(monkeypatch): + """A later async 409 must advance the retry ladder after PTB start returns.""" + adapter = TelegramAdapter(PlatformConfig(enabled=True, token="***")) + callbacks = [] + conflict_tasks = [] + + async def capture_start(**kwargs): + callbacks.append(kwargs["error_callback"]) + + updater = SimpleNamespace( + start_polling=AsyncMock(side_effect=capture_start), + stop=AsyncMock(), + running=False, + ) + app = SimpleNamespace(updater=updater) + adapter._app = app + adapter._drain_polling_connections = AsyncMock() + monkeypatch.setattr("asyncio.sleep", AsyncMock()) + + def dispatch_conflict(error): + conflict_tasks.append( + asyncio.create_task(adapter._handle_polling_conflict(error)) + ) + + adapter._polling_error_callback_ref = dispatch_conflict + await adapter._start_polling_once( + app, + drop_pending_updates=False, + error_callback=dispatch_conflict, + ) + conflict = type("Conflict", (Exception,), {}) + + try: + callbacks[0](conflict("first async conflict")) + await conflict_tasks[-1] + assert adapter._polling_conflict_count == 1 + + callbacks[1](conflict("second async conflict")) + await conflict_tasks[-1] + assert adapter._polling_conflict_count == 2 + finally: + verifier = adapter._polling_progress_verifier_task + if verifier is not None and not verifier.done(): + verifier.cancel() + await asyncio.gather(verifier, return_exceptions=True) + + @pytest.mark.asyncio async def test_polling_conflict_becomes_fatal_after_retries(monkeypatch): """After exhausting retries, the conflict should become fatal.""" diff --git a/tests/gateway/test_telegram_network_reconnect.py b/tests/gateway/test_telegram_network_reconnect.py index c7e274d74ebb..c1c10726755d 100644 --- a/tests/gateway/test_telegram_network_reconnect.py +++ b/tests/gateway/test_telegram_network_reconnect.py @@ -35,6 +35,7 @@ def _ensure_telegram_mock(): _ensure_telegram_mock() +from plugins.platforms.telegram import adapter as tg_adapter # noqa: E402 from plugins.platforms.telegram.adapter import TelegramAdapter # noqa: E402 @@ -50,6 +51,13 @@ def _make_adapter() -> TelegramAdapter: return TelegramAdapter(PlatformConfig(enabled=True, token="test-token")) +async def _complete_current_polling_generation(adapter: TelegramAdapter) -> None: + verifier = adapter._polling_progress_verifier_task + adapter._record_polling_progress(adapter._polling_generation) + if verifier is not None: + await verifier + + @pytest.mark.asyncio async def test_reconnect_self_schedules_on_start_polling_failure(): """ @@ -157,9 +165,9 @@ async def test_reconnect_chained_retry_updates_polling_error_task(): @pytest.mark.asyncio -async def test_reconnect_success_resets_error_count(): +async def test_reconnect_success_waits_for_progress_to_reset_error_count(): """ - When start_polling() succeeds, _polling_network_error_count should reset to 0. + start_polling() return alone cannot reset the network-error count. """ adapter = _make_adapter() adapter._polling_network_error_count = 3 @@ -177,7 +185,12 @@ async def test_reconnect_success_resets_error_count(): with patch("asyncio.sleep", new_callable=AsyncMock): await adapter._handle_polling_network_error(Exception("Bad Gateway")) + assert adapter._polling_network_error_count == 4 + assert adapter._send_path_degraded is True + + await _complete_current_polling_generation(adapter) assert adapter._polling_network_error_count == 0 + assert adapter._send_path_degraded is False # Clean up the heartbeat-probe task scheduled after a successful reconnect. pending = [t for t in adapter._background_tasks if not t.done()] @@ -263,6 +276,8 @@ async def test_reconnect_drains_polling_request_only(): # Reconnect must still succeed mock_app.updater.start_polling.assert_called_once() + assert adapter._polling_network_error_count == 2 + await _complete_current_polling_generation(adapter) assert adapter._polling_network_error_count == 0 @@ -283,6 +298,8 @@ async def test_reconnect_continues_if_drain_fails(): # start_polling must still be called despite drain failure mock_app.updater.start_polling.assert_called_once() + assert adapter._polling_network_error_count == 2 + await _complete_current_polling_generation(adapter) assert adapter._polling_network_error_count == 0 @@ -339,10 +356,9 @@ async def test_drain_helper_noop_without_app(): @pytest.mark.asyncio -async def test_heartbeat_probe_no_op_when_polling_healthy(): +async def test_polling_verifier_exits_on_matching_progress(monkeypatch): """ - Probe scheduled after a successful reconnect: Updater.running=True and - bot.get_me() returns quickly → recovery confirmed, no further action. + Matching getUpdates progress exits without probing the general path. """ adapter = _make_adapter() @@ -355,19 +371,20 @@ async def test_heartbeat_probe_no_op_when_polling_healthy(): adapter._app = mock_app adapter._handle_polling_network_error = AsyncMock() + generation, progress = adapter._begin_polling_generation() + adapter._record_polling_progress(generation) + monkeypatch.setattr(tg_adapter, "_POLLING_PROGRESS_TIMEOUT", 0) - with patch("asyncio.sleep", new_callable=AsyncMock): - await adapter._verify_polling_after_reconnect() + await adapter._verify_polling_after_reconnect(generation, progress) - mock_app.bot.get_me.assert_awaited_once() + mock_app.bot.get_me.assert_not_awaited() adapter._handle_polling_network_error.assert_not_awaited() @pytest.mark.asyncio -async def test_heartbeat_probe_reenters_ladder_when_updater_not_running(): +async def test_heartbeat_probe_reenters_ladder_when_updater_not_running(monkeypatch): """ - If Updater.running has flipped to False by the heartbeat delay, treat - as wedged: re-enter the reconnect ladder. + If Updater.running is False at the progress deadline, re-enter recovery. """ adapter = _make_adapter() @@ -380,9 +397,10 @@ async def test_heartbeat_probe_reenters_ladder_when_updater_not_running(): adapter._app = mock_app adapter._handle_polling_network_error = AsyncMock() + generation, progress = adapter._begin_polling_generation() + monkeypatch.setattr(tg_adapter, "_POLLING_PROGRESS_TIMEOUT", 0) - with patch("asyncio.sleep", new_callable=AsyncMock): - await adapter._verify_polling_after_reconnect() + await adapter._verify_polling_after_reconnect(generation, progress) mock_app.bot.get_me.assert_not_called() # Recovery is scheduled through _schedule_polling_recovery (#63243), so @@ -397,7 +415,7 @@ async def test_heartbeat_probe_reenters_ladder_when_updater_not_running(): @pytest.mark.asyncio -async def test_heartbeat_probe_reenters_ladder_when_get_me_times_out(): +async def test_heartbeat_probe_reenters_ladder_when_get_me_times_out(monkeypatch): """ If bot.get_me() hangs longer than PROBE_TIMEOUT, treat as wedged. Simulates the connection-pool wedge that motivated this fix. @@ -416,15 +434,16 @@ async def test_heartbeat_probe_reenters_ladder_when_get_me_times_out(): adapter._app = mock_app adapter._handle_polling_network_error = AsyncMock() + generation, progress = adapter._begin_polling_generation() + monkeypatch.setattr(tg_adapter, "_POLLING_PROGRESS_TIMEOUT", 0) async def fast_wait_for(coro, timeout): if asyncio.iscoroutine(coro): coro.close() raise asyncio.TimeoutError() - with patch("asyncio.sleep", new_callable=AsyncMock): - with patch("plugins.platforms.telegram.adapter.asyncio.wait_for", new=fast_wait_for): - await adapter._verify_polling_after_reconnect() + with patch("plugins.platforms.telegram.adapter.asyncio.wait_for", new=fast_wait_for): + await adapter._verify_polling_after_reconnect(generation, progress) task = adapter._polling_error_task assert task is not None @@ -433,7 +452,7 @@ async def test_heartbeat_probe_reenters_ladder_when_get_me_times_out(): @pytest.mark.asyncio -async def test_heartbeat_probe_reenters_ladder_on_get_me_network_error(): +async def test_heartbeat_probe_reenters_ladder_on_get_me_network_error(monkeypatch): """ Any exception raised by bot.get_me() (NetworkError, ConnectionError, etc.) should re-enter the reconnect ladder with the original exception. @@ -449,9 +468,10 @@ async def test_heartbeat_probe_reenters_ladder_on_get_me_network_error(): adapter._app = mock_app adapter._handle_polling_network_error = AsyncMock() + generation, progress = adapter._begin_polling_generation() + monkeypatch.setattr(tg_adapter, "_POLLING_PROGRESS_TIMEOUT", 0) - with patch("asyncio.sleep", new_callable=AsyncMock): - await adapter._verify_polling_after_reconnect() + await adapter._verify_polling_after_reconnect(generation, progress) task = adapter._polling_error_task assert task is not None @@ -466,7 +486,7 @@ async def test_heartbeat_probe_reenters_ladder_on_get_me_network_error(): @pytest.mark.asyncio -async def test_heartbeat_probe_ignores_auth_errors(): +async def test_heartbeat_probe_ignores_auth_errors(monkeypatch): """ Auth/validation failures from the post-reconnect probe must not enter the network-reconnect ladder (#63243): a revoked token would otherwise churn @@ -487,16 +507,17 @@ async def test_heartbeat_probe_ignores_auth_errors(): adapter._app = mock_app adapter._handle_polling_network_error = AsyncMock() + generation, progress = adapter._begin_polling_generation() + monkeypatch.setattr(tg_adapter, "_POLLING_PROGRESS_TIMEOUT", 0) - with patch("asyncio.sleep", new_callable=AsyncMock): - await adapter._verify_polling_after_reconnect() + await adapter._verify_polling_after_reconnect(generation, progress) assert adapter._polling_error_task is None adapter._handle_polling_network_error.assert_not_awaited() @pytest.mark.asyncio -async def test_heartbeat_probe_defers_to_inflight_recovery(): +async def test_heartbeat_probe_defers_to_inflight_recovery(monkeypatch): """ A probe failure while another recovery is mid-flight must not start a second concurrent stop/drain/start_polling sequence (#63243) — overlapping @@ -517,16 +538,17 @@ async def test_heartbeat_probe_defers_to_inflight_recovery(): adapter._polling_error_task = inflight adapter._handle_polling_network_error = AsyncMock() + generation, progress = adapter._begin_polling_generation() + monkeypatch.setattr(tg_adapter, "_POLLING_PROGRESS_TIMEOUT", 0) - with patch("asyncio.sleep", new_callable=AsyncMock): - await adapter._verify_polling_after_reconnect() + await adapter._verify_polling_after_reconnect(generation, progress) assert adapter._polling_error_task is inflight adapter._handle_polling_network_error.assert_not_awaited() @pytest.mark.asyncio -async def test_heartbeat_probe_skips_when_already_fatal(): +async def test_heartbeat_probe_skips_when_already_fatal(monkeypatch): """ If the adapter is already in fatal-error state by the time the probe delay elapses, the probe should bail without further action. @@ -539,9 +561,10 @@ async def test_heartbeat_probe_skips_when_already_fatal(): adapter._app = mock_app adapter._handle_polling_network_error = AsyncMock() + generation, progress = adapter._begin_polling_generation() + monkeypatch.setattr(tg_adapter, "_POLLING_PROGRESS_TIMEOUT", 0) - with patch("asyncio.sleep", new_callable=AsyncMock): - await adapter._verify_polling_after_reconnect() + await adapter._verify_polling_after_reconnect(generation, progress) mock_app.bot.get_me.assert_not_called() adapter._handle_polling_network_error.assert_not_awaited() @@ -1060,4 +1083,3 @@ async def test_handle_polling_network_error_updater_stop_timeout(): # The reconnect ladder must have advanced past the hung stop(). assert drain_called, "_drain_polling_connections was not called after stop() timeout" assert start_polling_called, "start_polling was not called after stop() timeout" - diff --git a/tests/gateway/test_telegram_polling_progress.py b/tests/gateway/test_telegram_polling_progress.py index 717eaaf868a5..fdf00e17c99b 100644 --- a/tests/gateway/test_telegram_polling_progress.py +++ b/tests/gateway/test_telegram_polling_progress.py @@ -1,6 +1,8 @@ """Behavior contract for generation-safe Telegram polling progress.""" import asyncio +import json +from unittest.mock import AsyncMock, MagicMock import pytest @@ -14,6 +16,11 @@ class _ControlledRequest: instances = [] + @staticmethod + def parse_json_payload(payload): + """Match PTB's response authority used by the progress observer.""" + return json.loads(payload.decode("utf-8", "replace")) + def __init__(self, *args, result=None, error=None, entered=None, release=None, **kwargs): self.result = result self.error = error @@ -37,18 +44,183 @@ def _make_adapter() -> TelegramAdapter: return TelegramAdapter(PlatformConfig(enabled=True, token="test-token")) +def _mock_polling_app(*, get_me=None): + app = MagicMock() + app.updater = MagicMock() + app.updater.running = True + app.updater.stop = AsyncMock() + app.updater.start_polling = AsyncMock() + app.bot = MagicMock() + app.bot.get_me = get_me or AsyncMock(return_value=MagicMock()) + app.running = False + app.shutdown = AsyncMock() + return app + + +class _LifecycleBuilder: + def __init__(self, app): + self.app = app + self.polling_request = None + + def token(self, _token): + return self + + def request(self, _request): + return self + + def get_updates_request(self, request): + self.polling_request = request + return self + + def build(self): + return self.app + + +def _lifecycle_app(): + app = MagicMock() + app.updater = MagicMock() + app.updater.running = True + app.updater.start_polling = AsyncMock() + app.updater.start_webhook = AsyncMock() + app.updater.stop = AsyncMock() + app.bot = MagicMock() + app.bot.delete_webhook = AsyncMock() + app.initialize = AsyncMock() + app.start = AsyncMock() + app.stop = AsyncMock() + app.shutdown = AsyncMock() + app.running = True + return app + + +def _configure_lifecycle_connect(monkeypatch, adapter, apps): + builders = [_LifecycleBuilder(app) for app in apps] + remaining = iter(builders) + + class _Application: + @staticmethod + def builder(): + return next(remaining) + + async def _no_fallback_ips(): + return [] + + monkeypatch.setattr(tg_adapter, "Application", _Application) + monkeypatch.setattr(tg_adapter, "HTTPXRequest", _ControlledRequest) + monkeypatch.setattr(tg_adapter, "discover_fallback_ips", _no_fallback_ips) + monkeypatch.setattr(tg_adapter, "resolve_proxy_url", lambda *args, **kwargs: None) + monkeypatch.setattr(adapter, "_acquire_platform_lock", lambda *args, **kwargs: True) + monkeypatch.setattr(adapter, "_release_platform_lock", MagicMock()) + monkeypatch.setattr(adapter, "_fallback_ips", lambda: []) + monkeypatch.setattr(adapter, "_start_post_connect_housekeeping", MagicMock()) + return builders + + +async def _cancel_task(task): + if task is None or task.done(): + return + task.cancel() + await asyncio.gather(task, return_exceptions=True) + + +async def _request_for_generation(generation, request, *args): + """Run a direct request double under the production polling context.""" + generation_context = tg_adapter._POLLING_GENERATION_CONTEXT + token = generation_context.set(generation) + try: + return await request.do_request(*args) + finally: + generation_context.reset(token) + + +@pytest.mark.asyncio +async def test_polling_disconnect_webhook_reconnect_heals_webhook_send_path(monkeypatch): + adapter = _make_adapter() + polling_app = _lifecycle_app() + webhook_app = _lifecycle_app() + _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 + await adapter.disconnect() + + monkeypatch.setenv("TELEGRAM_WEBHOOK_URL", "https://example.test/telegram") + monkeypatch.setenv("TELEGRAM_WEBHOOK_SECRET", "test-secret") + try: + assert await adapter.connect(is_reconnect=True) is True + webhook_app.updater.start_webhook.assert_awaited_once() + assert adapter._webhook_mode is True + assert adapter._polling_progress_accepting is False + assert adapter._send_path_degraded is False + finally: + await adapter.disconnect() + + +@pytest.mark.asyncio +async def test_webhook_disconnect_polling_reconnect_resets_mode_and_waits_for_progress( + monkeypatch, +): + adapter = _make_adapter() + webhook_app = _lifecycle_app() + polling_app = _lifecycle_app() + builders = _configure_lifecycle_connect( + monkeypatch, adapter, [webhook_app, polling_app] + ) + heartbeat_started = asyncio.Event() + heartbeat_modes = [] + + async def heartbeat(): + heartbeat_modes.append(adapter._webhook_mode) + heartbeat_started.set() + await asyncio.Event().wait() + + monkeypatch.setattr(adapter, "_polling_heartbeat_loop", heartbeat) + monkeypatch.setenv("TELEGRAM_WEBHOOK_URL", "https://example.test/telegram") + monkeypatch.setenv("TELEGRAM_WEBHOOK_SECRET", "test-secret") + + assert await adapter.connect() is True + assert adapter._webhook_mode is True + assert adapter._polling_heartbeat_task is None + await adapter.disconnect() + + monkeypatch.delenv("TELEGRAM_WEBHOOK_URL") + monkeypatch.delenv("TELEGRAM_WEBHOOK_SECRET") + try: + assert await adapter.connect(is_reconnect=True) is True + assert adapter._webhook_mode is False + assert adapter._polling_heartbeat_task is not None + assert not adapter._polling_heartbeat_task.done() + await asyncio.wait_for(heartbeat_started.wait(), timeout=1) + assert heartbeat_modes == [False] + assert adapter._send_path_degraded is True + + generation = adapter._polling_generation + polling_request = builders[1].polling_request + polling_request.result = (200, b'{"ok":true,"result":[]}') + await _request_for_generation(generation, polling_request, "getUpdates") + await asyncio.wait_for(adapter._polling_progress_verifier_task, timeout=1) + assert adapter._send_path_degraded is False + finally: + await adapter.disconnect() + + @pytest.mark.asyncio async def test_current_polling_generation_success_records_progress(): adapter = _make_adapter() generation, progress = adapter._begin_polling_generation() adapter._polling_network_error_count = 3 - request = _ControlledRequest(result=(200, b'{"ok":true}')) + request = _ControlledRequest(result=(200, b'{"ok":true,"result":[]}')) instrumented = adapter._instrument_polling_request(request) - result = await instrumented.do_request("https://api.telegram.org/getUpdates") + result = await _request_for_generation( + generation, instrumented, "https://api.telegram.org/getUpdates" + ) assert instrumented is request - assert result == (200, b'{"ok":true}') + assert result == (200, b'{"ok":true,"result":[]}') assert progress.is_set() assert adapter._polling_network_error_count == 0 assert adapter._send_path_degraded is False @@ -59,14 +231,16 @@ async def test_current_polling_generation_success_records_progress(): @pytest.mark.parametrize("error_type", [RuntimeError, asyncio.CancelledError]) async def test_unsuccessful_polling_request_does_not_record_progress(error_type): adapter = _make_adapter() - _, progress = adapter._begin_polling_generation() + generation, progress = adapter._begin_polling_generation() adapter._polling_network_error_count = 3 request = adapter._instrument_polling_request( _ControlledRequest(error=error_type("request did not complete")) ) with pytest.raises(error_type): - await request.do_request("https://api.telegram.org/getUpdates") + await _request_for_generation( + generation, request, "https://api.telegram.org/getUpdates" + ) assert not progress.is_set() assert adapter._polling_network_error_count == 3 @@ -76,13 +250,15 @@ async def test_unsuccessful_polling_request_does_not_record_progress(error_type) @pytest.mark.asyncio async def test_http_error_response_does_not_record_polling_progress(): adapter = _make_adapter() - _, progress = adapter._begin_polling_generation() + generation, progress = adapter._begin_polling_generation() adapter._polling_network_error_count = 3 request = adapter._instrument_polling_request( _ControlledRequest(result=(500, b"bad")) ) - result = await request.do_request("https://api.telegram.org/getUpdates") + result = await _request_for_generation( + generation, request, "https://api.telegram.org/getUpdates" + ) assert result == (500, b"bad") assert not progress.is_set() @@ -155,17 +331,444 @@ async def test_late_previous_generation_completion_cannot_heal_current_generatio entered = asyncio.Event() release = asyncio.Event() request = adapter._instrument_polling_request( - _ControlledRequest(result=(200, b'{"ok":true}'), entered=entered, release=release) + _ControlledRequest( + result=(200, b'{"ok":true,"result":[]}'), + entered=entered, + release=release, + ) ) - completion = asyncio.create_task(request.do_request("getUpdates")) + completion = asyncio.create_task( + _request_for_generation(generation_1, request, "getUpdates") + ) await entered.wait() generation_2, progress_2 = adapter._begin_polling_generation() adapter._polling_network_error_count = 4 release.set() - assert await completion == (200, b'{"ok":true}') + assert await completion == (200, b'{"ok":true,"result":[]}') assert generation_2 == generation_1 + 1 assert not progress_2.is_set() assert adapter._polling_network_error_count == 4 assert adapter._send_path_degraded is True + + +@pytest.mark.asyncio +async def test_old_polling_child_keeps_generation_when_request_entry_is_delayed(): + adapter = _make_adapter() + adapter._app = _mock_polling_app() + release_old_request = asyncio.Event() + start_count = 0 + old_child = None + request = adapter._instrument_polling_request( + _ControlledRequest(result=(200, b'{"ok":true,"result":[]}')) + ) + + async def start_polling(**_kwargs): + nonlocal start_count, old_child + start_count += 1 + if start_count == 1: + + async def delayed_old_request(): + await release_old_request.wait() + return await request.do_request("getUpdates") + + old_child = asyncio.create_task(delayed_old_request()) + + adapter._app.updater.start_polling = start_polling + + await adapter._start_polling_once( + adapter._app, + drop_pending_updates=False, + error_callback=MagicMock(), + ) + generation_1 = adapter._polling_generation + await adapter._start_polling_once( + adapter._app, + drop_pending_updates=False, + error_callback=MagicMock(), + ) + generation_2 = adapter._polling_generation + progress_2 = adapter._polling_progress_event + verifier_2 = adapter._polling_progress_verifier_task + + try: + release_old_request.set() + assert await old_child == (200, b'{"ok":true,"result":[]}') + + assert generation_2 == generation_1 + 1 + assert not progress_2.is_set() + assert adapter._send_path_degraded is True + finally: + await _cancel_task(verifier_2) + + +@pytest.mark.asyncio +async def test_error_callback_is_bound_to_its_polling_generation(): + adapter = _make_adapter() + adapter._app = _mock_polling_app() + callbacks = [] + delegated = MagicMock() + + async def capture_start(**kwargs): + callbacks.append(kwargs["error_callback"]) + + adapter._app.updater.start_polling = capture_start + await adapter._start_polling_once( + adapter._app, + drop_pending_updates=False, + error_callback=delegated, + ) + await adapter._start_polling_once( + adapter._app, + drop_pending_updates=False, + error_callback=delegated, + ) + verifier = adapter._polling_progress_verifier_task + stale_error = ConnectionError("stale generation") + current_error = ConnectionError("current generation") + + try: + callbacks[0](stale_error) + delegated.assert_not_called() + + callbacks[1](current_error) + delegated.assert_called_once_with(current_error) + finally: + await _cancel_task(verifier) + + +@pytest.mark.asyncio +async def test_cold_start_waits_for_get_updates_progress_before_healing(): + adapter = _make_adapter() + adapter._app = _mock_polling_app() + + started = await adapter._start_polling_resilient( + drop_pending_updates=True, + error_callback=MagicMock(), + ) + + verifier = adapter._polling_progress_verifier_task + assert started is True + assert adapter._app.updater.running is True + assert adapter._send_path_degraded is True + assert verifier is not None and not verifier.done() + assert verifier in adapter._background_tasks + assert [task for task in adapter._background_tasks if not task.done()] == [verifier] + await _cancel_task(verifier) + + +@pytest.mark.asyncio +async def test_matching_get_updates_progress_heals_and_stops_verifier(monkeypatch): + adapter = _make_adapter() + adapter._app = _mock_polling_app() + recovery = MagicMock() + monkeypatch.setattr(adapter, "_schedule_polling_recovery", recovery) + + await adapter._start_polling_resilient( + drop_pending_updates=False, + error_callback=MagicMock(), + ) + verifier = adapter._polling_progress_verifier_task + request = adapter._instrument_polling_request( + _ControlledRequest(result=(200, b'{"ok":true,"result":[]}')) + ) + await _request_for_generation( + adapter._polling_generation, request, "getUpdates" + ) + await asyncio.wait_for(verifier, timeout=1) + + assert adapter._send_path_degraded is False + assert verifier.done() + recovery.assert_not_called() + + +@pytest.mark.asyncio +async def test_general_path_success_without_get_updates_progress_recovers_once(monkeypatch): + adapter = _make_adapter() + adapter._app = _mock_polling_app() + recovery = MagicMock() + monkeypatch.setattr(tg_adapter, "_POLLING_PROGRESS_TIMEOUT", 0.01, raising=False) + monkeypatch.setattr(adapter, "_schedule_polling_recovery", recovery) + + await adapter._start_polling_resilient( + drop_pending_updates=False, + error_callback=MagicMock(), + ) + verifier = adapter._polling_progress_verifier_task + await asyncio.wait_for(verifier, timeout=1) + + assert adapter._app.bot.get_me.await_count == 1 + recovery.assert_called_once() + error = recovery.call_args.args[0] + assert isinstance(error, RuntimeError) + assert str(error) == "getUpdates made no progress before verifier deadline" + assert adapter._send_path_degraded is True + + +@pytest.mark.asyncio +@pytest.mark.parametrize( + ("probe_error", "should_recover"), + [ + (ConnectionError("pool wedged"), True), + (type("InvalidToken", (Exception,), {})("token revoked"), False), + ], +) +async def test_general_path_error_only_recovers_connectivity_failures( + monkeypatch, probe_error, should_recover +): + adapter = _make_adapter() + adapter._app = _mock_polling_app(get_me=AsyncMock(side_effect=probe_error)) + recovery = MagicMock() + monkeypatch.setattr(tg_adapter, "_POLLING_PROGRESS_TIMEOUT", 0.01, raising=False) + monkeypatch.setattr(adapter, "_schedule_polling_recovery", recovery) + + await adapter._start_polling_resilient( + drop_pending_updates=False, + error_callback=MagicMock(), + ) + await asyncio.wait_for(adapter._polling_progress_verifier_task, timeout=1) + + assert recovery.called is should_recover + if should_recover: + assert recovery.call_args.args[0] is probe_error + assert adapter._send_path_degraded is True + + +@pytest.mark.asyncio +@pytest.mark.parametrize("retry_kind", ["network", "conflict"]) +async def test_retry_start_requires_matching_progress_to_heal(monkeypatch, retry_kind): + adapter = _make_adapter() + adapter._app = _mock_polling_app() + adapter._polling_error_callback_ref = MagicMock() + adapter._polling_network_error_count = 3 + monkeypatch.setattr(tg_adapter.asyncio, "sleep", AsyncMock()) + + if retry_kind == "network": + await adapter._handle_polling_network_error(ConnectionError("offline")) + assert adapter._polling_network_error_count == 4 + else: + await adapter._handle_polling_conflict( + RuntimeError("Conflict: terminated by other getUpdates request") + ) + assert adapter._polling_network_error_count == 3 + assert adapter._polling_conflict_count == 1 + + generation = adapter._polling_generation + verifier = adapter._polling_progress_verifier_task + assert generation > 0 + assert verifier is not None and not verifier.done() + assert adapter._send_path_degraded is True + + request = adapter._instrument_polling_request( + _ControlledRequest(result=(200, b'{"ok":true,"result":[]}')) + ) + await _request_for_generation(generation, request, "getUpdates") + await asyncio.wait_for(verifier, timeout=1) + assert adapter._polling_network_error_count == 0 + assert adapter._polling_conflict_count == 0 + assert adapter._send_path_degraded is False + + +@pytest.mark.asyncio +async def test_repeated_starts_replace_verifier_and_stale_verifier_cannot_heal(): + adapter = _make_adapter() + adapter._app = _mock_polling_app() + + await adapter._start_polling_resilient( + drop_pending_updates=False, error_callback=MagicMock() + ) + generation_1 = adapter._polling_generation + progress_1 = adapter._polling_progress_event + verifier_1 = adapter._polling_progress_verifier_task + + await adapter._start_polling_resilient( + drop_pending_updates=False, error_callback=MagicMock() + ) + verifier_2 = adapter._polling_progress_verifier_task + await asyncio.sleep(0) + + assert adapter._polling_generation == generation_1 + 1 + assert verifier_1.cancelled() + assert verifier_2 is not verifier_1 and not verifier_2.done() + assert [task for task in adapter._background_tasks if not task.done()] == [verifier_2] + + progress_1.set() + await asyncio.sleep(0) + assert adapter._send_path_degraded is True + await _cancel_task(verifier_2) + + +@pytest.mark.asyncio +async def test_disconnect_fences_verifier_and_late_progress_completion(): + adapter = _make_adapter() + adapter._app = _mock_polling_app() + await adapter._start_polling_resilient( + drop_pending_updates=False, error_callback=MagicMock() + ) + generation = adapter._polling_generation + verifier = adapter._polling_progress_verifier_task + + entered = asyncio.Event() + release = asyncio.Event() + request = adapter._instrument_polling_request( + _ControlledRequest( + result=(200, b'{"ok":true,"result":[]}'), + entered=entered, + release=release, + ) + ) + completion = asyncio.create_task( + _request_for_generation(generation, request, "getUpdates") + ) + await entered.wait() + + await adapter.disconnect() + + assert verifier.done() + assert adapter._polling_progress_verifier_task is None + assert adapter._polling_progress_accepting is False + assert adapter._polling_generation > generation + assert adapter._send_path_degraded is True + + release.set() + assert await completion == (200, b'{"ok":true,"result":[]}') + assert adapter._send_path_degraded is True + + +@pytest.mark.asyncio +async def test_disconnect_during_polling_start_returns_false_without_recovery(): + adapter = _make_adapter() + adapter._app = _mock_polling_app() + entered = asyncio.Event() + release = asyncio.Event() + recovery = MagicMock() + adapter._schedule_polling_recovery = recovery + + async def blocked_start(**_kwargs): + entered.set() + await release.wait() + + adapter._app.updater.start_polling = blocked_start + start = asyncio.create_task( + adapter._start_polling_resilient( + drop_pending_updates=False, + error_callback=MagicMock(), + ) + ) + await entered.wait() + + await adapter.disconnect() + release.set() + result = await start + + assert result is False + recovery.assert_not_called() + assert adapter._polling_error_task is None + assert not adapter.has_fatal_error + assert adapter._send_path_degraded is True + + +@pytest.mark.asyncio +async def test_caller_cancellation_during_polling_start_still_propagates(): + adapter = _make_adapter() + adapter._app = _mock_polling_app() + entered = asyncio.Event() + release = asyncio.Event() + recovery = MagicMock() + adapter._schedule_polling_recovery = recovery + + async def blocked_start(**_kwargs): + entered.set() + await release.wait() + + adapter._app.updater.start_polling = blocked_start + start = asyncio.create_task( + adapter._start_polling_resilient( + drop_pending_updates=False, + error_callback=MagicMock(), + ) + ) + await entered.wait() + + start.cancel() + with pytest.raises(asyncio.CancelledError): + await start + + assert start.cancelled() + recovery.assert_not_called() + assert not adapter.has_fatal_error + assert adapter._send_path_degraded is True + await adapter.disconnect() + + +@pytest.mark.asyncio +async def test_disconnect_cancels_recovery_before_it_can_rearm_progress(monkeypatch): + adapter = _make_adapter() + adapter._app = _mock_polling_app() + adapter._app.updater.running = False + adapter._polling_error_callback_ref = MagicMock() + + drain_entered = asyncio.Event() + release_drain = asyncio.Event() + start_entered = asyncio.Event() + release_start = asyncio.Event() + teardown_paused = asyncio.Event() + release_teardown = asyncio.Event() + + async def immediate_backoff(_delay): + return None + + async def blocked_drain(): + drain_entered.set() + await release_drain.wait() + + async def blocked_start_polling(**_kwargs): + start_entered.set() + await release_start.wait() + + async def blocked_status_indicator(*, online): + assert online is False + teardown_paused.set() + await release_teardown.wait() + + monkeypatch.setattr(tg_adapter.asyncio, "sleep", immediate_backoff) + monkeypatch.setattr(adapter, "_drain_polling_connections", blocked_drain) + monkeypatch.setattr( + adapter._app.updater, "start_polling", blocked_start_polling + ) + monkeypatch.setattr(adapter, "_set_status_indicator", blocked_status_indicator) + + recovery = asyncio.create_task( + adapter._handle_polling_network_error(ConnectionError("offline")) + ) + adapter._polling_error_task = recovery + await drain_entered.wait() + + disconnect = asyncio.create_task(adapter.disconnect()) + await teardown_paused.wait() + + try: + # Before the fix, disconnect pauses here before cancelling recovery. + # Releasing the recovery lets it begin a fresh generation after the + # teardown fence, and matching progress can then heal the adapter. + if not recovery.done(): + release_drain.set() + await start_entered.wait() + + rearmed_after_fence = adapter._polling_progress_accepting + adapter._record_polling_progress(adapter._polling_generation) + + assert rearmed_after_fence is False + assert getattr(adapter, "_polling_teardown_started", False) is True + assert adapter._polling_progress_accepting is False + assert adapter._send_path_degraded is True + assert recovery.done() + finally: + release_drain.set() + release_start.set() + release_teardown.set() + for task in (recovery, disconnect): + if not task.done(): + task.cancel() + await asyncio.gather(recovery, disconnect, return_exceptions=True) diff --git a/tests/gateway/test_telegram_send_path_health.py b/tests/gateway/test_telegram_send_path_health.py index 37c76152c913..b533b3492dd4 100644 --- a/tests/gateway/test_telegram_send_path_health.py +++ b/tests/gateway/test_telegram_send_path_health.py @@ -65,49 +65,32 @@ async def test_send_short_circuits_when_path_degraded(): @pytest.mark.asyncio -async def test_reconnect_storm_sets_and_heartbeat_clears_flag(monkeypatch): - """_handle_polling_network_error sets the flag while reconnecting; if the - reconnect attempt itself raises (polling not yet healthy), the flag stays - True until a later successful heartbeat probe in - _verify_polling_after_reconnect clears it.""" +async def test_get_me_success_without_polling_progress_does_not_heal(monkeypatch): + """A responsive general Bot API path is not proof that getUpdates works.""" adapter = _make_adapter() adapter._app = MagicMock() adapter._app.updater = MagicMock() adapter._app.updater.running = True - adapter._app.updater.stop = AsyncMock() - # First start_polling attempt fails — the reconnect handler must leave the - # flag set (path still unhealthy) and not clear it prematurely. - adapter._app.updater.start_polling = AsyncMock(side_effect=OSError("still down")) adapter._app.bot = MagicMock() adapter._app.bot.get_me = AsyncMock(return_value=MagicMock()) - adapter._polling_error_callback_ref = AsyncMock() - monkeypatch.setattr( - "plugins.platforms.telegram.adapter.Update", MagicMock(ALL_TYPES=[]) - ) - # Suppress the self-rescheduled retry so the test doesn't recurse. - monkeypatch.setattr( - "plugins.platforms.telegram.adapter.asyncio.ensure_future", MagicMock() - ) - with patch("plugins.platforms.telegram.adapter.asyncio.sleep", new_callable=AsyncMock): - await adapter._handle_polling_network_error(OSError("Bad Gateway")) - # start_polling failed → path still degraded. + generation, progress = adapter._begin_polling_generation() + recovery = MagicMock() + monkeypatch.setattr(adapter, "_schedule_polling_recovery", recovery) + monkeypatch.setattr( + "plugins.platforms.telegram.adapter._POLLING_PROGRESS_TIMEOUT", 0, + raising=False, + ) + await adapter._verify_polling_after_reconnect(generation, progress) + + adapter._app.bot.get_me.assert_awaited_once() + recovery.assert_called_once() assert adapter._send_path_degraded is True - # Now the deferred probe runs against a recovered (running) updater and - # a responsive bot — it clears the flag. - adapter._app.updater.running = True - with patch("plugins.platforms.telegram.adapter.asyncio.sleep", new_callable=AsyncMock): - await adapter._verify_polling_after_reconnect() - assert adapter._send_path_degraded is False - @pytest.mark.asyncio -async def test_successful_reconnect_clears_flag_without_probe(monkeypatch): - """Regression for #35205: a successful start_polling() clears - _send_path_degraded immediately, so outbound sends are not blocked for - the full HEARTBEAT_PROBE_DELAY window (and never get stuck True if the - deferred probe is never scheduled / never runs).""" +async def test_successful_reconnect_waits_for_get_updates_progress(monkeypatch): + """start_polling() return alone cannot heal; matching progress can.""" adapter = _make_adapter() adapter._app = MagicMock() adapter._app.updater = MagicMock() @@ -120,17 +103,18 @@ async def test_successful_reconnect_clears_flag_without_probe(monkeypatch): monkeypatch.setattr( "plugins.platforms.telegram.adapter.Update", MagicMock(ALL_TYPES=[]) ) - # Don't let the deferred probe run — prove the clear happens in the - # reconnect handler itself, not in _verify_polling_after_reconnect. - monkeypatch.setattr( - adapter, "_verify_polling_after_reconnect", AsyncMock() - ) - with patch("plugins.platforms.telegram.adapter.asyncio.sleep", new_callable=AsyncMock): await adapter._handle_polling_network_error(OSError("Bad Gateway")) + verifier = adapter._polling_progress_verifier_task + assert adapter._send_path_degraded is True + assert adapter._polling_network_error_count == 1 + blocked = await adapter.send("123", "hello") + assert blocked.success is False + + adapter._record_polling_progress(adapter._polling_generation) + await verifier assert adapter._send_path_degraded is False assert adapter._polling_network_error_count == 0 - # And send() works again right away. result = await adapter.send("123", "hello") assert result.success is True diff --git a/tests/test_telegram_polling_progress_ptb.py b/tests/test_telegram_polling_progress_ptb.py new file mode 100644 index 000000000000..ad4ed2caad84 --- /dev/null +++ b/tests/test_telegram_polling_progress_ptb.py @@ -0,0 +1,292 @@ +"""Integration coverage for polling progress against the installed PTB runtime.""" + +import asyncio + +import pytest +from telegram.error import Conflict, TelegramError +from telegram.request import BaseRequest + +from gateway.config import PlatformConfig +from plugins.platforms.telegram import adapter as tg_adapter +from plugins.platforms.telegram.adapter import TelegramAdapter + + +class _GeneralRequest(BaseRequest): + @property + def read_timeout(self): + return 10 + + async def initialize(self): + return None + + async def shutdown(self): + return None + + async def do_request(self, url, method, request_data=None, **_kwargs): + if url.endswith("/getMe"): + return ( + 200, + b'{"ok":true,"result":{"id":1,"is_bot":true,' + b'"first_name":"Test","username":"test_bot"}}', + ) + return 200, b'{"ok":true,"result":true}' + + +class _GetUpdatesRequest(BaseRequest): + def __init__(self): + self.initial_conflict_sent = False + self.replacement_enabled = False + self.replacement_progress_sent = False + self.cleanup_calls = 0 + self.block = asyncio.Event() + + @property + def read_timeout(self): + return 10 + + async def initialize(self): + return None + + async def shutdown(self): + return None + + async def do_request(self, url, method, request_data=None, **_kwargs): + parameters = request_data.parameters if request_data is not None else {} + timeout = parameters.get("timeout") + timeout_seconds = ( + timeout.total_seconds() if hasattr(timeout, "total_seconds") else timeout + ) + if timeout_seconds == 0: + self.cleanup_calls += 1 + return 200, b'{"ok":true,"result":[]}' + if not self.initial_conflict_sent: + self.initial_conflict_sent = True + return ( + 409, + b'{"ok":false,"error_code":409,' + b'"description":"Conflict: another getUpdates request"}', + ) + if self.replacement_enabled and not self.replacement_progress_sent: + self.replacement_progress_sent = True + return 200, b'{"ok":true,"result":[]}' + await self.block.wait() + return 200, b'{"ok":true,"result":[]}' + + +class _EnvelopeRequest(BaseRequest): + def __init__(self, payload): + self.payload = payload + + @property + def read_timeout(self): + return 10 + + async def initialize(self): + return None + + async def shutdown(self): + return None + + async def do_request(self, url, method, request_data=None, **_kwargs): + return 200, self.payload + + +async def _cancel_task(task): + if task is None or task.done(): + return + task.cancel() + await asyncio.gather(task, return_exceptions=True) + + +@pytest.mark.asyncio +async def test_real_base_request_invalid_200_body_cannot_record_progress(): + adapter = TelegramAdapter(PlatformConfig(enabled=True, token="123456:test-token")) + generation, progress = adapter._begin_polling_generation() + adapter._polling_network_error_count = 4 + adapter._polling_conflict_count = 3 + request = adapter._instrument_polling_request(_EnvelopeRequest(b"not-json")) + context_token = tg_adapter._POLLING_GENERATION_CONTEXT.set(generation) + + try: + with pytest.raises(TelegramError, match="Invalid server response"): + await request.post("https://api.telegram.org/bot-token/getUpdates") + finally: + tg_adapter._POLLING_GENERATION_CONTEXT.reset(context_token) + + assert not progress.is_set() + assert adapter._polling_network_error_count == 4 + assert adapter._polling_conflict_count == 3 + assert adapter._send_path_degraded is True + + +@pytest.mark.asyncio +async def test_real_base_request_bom_rejected_by_ptb_cannot_record_progress(): + adapter = TelegramAdapter(PlatformConfig(enabled=True, token="123456:test-token")) + generation, progress = adapter._begin_polling_generation() + adapter._polling_network_error_count = 4 + adapter._polling_conflict_count = 3 + request = adapter._instrument_polling_request( + _EnvelopeRequest(b'\xef\xbb\xbf{"ok":true,"result":[]}') + ) + context_token = tg_adapter._POLLING_GENERATION_CONTEXT.set(generation) + + try: + with pytest.raises(TelegramError, match="Invalid server response"): + await request.post("https://api.telegram.org/bot-token/getUpdates") + finally: + tg_adapter._POLLING_GENERATION_CONTEXT.reset(context_token) + + assert not progress.is_set() + assert adapter._polling_network_error_count == 4 + assert adapter._polling_conflict_count == 3 + assert adapter._send_path_degraded is True + + +@pytest.mark.asyncio +async def test_real_base_request_ptb_replacement_decode_records_progress(): + adapter = TelegramAdapter(PlatformConfig(enabled=True, token="123456:test-token")) + generation, progress = adapter._begin_polling_generation() + adapter._polling_network_error_count = 4 + adapter._polling_conflict_count = 3 + request = adapter._instrument_polling_request( + _EnvelopeRequest(b'{"ok":true,"result":[],"note":"\xff"}') + ) + context_token = tg_adapter._POLLING_GENERATION_CONTEXT.set(generation) + + try: + result = await request.post("https://api.telegram.org/bot-token/getUpdates") + finally: + tg_adapter._POLLING_GENERATION_CONTEXT.reset(context_token) + + assert result == [] + assert progress.is_set() + assert adapter._polling_network_error_count == 0 + assert adapter._polling_conflict_count == 0 + assert adapter._send_path_degraded is False + + +@pytest.mark.asyncio +@pytest.mark.parametrize( + ("payload", "missing_result"), + [ + (b'{"ok":false,"result":[]}', False), + (b'{"ok":true}', True), + ], +) +async def test_real_base_request_unsuccessful_200_envelope_cannot_record_progress( + payload, missing_result +): + adapter = TelegramAdapter(PlatformConfig(enabled=True, token="123456:test-token")) + generation, progress = adapter._begin_polling_generation() + adapter._polling_network_error_count = 4 + adapter._polling_conflict_count = 3 + request = adapter._instrument_polling_request(_EnvelopeRequest(payload)) + context_token = tg_adapter._POLLING_GENERATION_CONTEXT.set(generation) + + try: + if missing_result: + with pytest.raises(KeyError, match="result"): + await request.post("https://api.telegram.org/bot-token/getUpdates") + else: + assert await request.post( + "https://api.telegram.org/bot-token/getUpdates" + ) == [] + finally: + tg_adapter._POLLING_GENERATION_CONTEXT.reset(context_token) + + assert not progress.is_set() + assert adapter._polling_network_error_count == 4 + assert adapter._polling_conflict_count == 3 + assert adapter._send_path_degraded is True + + +@pytest.mark.asyncio +async def test_real_base_request_valid_success_envelope_records_progress(): + adapter = TelegramAdapter(PlatformConfig(enabled=True, token="123456:test-token")) + generation, progress = adapter._begin_polling_generation() + adapter._polling_network_error_count = 4 + adapter._polling_conflict_count = 3 + request = adapter._instrument_polling_request( + _EnvelopeRequest(b'{"ok":true,"result":[]}') + ) + context_token = tg_adapter._POLLING_GENERATION_CONTEXT.set(generation) + + try: + result = await request.post( + "https://api.telegram.org/bot-token/getUpdates" + ) + finally: + tg_adapter._POLLING_GENERATION_CONTEXT.reset(context_token) + + assert result == [] + assert progress.is_set() + assert adapter._polling_network_error_count == 0 + assert adapter._polling_conflict_count == 0 + assert adapter._send_path_degraded is False + + +@pytest.mark.asyncio +async def test_real_ptb_stop_cleanup_cannot_heal_recovery_generation(): + assert tg_adapter.TELEGRAM_AVAILABLE is True + adapter = TelegramAdapter(PlatformConfig(enabled=True, token="123456:test-token")) + polling_request = _GetUpdatesRequest() + app = ( + tg_adapter.Application.builder() + .token("123456:test-token") + .request(_GeneralRequest()) + .get_updates_request(adapter._instrument_polling_request(polling_request)) + .build() + ) + adapter._app = app + adapter._polling_network_error_count = 4 + adapter._polling_conflict_count = 3 + callback_called = asyncio.Event() + recovery_task = None + + async def stop_for_recovery(): + await app.updater.stop() + + def schedule_recovery(error): + nonlocal recovery_task + assert isinstance(error, Conflict) + recovery_task = asyncio.create_task(stop_for_recovery()) + callback_called.set() + + await app.initialize() + try: + await adapter._start_polling_once( + app, + drop_pending_updates=False, + error_callback=schedule_recovery, + ) + generation = adapter._polling_generation + progress = adapter._polling_progress_event + await asyncio.wait_for(callback_called.wait(), timeout=2) + await asyncio.wait_for(recovery_task, timeout=3) + + assert polling_request.cleanup_calls == 1 + assert not progress.is_set() + assert adapter._polling_network_error_count == 4 + assert adapter._polling_conflict_count == 3 + assert adapter._send_path_degraded is True + + polling_request.replacement_enabled = True + await adapter._start_polling_once( + app, + drop_pending_updates=False, + error_callback=schedule_recovery, + ) + replacement_generation = adapter._polling_generation + replacement_progress = adapter._polling_progress_event + await asyncio.wait_for(replacement_progress.wait(), timeout=2) + + assert replacement_generation == generation + 1 + assert adapter._polling_network_error_count == 0 + assert adapter._polling_conflict_count == 0 + assert adapter._send_path_degraded is False + finally: + polling_request.block.set() + if app.updater.running: + await app.updater.stop() + await _cancel_task(adapter._polling_progress_verifier_task) + await app.shutdown()