fix(telegram): reset failed primary transport pool

Retryable primary errors can leave pooled sockets in CLOSE_WAIT while fallback retries continue. Replace and close failed primary generation before fallback selection.\n\nRefs #82920

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()

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())

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):