fix(telegram): recover Windows CLOSE-WAIT getUpdates deadlock

After updater.stop() times out, HTTPXRequest.initialize() is a no-op unless
the client is already closed, so start_polling reused the wedged socket and
the gateway stayed alive but deaf. Rebuild the polling client after a hung
drain, watch getUpdates I/O independently of get_me(), and enable TCP
keepalive on the fallback transport.
This commit is contained in:
HexLab98
2026-08-15 23:04:10 +07:00
committed by Teknium
parent 7cefa87ea7
commit a06c0d0a2f
2 changed files with 101 additions and 0 deletions
+69
View File
@@ -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={
@@ -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()