diff --git a/gateway/run.py b/gateway/run.py index 5cced736f0c5c..ddb7de0e18516 100644 --- a/gateway/run.py +++ b/gateway/run.py @@ -7254,6 +7254,33 @@ class GatewayRunner(GatewayAuthorizationMixin, GatewayKanbanWatchersMixin, Gatew # it: the caller sees CancelledError, the handler runs to completion. await asyncio.shield(task) + def _queue_retryable_fatal_platform(self, adapter: BasePlatformAdapter) -> bool: + """Queue a retryable fatal adapter for background reconnection. + + Returns True when the platform was newly queued. Idempotent if already + queued. Must not await: callers invoke this *before* any disconnect + await so a wedged close cannot strand the platform (#80598). + """ + if not adapter.fatal_error_retryable: + return False + platform_config = self.config.platforms.get(adapter.platform) + if not platform_config or adapter.platform in self._failed_platforms: + return False + self._failed_platforms[adapter.platform] = { + "config": platform_config, + "attempts": 0, + "next_retry": time.monotonic(), + } + logger.info( + "%s queued for background reconnection", + adapter.platform.value, + ) + # Ensure the reconnect watcher is alive — if it died (e.g. from + # exhausting its restart budget), respawn it so queued platforms + # are not permanently stranded (#70344). + self._ensure_reconnect_watcher_running() + return True + async def _handle_adapter_fatal_error_detached( self, adapter: BasePlatformAdapter ) -> None: @@ -7262,12 +7289,57 @@ class GatewayRunner(GatewayAuthorizationMixin, GatewayKanbanWatchersMixin, Gatew gateway with failure so the service manager restarts it instead of leaving a silent partial outage.""" try: - await self._handle_adapter_fatal_error_impl(adapter) + # Outer hard deadline (#80598): even with queue-before-disconnect, + # a hang anywhere in the impl (status write side effects, detach + # races, etc.) must not leave this task wedged forever — the + # stranded check in ``finally`` only runs when we return. + timeout = self._adapter_disconnect_timeout_secs() + if timeout <= 0: + await self._handle_adapter_fatal_error_impl(adapter) + else: + # Disconnect budget plus a small overhead for queue/status + # bookkeeping. Keep the additive proportional so tests that + # shrink the disconnect timeout still finish promptly. + outer = timeout + min(2.0, max(0.05, timeout)) + completed = await self._await_adapter_cleanup_with_timeout( + self._handle_adapter_fatal_error_impl(adapter), + outer, + ) + if not completed: + logger.error( + "Fatal-error handling for %s timed out after %.1fs; " + "ensuring reconnect queue is populated", + adapter.platform.value, + outer, + ) + self._queue_retryable_fatal_platform(adapter) + except asyncio.CancelledError: + # Best-effort queue before re-raising: a cancelled fatal handler + # must not strand a retryable platform (#80598). + try: + self._queue_retryable_fatal_platform(adapter) + except Exception: + logger.debug( + "Failed to queue %s after fatal-handler cancellation", + adapter.platform.value, + exc_info=True, + ) + raise except Exception: logger.exception( "Fatal-error handling for %s raised unexpectedly", adapter.platform.value, ) + # Best-effort queue so an unexpected raise mid-handler cannot + # leave a retryable platform permanently deaf (#80598). + try: + self._queue_retryable_fatal_platform(adapter) + except Exception: + logger.debug( + "Failed to queue %s after fatal-handler exception", + adapter.platform.value, + exc_info=True, + ) finally: platform = adapter.platform shutdown_event = getattr(self, "_shutdown_event", None) @@ -7338,29 +7410,21 @@ class GatewayRunner(GatewayAuthorizationMixin, GatewayKanbanWatchersMixin, Gatew # the same object twice. self.adapters.pop(adapter.platform, None) self.delivery_router.adapters = self.adapters + + # Queue retryable failures BEFORE any disconnect await (#80598). + # A half-dead transport can wedge native close() (or swallow + # CancelledError inside it) so the previous "disconnect then queue" + # order left platforms permanently deaf inside a live process even + # after the network recovered. Populate the queue first so the + # reconnect watcher always has work; teardown is best-effort after. + self._queue_retryable_fatal_platform(adapter) + + if existing is adapter: # A half-closed transport can wedge an adapter's native close() # indefinitely. Reuse the shutdown-path timeout so this runtime - # fatal handler always reaches the reconnect queue. + # fatal handler always returns to the stay-alive / stranded path. await self._safe_adapter_disconnect(adapter, adapter.platform) - # Queue retryable failures for background reconnection - if adapter.fatal_error_retryable: - platform_config = self.config.platforms.get(adapter.platform) - if platform_config and adapter.platform not in self._failed_platforms: - self._failed_platforms[adapter.platform] = { - "config": platform_config, - "attempts": 0, - "next_retry": time.monotonic(), - } - logger.info( - "%s queued for background reconnection", - adapter.platform.value, - ) - # Ensure the reconnect watcher is alive — if it died (e.g. from - # exhausting its restart budget), respawn it so queued platforms - # are not permanently stranded (#70344). - self._ensure_reconnect_watcher_running() - if not self.adapters and not self._failed_platforms: self._exit_reason = adapter.fatal_error_message or "All messaging adapters disconnected" if adapter.fatal_error_retryable: diff --git a/plugins/platforms/telegram/adapter.py b/plugins/platforms/telegram/adapter.py index 0a7f8dea7b127..59da0a5a089bc 100644 --- a/plugins/platforms/telegram/adapter.py +++ b/plugins/platforms/telegram/adapter.py @@ -578,6 +578,11 @@ def _rich_normalize_linebreaks(text: str) -> str: # reconnect/teardown ladder. This is an internal safety bound (not a user knob), # applied identically at every stop() site so no path can hang on a dead socket. _UPDATER_STOP_TIMEOUT = 15.0 +# Per-step bound for disconnect() awaits that are not updater.stop() itself. +# Kept short so a cancellation-swallowing lifecycle/PTB close cannot burn the +# gateway's whole fatal-handler budget before the reconnect queue is useful +# (#80598). updater.stop() keeps the longer _UPDATER_STOP_TIMEOUT. +_DISCONNECT_STEP_TIMEOUT = 2.0 # start_polling() can also hang when the connection pool is in a degraded state # after _drain_polling_connections(), particularly when both primary and fallback # Telegram endpoints are unreachable. Bounding start_polling() prevents the @@ -4303,6 +4308,38 @@ class TelegramAdapter(BasePlatformAdapter): if getattr(self, "_polling_progress_verifier_task", None) is not current_task: self._polling_progress_verifier_task = None + async def _await_disconnect_step(self, awaitable, timeout: float, step: str) -> bool: + """Await one disconnect step; detach on timeout so teardown advances. + + ``asyncio.wait_for`` cancels an overdue child but then waits for it to + exit. Lifecycle / PTB close paths that swallow ``CancelledError`` on a + half-dead socket can therefore wedge disconnect forever (#80598). + Detach at the deadline and continue — the abandoned task is observed + via ``_consume_abandoned_task``. + """ + task = asyncio.ensure_future(awaitable) + if timeout <= 0: + done, _pending = await asyncio.wait({task}) + else: + done, _pending = await asyncio.wait({task}, timeout=timeout) + if task in done: + # Intentional cancels (heartbeat / identity / lifecycle) surface as + # CancelledError — swallow so disconnect keeps advancing. + try: + await task + except asyncio.CancelledError: + pass + return True + task.cancel() + task.add_done_callback(_consume_abandoned_task) + logger.warning( + "[%s] %s timed out after %.1fs during disconnect; continuing teardown", + self.name, + step, + timeout, + ) + return False + async def disconnect(self) -> None: """Stop polling/webhook, cancel pending delayed deliveries, and disconnect.""" # Mark disconnected first so the drop guard short-circuits any flush @@ -4315,6 +4352,11 @@ class TelegramAdapter(BasePlatformAdapter): self._polling_progress_event = asyncio.Event() self._send_path_degraded = True + # Release the bot-token lock immediately so a wedged close cannot block + # the reconnect watcher from acquiring it (#80598). The rest of teardown + # is best-effort against a half-dead transport. + self._release_platform_lock() + # 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. @@ -4335,7 +4377,11 @@ class TelegramAdapter(BasePlatformAdapter): if asyncio.isfuture(task) or asyncio.iscoroutine(task): lifecycle_tasks.append(task) if lifecycle_tasks: - await asyncio.gather(*lifecycle_tasks, return_exceptions=True) + await self._await_disconnect_step( + asyncio.gather(*lifecycle_tasks, return_exceptions=True), + _DISCONNECT_STEP_TIMEOUT, + "lifecycle-task cancel", + ) 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: @@ -4353,7 +4399,11 @@ class TelegramAdapter(BasePlatformAdapter): post_connect_task = getattr(self, "_post_connect_task", None) if post_connect_task and not post_connect_task.done(): post_connect_task.cancel() - await asyncio.gather(post_connect_task, return_exceptions=True) + await self._await_disconnect_step( + asyncio.gather(post_connect_task, return_exceptions=True), + _DISCONNECT_STEP_TIMEOUT, + "post-connect cancel", + ) self._post_connect_task = None # Cancel the heartbeat before tearing down the app so the probe task @@ -4361,10 +4411,11 @@ class TelegramAdapter(BasePlatformAdapter): 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 polling_heartbeat_task - except asyncio.CancelledError: - pass + await self._await_disconnect_step( + polling_heartbeat_task, + _DISCONNECT_STEP_TIMEOUT, + "heartbeat cancel", + ) self._polling_heartbeat_task = None # Cancel the webhook-mode identity refresh loop on the same fence as @@ -4372,10 +4423,11 @@ class TelegramAdapter(BasePlatformAdapter): identity_task = getattr(self, "_bot_identity_refresh_task", None) if identity_task and not identity_task.done(): identity_task.cancel() - try: - await identity_task - except asyncio.CancelledError: - pass + await self._await_disconnect_step( + identity_task, + _DISCONNECT_STEP_TIMEOUT, + "identity-refresh cancel", + ) self._bot_identity_refresh_task = None # Mark the bot "Offline" in its short description while the bot's HTTP @@ -4384,11 +4436,19 @@ class TelegramAdapter(BasePlatformAdapter): # a hard crash leaves the last-known status, which is the expected # limitation of a profile-text indicator. try: - await self._set_status_indicator(online=False) + await self._await_disconnect_step( + self._set_status_indicator(online=False), + _DISCONNECT_STEP_TIMEOUT, + "status-indicator update", + ) except Exception: pass - await self._cancel_pending_delivery_tasks() + await self._await_disconnect_step( + self._cancel_pending_delivery_tasks(), + _DISCONNECT_STEP_TIMEOUT, + "pending-delivery cancel", + ) if self._app: try: @@ -4399,22 +4459,35 @@ class TelegramAdapter(BasePlatformAdapter): # we fall through to app.stop()/shutdown() to force teardown. if self._app.updater and self._app.updater.running: try: - await asyncio.wait_for(self._app.updater.stop(), timeout=_UPDATER_STOP_TIMEOUT) - except asyncio.TimeoutError: - logger.warning( - "[%s] updater.stop() timed out during disconnect " - "(likely CLOSE-WAIT socket); forcing app shutdown", - self.name, + await self._await_disconnect_step( + self._app.updater.stop(), + _UPDATER_STOP_TIMEOUT, + "updater.stop()", ) + except Exception as stop_error: + logger.warning( + "[%s] updater.stop() failed during disconnect: %s", + self.name, + _redact_telegram_error_text(stop_error), + ) + # app.stop()/shutdown() can also block on a half-dead httpx + # pool. Detach-on-timeout so disconnect always returns (#80598). if self._app.running: - await self._app.stop() - await self._app.shutdown() + await self._await_disconnect_step( + self._app.stop(), + _DISCONNECT_STEP_TIMEOUT, + "app.stop()", + ) + await self._await_disconnect_step( + self._app.shutdown(), + _DISCONNECT_STEP_TIMEOUT, + "app.shutdown()", + ) except Exception as e: logger.warning( "[%s] Error during Telegram disconnect: %s", self.name, _redact_telegram_error_text(e), ) - self._release_platform_lock() self._app = None self._bot = None