fix(gateway): queue reconnect before fatal disconnect wedges (#80598)
After a network outage the Telegram fatal handler could hang inside disconnect() and never populate _failed_platforms, so the reconnect watcher had nothing to retry and the process stayed permanently deaf. Queue retryable platforms before any disconnect await, bound the fatal handler with an outer detach deadline, and release the Telegram token lock / PTB close steps with detach-on-timeout so recovery cannot stall.
This commit is contained in:
HexLab98
kshitij
parent
6d1f9f8ed4
commit
7141a6dc3a
104
gateway/run.py
104
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:
|
||||
|
|
|
|||
|
|
@ -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
|
||||
|
|
|
|||
Loading…
Reference in New Issue