Telegram CLOSE_WAIT fd leak keepalive fix

Fixes on record

Merged to main. A stamp counts the issues a fix is linked to.

• #58982 fix(telegram): forward keepalive limits into fallback transport (stop CLOSE_WAIT fd leak)

  merged to main

  files: plugins/platforms/telegram/adapter.py, tests/gateway/test_telegram_network.py

  Show the diff

  • plugins/platforms/telegram/adapter.py

    @@ -3025,8 +3025,11 @@ def _env_float(name: str, default: float) -> float:
                 def _with_limits(httpx_kwargs: Optional[dict] = None) -> dict:
                     """Merge tuned keepalive limits into httpx client kwargs.
     
    -                A caller-supplied ``limits`` (none today) is left untouched;
    -                otherwise the CLOSE_WAIT-safe limits are injected.
    +                Used by the proxy and direct-DNS branches, where httpx honours
    +                the client-level ``limits`` kwarg. A caller-supplied ``limits``
    +                is left untouched; otherwise the CLOSE_WAIT-safe limits are
    +                injected. The fallback-IP branch does NOT use this helper — see
    +                the ``_transport_kwargs`` note below for why.
                     """
                     kwargs = dict(httpx_kwargs or {})
                     if _pool_limits is not None and "limits" not in kwargs:
    @@ -3054,17 +3057,31 @@ def _with_limits(httpx_kwargs: Optional[dict] = None) -> dict:
                     )
                     # Keep request/update pools separate to reduce contention during
                     # polling reconnect + bot API bootstrap/delete_webhook calls.
    +                # httpx ignores the client-level `limits` kwarg when a custom
    +                # `transport` is supplied (#58790).  Unlike the proxy/direct
    +                # branches (which inject limits at the client level via
    +                # `_with_limits`), this branch MUST pass the tuned limits
    +                # directly into TelegramFallbackTransport so its inner
    +                # AsyncHTTPTransport instances honour keepalive_expiry — do not
    +                # route this through `_with_limits`, httpx would discard it.
    +                _transport_kwargs: dict = {}
    +                if _pool_limits is not None:
    +                    _transport_kwargs["limits"] = _pool_limits
                     request = HTTPXRequest(
                         **request_kwargs,
    -                    httpx_kwargs=_with_limits(
    -                        {"transport": TelegramFallbackTransport(fallback_ips)}
    -                    ),
    +                    httpx_kwargs={
    +                        "transport": TelegramFallbackTransport(
    +                            fallback_ips, **_transport_kwargs
    +                        )
    +                    },
                     )
                     get_updates_request = HTTPXRequest(
                         **request_kwargs,
    -                    httpx_kwargs=_with_limits(
    -                        {"transport": TelegramFallbackTransport(fallback_ips)}
    -                    ),
    +                    httpx_kwargs={
    +                        "transport": TelegramFallbackTransport(
    +                            fallback_ips, **_transport_kwargs
    +                        )
    +                    },
                     )
                 elif proxy_url:
                     logger.info("[%s] Proxy detected; passing explicitly to HTTPXRequest: %s", self.name, proxy_url)

  • tests/gateway/test_telegram_network.py

    @@ -354,6 +354,37 @@ def factory(**kwargs):
             assert len(seen_kwargs) == 2
             assert all("proxy" not in kwargs for kwargs in seen_kwargs)
     
    +    def test_forwards_limits_to_inner_transports(self, monkeypatch):
    +        """Verify that caller-supplied limits reach the inner
    +        AsyncHTTPTransport instances (#58790).  httpx ignores the
    +        client-level limits kwarg when a custom transport is
    +        supplied, so the limits must be forwarded via transport_kwargs.
    +        """
    +        seen_kwargs = []
    +
    +        def factory(**kwargs):
    +            seen_kwargs.append(kwargs.copy())
    +            return FakeTransport([], {})
    +
    +        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)
    +        monkeypatch.setattr(tnet.httpx, "AsyncHTTPTransport", factory)
    +
    +        custom_limits = httpx.Limits(
    +            max_connections=42,
    +            max_keepalive_connections=10,
    +            keepalive_expiry=30.0,
    +        )
    +        transport = tnet.TelegramFallbackTransport(
    +            ["149.154.167.220"], limits=custom_limits
    +        )
    +
    +        # 1 primary + 1 fallback = 2 AsyncHTTPTransport instances
    +        assert len(seen_kwargs) == 2
    +        for kw in seen_kwargs:
    +            assert "limits" in kw
    +            assert kw["limits"] is custom_limits
    +
     
     class TestFallbackTransportClose:
         @pytest.mark.asyncio