From ca2491f201995045496d37a18d65be8344251f54 Mon Sep 17 00:00:00 2001 From: alexneyman Date: Sat, 25 Jul 2026 17:21:27 -0400 Subject: [PATCH] fix(telegram): release fallback transport pools on connect failure MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit The per-IP httpx transports were built once in __init__ and never torn down. A connect that reached ESTABLISHED and was then closed by the peer left its socket in CLOSE_WAIT inside the pool, and the failure path only logged and continued — so the poisoned pool was retained and leaked one descriptor per retry. With DNS for api.telegram.org failing, every poll fell through to the seed IP and leaked another fd every ~2.5s. The bot gateway reached 177 CLOSE_WAIT sockets against launchd's 256 soft limit and wedged: accept() on the gateway port, config reads and DNS resolution all failed with EMFILE, which in turn made the primary path fail and fed the loop. Build fallback transports lazily and discard them on a retryable connect failure, and bound every pool at 8 connections (httpx defaults to 100, so two seed IPs plus primary could alone exceed the fd ceiling). Generated with [Claude Code](https://claude.ai/code) via [Happy](https://happy.engineering) Co-Authored-By: Claude Co-Authored-By: Happy --- contributors/emails/a.neyman17@gmail.com | 1 + .../platforms/telegram/telegram_network.py | 46 +++++++++++++++++-- 2 files changed, 42 insertions(+), 5 deletions(-) create mode 100644 contributors/emails/a.neyman17@gmail.com diff --git a/contributors/emails/a.neyman17@gmail.com b/contributors/emails/a.neyman17@gmail.com new file mode 100644 index 00000000000..1f4c332c51b --- /dev/null +++ b/contributors/emails/a.neyman17@gmail.com @@ -0,0 +1 @@ +aneym diff --git a/plugins/platforms/telegram/telegram_network.py b/plugins/platforms/telegram/telegram_network.py index a0fd14ebb5d..5b1d8d12bfb 100644 --- a/plugins/platforms/telegram/telegram_network.py +++ b/plugins/platforms/telegram/telegram_network.py @@ -58,18 +58,50 @@ class TelegramFallbackTransport(httpx.AsyncBaseTransport): ``curl --resolve api.telegram.org:443:``. """ + # Bound every pool. httpx defaults to 100 connections per pool, so a wedged + # endpoint plus the seed IPs can outgrow the process file-descriptor limit + # on its own (#63311). + _POOL_LIMITS = httpx.Limits(max_connections=8, max_keepalive_connections=4) + def __init__(self, fallback_ips: Iterable[str], **transport_kwargs): self._fallback_ips = list(dict.fromkeys(_normalize_fallback_ips(fallback_ips))) proxy_url = _resolve_proxy_url(target_hosts=[_TELEGRAM_API_HOST, *self._fallback_ips]) if proxy_url and "proxy" not in transport_kwargs: transport_kwargs["proxy"] = proxy_url + transport_kwargs.setdefault("limits", self._POOL_LIMITS) + self._transport_kwargs = transport_kwargs self._primary = httpx.AsyncHTTPTransport(**transport_kwargs) - self._fallbacks = { - ip: httpx.AsyncHTTPTransport(**transport_kwargs) for ip in self._fallback_ips - } + # Built on demand and discarded on failure — see _reset_fallback. + self._fallbacks: dict[str, httpx.AsyncHTTPTransport] = {} + self._fallback_lock = asyncio.Lock() self._sticky_ip: Optional[str] = None self._sticky_lock = asyncio.Lock() + async def _get_fallback(self, ip: str) -> httpx.AsyncHTTPTransport: + async with self._fallback_lock: + transport = self._fallbacks.get(ip) + if transport is None: + transport = httpx.AsyncHTTPTransport(**self._transport_kwargs) + self._fallbacks[ip] = transport + return transport + + async def _reset_fallback(self, ip: str) -> None: + """Discard a failed fallback pool so its dead sockets are released. + + A connect that reaches ESTABLISHED and is then closed by the peer leaves + its socket in CLOSE_WAIT inside the pool. Retaining the poisoned pool + leaks one descriptor per retry until the process hits its file limit and + can no longer accept connections or resolve DNS (#63311). + """ + async with self._fallback_lock: + transport = self._fallbacks.pop(ip, None) + if transport is None: + return + try: + await transport.aclose() + except Exception as exc: # closing a broken pool must never mask the real error + logger.debug("[Telegram] Error closing fallback transport %s: %s", ip, exc) + async def handle_async_request(self, request: httpx.Request) -> httpx.Response: if request.url.host != _TELEGRAM_API_HOST or not self._fallback_ips: return await self._primary.handle_async_request(request) @@ -85,7 +117,7 @@ class TelegramFallbackTransport(httpx.AsyncBaseTransport): last_error: Exception | None = None for ip in attempt_order: candidate = request if ip is None else _rewrite_request_for_ip(request, ip) - transport = self._primary if ip is None else self._fallbacks[ip] + transport = self._primary if ip is None else await self._get_fallback(ip) try: response = await transport.handle_async_request(candidate) if ip is not None and self._sticky_ip != ip: @@ -117,6 +149,7 @@ class TelegramFallbackTransport(httpx.AsyncBaseTransport): ) continue logger.warning("[Telegram] Fallback IP %s failed: %s", ip, exc) + await self._reset_fallback(ip) continue if last_error is None: @@ -125,7 +158,10 @@ class TelegramFallbackTransport(httpx.AsyncBaseTransport): async def aclose(self) -> None: await self._primary.aclose() - for transport in self._fallbacks.values(): + async with self._fallback_lock: + transports = list(self._fallbacks.values()) + self._fallbacks.clear() + for transport in transports: await transport.aclose()