diff --git a/plugins/platforms/telegram/telegram_network.py b/plugins/platforms/telegram/telegram_network.py index 5b1d8d12bf..aad9361767 100644 --- a/plugins/platforms/telegram/telegram_network.py +++ b/plugins/platforms/telegram/telegram_network.py @@ -71,6 +71,8 @@ class TelegramFallbackTransport(httpx.AsyncBaseTransport): transport_kwargs.setdefault("limits", self._POOL_LIMITS) self._transport_kwargs = transport_kwargs self._primary = httpx.AsyncHTTPTransport(**transport_kwargs) + self._primary_lock = asyncio.Lock() + self._primary_closed = False # Built on demand and discarded on failure — see _reset_fallback. self._fallbacks: dict[str, httpx.AsyncHTTPTransport] = {} self._fallback_lock = asyncio.Lock() @@ -85,6 +87,18 @@ class TelegramFallbackTransport(httpx.AsyncBaseTransport): self._fallbacks[ip] = transport return transport + async def _reset_primary(self, transport: httpx.AsyncHTTPTransport) -> None: + # Retryable primary failures can leave half-closed sockets in the pool; + # replace and close the failed generation before trying fallback. + async with self._primary_lock: + if self._primary_closed or transport is not self._primary: + return + self._primary = httpx.AsyncHTTPTransport(**self._transport_kwargs) + try: + await transport.aclose() + except Exception as exc: + logger.debug("[Telegram] Error closing primary transport: %s", exc) + async def _reset_fallback(self, ip: str) -> None: """Discard a failed fallback pool so its dead sockets are released. @@ -142,6 +156,7 @@ class TelegramFallbackTransport(httpx.AsyncBaseTransport): ip, ) if ip is None: + await self._reset_primary(transport) logger.warning( "[Telegram] Primary api.telegram.org connection failed (%s); trying fallback IPs %s", exc, @@ -157,7 +172,10 @@ class TelegramFallbackTransport(httpx.AsyncBaseTransport): raise last_error async def aclose(self) -> None: - await self._primary.aclose() + async with self._primary_lock: + self._primary_closed = True + primary = self._primary + await primary.aclose() async with self._fallback_lock: transports = list(self._fallbacks.values()) self._fallbacks.clear() diff --git a/tests/gateway/test_telegram_closewait_limits_31599.py b/tests/gateway/test_telegram_closewait_limits_31599.py index cc8e277040..3c46bf5760 100644 --- a/tests/gateway/test_telegram_closewait_limits_31599.py +++ b/tests/gateway/test_telegram_closewait_limits_31599.py @@ -16,13 +16,13 @@ The fix wires the shared ``gateway.platforms._http_client_limits`` builds — the fallback-transport branch, the proxy branch, and the plain branch — so idle keepalive sockets drain aggressively. -Contract asserted here (mutation-survivable) ---------------------------------------------- -Every ``HTTPXRequest`` constructed by ``TelegramAdapter.connect()`` must -receive ``httpx_kwargs["limits"]`` that is an ``httpx.Limits`` with a -``keepalive_expiry`` strictly below httpx's 5.0 default and a positive, -bounded ``max_keepalive_connections``. Reverting the limits wiring (so -HTTPXRequest falls back to PTB's default 5.0s keepalive) fails this test. +Contracts asserted here (mutation-survivable) +---------------------------------------------- +Proxy and direct-DNS ``HTTPXRequest`` instances must receive +``httpx_kwargs["limits"]`` with a ``keepalive_expiry`` strictly below +httpx's 5.0 default. The fallback-IP instances must pass equivalent +limits into both inner ``AsyncHTTPTransport`` pools because httpx ignores +client-level limits when a custom transport is supplied. """ import asyncio @@ -74,7 +74,7 @@ def _make_adapter() -> TelegramAdapter: return TelegramAdapter(PlatformConfig(enabled=True, token="test-token")) -def _drive_connect(monkeypatch, *, proxy_url): +def _drive_connect(monkeypatch, *, proxy_url, fallback_ips=None): """Run connect() far enough to build the HTTPXRequests, then abort. Returns the list of recorded _RecordingHTTPXRequest instances. @@ -83,7 +83,7 @@ def _drive_connect(monkeypatch, *, proxy_url): # No DoH auto-discovery → exercise the proxy / plain branches, not fallback. async def _no_fallback(): - return [] + return list(fallback_ips or []) monkeypatch.setattr(tg_adapter, "discover_fallback_ips", _no_fallback) monkeypatch.setattr( @@ -98,6 +98,9 @@ def _drive_connect(monkeypatch, *, proxy_url): # Ensure the adapter reports no statically-configured fallback IPs. monkeypatch.setattr(adapter, "_fallback_ips", lambda: []) + if fallback_ips is not None: + monkeypatch.setattr(adapter, "_fallback_ips", lambda: list(fallback_ips)) + # builder.request(...).get_updates_request(...).build() must be harmless; # make build() raise our sentinel so connect() stops right after the # HTTPXRequests are constructed (before any real network/init). @@ -160,3 +163,25 @@ def test_proxy_branch_general_pool_has_tight_keepalive(monkeypatch): assert any(inst.kwargs.get("proxy") == "http://127.0.0.1:9/" for inst in instances) +def test_fallback_branch_forwards_tuned_limits_to_inner_transports(monkeypatch): + monkeypatch.delenv("HERMES_TELEGRAM_HTTP_POOL_SIZE", raising=False) + monkeypatch.delenv("HERMES_GATEWAY_HTTPX_KEEPALIVE_EXPIRY", raising=False) + + instances = _drive_connect( + monkeypatch, + proxy_url=None, + fallback_ips=["149.154.167.220"], + ) + + assert len(instances) >= 2 + for instance in instances: + transport = instance.kwargs["httpx_kwargs"]["transport"] + assert isinstance(transport, tg_adapter.TelegramFallbackTransport) + limits = transport._transport_kwargs["limits"] + assert isinstance(limits, httpx.Limits) + assert limits.keepalive_expiry is not None + assert limits.keepalive_expiry < 5.0 + assert limits.max_connections == 512 + + for instance in instances: + asyncio.run(instance.kwargs["httpx_kwargs"]["transport"].aclose()) diff --git a/tests/gateway/test_telegram_fallback_pool_release_71593.py b/tests/gateway/test_telegram_fallback_pool_release_71593.py index 70db93bc2b..d50cd36b96 100644 --- a/tests/gateway/test_telegram_fallback_pool_release_71593.py +++ b/tests/gateway/test_telegram_fallback_pool_release_71593.py @@ -105,13 +105,44 @@ async def test_failed_fallback_pool_is_discarded_and_closed(monkeypatch): "CLOSE_WAIT sockets leak (revert of _reset_fallback? #71593)." ) - # Each fallback pool that was built for a failing IP must have been - # aclose()d exactly once (two fallback IPs → two discards). - assert len(closed_log) == 2, ( - f"Expected 2 discarded/closed fallback pools, got {len(closed_log)} — " + # The failed primary plus each fallback pool must be + # aclose()d exactly once (one primary + two fallback IPs). + assert len(closed_log) == 3, ( + f"Expected 3 discarded/closed transports, got {len(closed_log)} — " "the discard-on-failure path did not aclose() the poisoned pools." ) assert all(t.closed for t in closed_log) + await transport.aclose() + + +@pytest.mark.asyncio +async def test_failed_primary_pool_is_discarded_and_closed(monkeypatch): + """A failed primary attempt must release its pool before fallback (#82920).""" + for key in ( + "HTTPS_PROXY", "HTTP_PROXY", "ALL_PROXY", "https_proxy", + "http_proxy", "all_proxy", "TELEGRAM_PROXY", "NO_PROXY", "no_proxy", + ): + monkeypatch.delenv(key, raising=False) + + behavior = {"api.telegram.org": "timeout", "149.154.167.220": "ok"} + instances = [] + + def factory(**kwargs): + transport = _CountingTransport(behavior, []) + instances.append(transport) + return transport + + monkeypatch.setattr(tnet.httpx, "AsyncHTTPTransport", factory) + transport = tnet.TelegramFallbackTransport(["149.154.167.220"]) + try: + response = await transport.handle_async_request(_telegram_request()) + assert response.status_code == 200 + assert len(instances) == 3 + assert instances[0].closed + assert not instances[1].closed + assert not instances[2].closed + finally: + await transport.aclose() def test_caller_limits_win_over_pool_default(monkeypatch):