diff --git a/plugins/platforms/telegram/adapter.py b/plugins/platforms/telegram/adapter.py index a6eb37fa00..6dbc588650 100644 --- a/plugins/platforms/telegram/adapter.py +++ b/plugins/platforms/telegram/adapter.py @@ -242,6 +242,7 @@ from plugins.platforms.telegram.telegram_network import ( TelegramFallbackTransport, discover_fallback_ips, parse_fallback_ip_env, + tcp_keepalive_socket_options, ) from utils import atomic_replace, env_float, env_int @@ -2494,6 +2495,7 @@ class TelegramAdapter(BasePlatformAdapter): polling_req = self._app.bot._request[0] # noqa: SLF001 except Exception: return + shutdown_ok = False try: # Bounded: a wedged CLOSE-WAIT socket can make this close hang # forever and freeze the reconnect ladder (#66377). The wall-clock @@ -2504,11 +2506,19 @@ class TelegramAdapter(BasePlatformAdapter): await _await_with_thread_deadline( polling_req.shutdown(), timeout=_DRAIN_TIMEOUT ) + shutdown_ok = True except Exception: logger.debug( "[%s] Polling request shutdown failed/timed out (non-fatal)", self.name, exc_info=True, ) + if not shutdown_ok: + # HTTPXRequest.initialize() rebuilds the client only when + # ``client.is_closed``. An abandoned aclose() leaves that flag + # false, so initialize() is a no-op and start_polling reuses the + # CLOSE-WAIT getUpdates socket — the gateway stays alive but + # deaf (#87057). Swap in a fresh client before initialize(). + self._orphan_and_rebuild_polling_client(polling_req) try: await _await_with_thread_deadline( polling_req.initialize(), timeout=_DRAIN_TIMEOUT @@ -2521,6 +2531,64 @@ class TelegramAdapter(BasePlatformAdapter): "[%s] Polling request re-initialize failed/timed out (non-fatal)", self.name, exc_info=True, ) + self._orphan_and_rebuild_polling_client(polling_req) + + def _orphan_and_rebuild_polling_client(self, polling_req) -> None: + """Replace a wedged HTTPXRequest client after a hung aclose(). + + PTB's ``HTTPXRequest.initialize()`` only calls ``_build_client()`` + when the current client reports ``is_closed``. If ``shutdown()`` was + abandoned on a CLOSE-WAIT socket, that flag stays false and the next + ``start_polling()`` reuses the dead getUpdates connection (#87057). + Swap in a fresh client and close the old one in a detached, bounded + background task so it cannot block the reconnect ladder. + """ + old = getattr(polling_req, "_client", None) + build = getattr(polling_req, "_build_client", None) + if old is None or not callable(build): + return + if getattr(old, "is_closed", True): + return + try: + polling_req._client = build() # noqa: SLF001 + except Exception: + logger.debug( + "[%s] Failed to rebuild polling HTTP client after hung drain", + self.name, exc_info=True, + ) + return + logger.warning( + "[%s] Replaced wedged getUpdates HTTP client after drain timeout " + "(likely CLOSE-WAIT socket)", + self.name, + ) + + async def _orphan_aclose() -> None: + try: + aclose = getattr(old, "aclose", None) + if not callable(aclose): + return + # The stale client can be wedged in the same cancellation- + # swallowing httpcore scope as shutdown(). Use the wall-clock + # thread deadline — not asyncio.wait_for — so this cleanup + # task cannot itself hang forever and accumulate one leaked + # task per reconnect attempt (#87265 review). + await _await_with_thread_deadline( + aclose(), timeout=_DRAIN_TIMEOUT + ) + except Exception: + logger.debug( + "[%s] Orphan polling client aclose failed (non-fatal)", + self.name, exc_info=True, + ) + + try: + task = asyncio.ensure_future(_orphan_aclose()) + self._background_tasks.add(task) + task.add_done_callback(self._background_tasks.discard) + task.add_done_callback(_consume_abandoned_task) + except Exception: + pass def _begin_polling_generation(self) -> tuple[int, asyncio.Event]: """Start accepting progress for a new getUpdates polling generation.""" @@ -4604,6 +4672,7 @@ class TelegramAdapter(BasePlatformAdapter): _transport_kwargs: dict = {} if _pool_limits is not None: _transport_kwargs["limits"] = _pool_limits + _transport_kwargs["socket_options"] = tcp_keepalive_socket_options() request = HTTPXRequest( **request_kwargs, httpx_kwargs={ diff --git a/plugins/platforms/telegram/telegram_network.py b/plugins/platforms/telegram/telegram_network.py index 9ba5ebc005..38d2b785a7 100644 --- a/plugins/platforms/telegram/telegram_network.py +++ b/plugins/platforms/telegram/telegram_network.py @@ -21,6 +21,37 @@ logger = logging.getLogger(__name__) _TELEGRAM_API_HOST = "api.telegram.org" +# TCP keepalive so a half-open or CLOSE-WAIT long-poll errors out instead of +# blocking getUpdates indefinitely. Windows does not enable SO_KEEPALIVE on +# new sockets by default, so a dead api.telegram.org peer can hang forever +# (#87057). Idle/interval knobs are best-effort — not every Python/OS combo +# exposes TCP_KEEPIDLE / TCP_KEEPALIVE. +_TCP_KEEPALIVE_IDLE_S = 30 +_TCP_KEEPALIVE_INTERVAL_S = 10 +_TCP_KEEPALIVE_COUNT = 3 + + +def tcp_keepalive_socket_options() -> list[tuple[int, int, int]]: + """Return ``setsockopt`` tuples that enable TCP keepalive on new sockets. + + Pure data for httpx/httpcore ``socket_options``. Safe on every host: the + list always includes ``SO_KEEPALIVE`` and adds idle/interval/count only + when the running interpreter exposes those option names. + """ + options: list[tuple[int, int, int]] = [ + (socket.SOL_SOCKET, socket.SO_KEEPALIVE, 1), + ] + idle = getattr(socket, "TCP_KEEPIDLE", None) or getattr(socket, "TCP_KEEPALIVE", None) + if idle is not None: + options.append((socket.IPPROTO_TCP, idle, _TCP_KEEPALIVE_IDLE_S)) + interval = getattr(socket, "TCP_KEEPINTVL", None) + if interval is not None: + options.append((socket.IPPROTO_TCP, interval, _TCP_KEEPALIVE_INTERVAL_S)) + count = getattr(socket, "TCP_KEEPCNT", None) + if count is not None: + options.append((socket.IPPROTO_TCP, count, _TCP_KEEPALIVE_COUNT)) + return options + # DNS-over-HTTPS providers used to discover Telegram API IPs that may differ # from the (potentially unreachable) IP returned by the local system resolver. _DOH_TIMEOUT = 4.0 # seconds — bounded so connect() isn't noticeably delayed @@ -73,6 +104,7 @@ class TelegramFallbackTransport(httpx.AsyncBaseTransport): if proxy_url and "proxy" not in transport_kwargs: transport_kwargs["proxy"] = proxy_url transport_kwargs.setdefault("limits", self._POOL_LIMITS) + transport_kwargs.setdefault("socket_options", tcp_keepalive_socket_options()) self._transport_kwargs = transport_kwargs self._primary = httpx.AsyncHTTPTransport(**transport_kwargs) self._primary_lock = asyncio.Lock()