Track connection idleness per connection, and cover proxies and Pyodide

Review follow-up, addressing three things.

The idle window was tracked on the adapter, behind a lock, and enforced by
clearing the whole pool manager. It is now tracked on each pooled connection: a
`urllib3` pool subclass stamps the connection on the way in and closes it on the
way out if it has sat idle too long, letting `urllib3` reconnect lazily exactly
as it does for a connection it detected as dropped. That removes the lock, and
with it the risk of a child process inheriting it held across a fork, and it is
also more accurate — one session carries traces, metrics and logs, so a steady
trace stream would keep a session-level clock from ever firing for the
connection that last served a metric export.

A proxied request goes through a manager built by `proxy_manager_for`, which
`init_poolmanager` never sees, so it had neither keepalive nor recycling. The
adapter now applies both there too. The pool classes are derived from whatever
the manager already uses, rather than named outright, because a SOCKS proxy
manager brings its own.

Under Pyodide `urllib3` swaps in an Emscripten connection that talks over
`fetch()` and has no `default_socket_options` at all, which broke import. The
defaults are now read defensively and every socket option is applied only where
the platform has it, which also covers the macOS-only spelling on Linux CI.

Co-Authored-By: Claude Opus 5 <noreply@anthropic.com>

Author: adriangb

logfire/_internal/http_transport.py

@@ -11,8 +11,8 @@
 Every session Logfire creates therefore gets two measures:
 
 - **TCP keepalive**, so an idle connection keeps producing traffic and the flow is not reclaimed.
-- **An idle recycle window**, so a session unused for longer than the window drops its pooled
-  connections and reconnects rather than gambling on one that may already be dead.
+- **An idle recycle window**, so a pooled connection unused for longer than the window is closed
+  and reconnected rather than gambling on one that may already be dead.
 
 This applies only to sessions Logfire itself creates. Clients belonging to the user are
 instrumented, never reconfigured.
@@ -26,13 +26,15 @@
 
 from requests import Session
 from requests.adapters import DEFAULT_POOLBLOCK, HTTPAdapter
-from urllib3.connection import HTTPConnection
-from urllib3.connectionpool import HTTPConnectionPool, HTTPSConnectionPool
+from urllib3 import connection as urllib3_connection
+from urllib3.connectionpool import HTTPConnectionPool
+from urllib3.poolmanager import PoolManager
 
 _now = time.monotonic
+"""Indirection so tests can drive the clock without touching the one `urllib3` itself uses."""
 
 IDLE_CONNECTION_RECYCLE_SECONDS = 30
-"""Drop pooled connections when the session has gone unused for longer than this.
+"""Close a pooled connection that has sat unused for longer than this.
 
 Chosen to sit below the things that would otherwise reclaim the connection first: the default
 60 second metric export interval (`OTEL_METRIC_EXPORT_INTERVAL`), and the idle timeouts of
@@ -54,29 +56,33 @@ def keepalive_socket_options() -> list[tuple[int, int, int | bytes]]:
     """`urllib3`'s default socket options plus TCP keepalive.
 
     Building on `HTTPConnection.default_socket_options` keeps `TCP_NODELAY`, which `urllib3` sets
-    and which we have no reason to drop.
+    and which we have no reason to drop. The class is read off the module at call time because
+    `urllib3` swaps it out under Pyodide, for an Emscripten connection that talks over `fetch()`
+    and so has neither a socket nor any default options.
     """
-    options: list[tuple[int, int, int | bytes]] = [
-        *HTTPConnection.default_socket_options,
-        (socket.SOL_SOCKET, socket.SO_KEEPALIVE, 1),
-    ]
+    defaults = getattr(urllib3_connection.HTTPConnection, 'default_socket_options', None)
+    options: list[tuple[int, int, int | bytes]] = [*(defaults or [])]
 
-    def add_if_supported(name: str, value: int) -> bool:
+    def add_if_supported(level: int, name: str, value: int) -> bool:
         # Looked up by name because these constants are platform specific: referring to them
         # directly would not type check on a platform that lacks them.
         option = getattr(socket, name, None)
         if option is None:
             return False
-        options.append((socket.IPPROTO_TCP, option, value))
+        options.append((level, option, value))
         return True
 
+    if not add_if_supported(socket.SOL_SOCKET, 'SO_KEEPALIVE', 1):
+        # Nothing to keep alive, so none of the knobs below mean anything either.
+        return options
+
     # How long a connection may be idle before probing starts. Without this the system default
     # applies, which is two hours on most platforms and so useless against a NAT timeout.
     # Linux (and recent Windows) call it TCP_KEEPIDLE; macOS calls the same thing TCP_KEEPALIVE.
-    if not add_if_supported('TCP_KEEPIDLE', TCP_KEEPALIVE_IDLE_SECONDS):
-        add_if_supported('TCP_KEEPALIVE', TCP_KEEPALIVE_IDLE_SECONDS)
-    add_if_supported('TCP_KEEPINTVL', TCP_KEEPALIVE_INTERVAL_SECONDS)
-    add_if_supported('TCP_KEEPCNT', TCP_KEEPALIVE_FAILED_PROBES)
+    if not add_if_supported(socket.IPPROTO_TCP, 'TCP_KEEPIDLE', TCP_KEEPALIVE_IDLE_SECONDS):
+        add_if_supported(socket.IPPROTO_TCP, 'TCP_KEEPALIVE', TCP_KEEPALIVE_IDLE_SECONDS)
+    add_if_supported(socket.IPPROTO_TCP, 'TCP_KEEPINTVL', TCP_KEEPALIVE_INTERVAL_SECONDS)
+    add_if_supported(socket.IPPROTO_TCP, 'TCP_KEEPCNT', TCP_KEEPALIVE_FAILED_PROBES)
 
     return options
 
@@ -104,22 +110,32 @@ def _put_conn(self, conn: Any) -> Any:
 
     def _get_conn(self, timeout: float | None = None) -> Any:
         conn = super()._get_conn(timeout)
+        # Absent on a connection this pool has never handed back, i.e. a brand new one.
         idle_since = getattr(conn, '_logfire_idle_since', None)
         if idle_since is not None and _now() - idle_since > self.idle_recycle_seconds:
             conn.close()
         return conn
 
 
-def _pool_classes(idle_recycle_seconds: float) -> dict[str, type[Any]]:
-    """Pool classes bound to one recycle window.
+def _recycling_pool_class(base: type[HTTPConnectionPool], idle_recycle_seconds: float) -> type[HTTPConnectionPool]:
+    return type(
+        f'IdleRecycling{base.__name__}', (_IdleRecyclingPoolMixin, base), {'idle_recycle_seconds': idle_recycle_seconds}
+    )
+
+
+def _install_recycling_pools(manager: PoolManager, idle_recycle_seconds: float) -> None:
+    """Point a pool manager at recycling versions of the pool classes it already uses.
 
-    Built per adapter rather than passed through `connection_pool_kw`, because `urllib3` feeds
-    that mapping to its pool-key normalizer, which rejects keys it does not know.
+    Derived from whatever the manager has rather than named outright, because the classes vary:
+    a SOCKS proxy manager brings its own, and Pyodide swaps in others again. Installed on the
+    manager rather than passed through `connection_pool_kw`, because `urllib3` feeds that mapping
+    to its pool-key normalizer, which rejects keys it does not know.
     """
-    namespace = {'idle_recycle_seconds': idle_recycle_seconds}
-    return {
-        'http': type('LogfireHTTPConnectionPool', (_IdleRecyclingPoolMixin, HTTPConnectionPool), namespace),
-        'https': type('LogfireHTTPSConnectionPool', (_IdleRecyclingPoolMixin, HTTPSConnectionPool), namespace),
+    manager.pool_classes_by_scheme = {
+        # `requests` hands back a proxy manager it has already built, so the classes may be ours
+        # from an earlier call; subclassing again each time would nest them without end.
+        scheme: cls if issubclass(cls, _IdleRecyclingPoolMixin) else _recycling_pool_class(cls, idle_recycle_seconds)
+        for scheme, cls in manager.pool_classes_by_scheme.items()
     }
 
 
@@ -143,9 +159,15 @@ def init_poolmanager(
     ) -> None:
         pool_kwargs.setdefault('socket_options', keepalive_socket_options())
         super().init_poolmanager(connections, maxsize, block=block, **pool_kwargs)
-        # `getattr` because unpickling restores attributes in an order we do not control.
-        window = getattr(self, '_idle_recycle_seconds', IDLE_CONNECTION_RECYCLE_SECONDS)
-        self.poolmanager.pool_classes_by_scheme = _pool_classes(window)
+        _install_recycling_pools(self.poolmanager, self._idle_recycle_seconds)
+
+    def proxy_manager_for(self, proxy: str, **proxy_kwargs: Any) -> Any:
+        # A proxied request goes through a manager of its own, built here rather than by
+        # `init_poolmanager`, so the policy has to be applied again as each one appears.
+        proxy_kwargs.setdefault('socket_options', keepalive_socket_options())
+        manager = super().proxy_manager_for(proxy, **proxy_kwargs)
+        _install_recycling_pools(manager, self._idle_recycle_seconds)
+        return manager
 
 
 def install_connection_policy(session: Session, *, idle_recycle_seconds: float | None = None) -> None:

tests/test_http_transport.py

@@ -8,6 +8,8 @@
 import pytest
 import requests
 from urllib3.connection import HTTPConnection
+from urllib3.connectionpool import HTTPSConnectionPool
+from urllib3.poolmanager import PoolManager
 
 from logfire._internal.auth import UserToken
 from logfire._internal.client import LogfireClient
@@ -18,7 +20,8 @@
     TCP_KEEPALIVE_IDLE_SECONDS,
     LogfireHTTPAdapter,
     _IdleRecyclingPoolMixin,  # pyright: ignore[reportPrivateUsage]
-    _pool_classes,  # pyright: ignore[reportPrivateUsage]
+    _install_recycling_pools,  # pyright: ignore[reportPrivateUsage]
+    _recycling_pool_class,  # pyright: ignore[reportPrivateUsage]
     install_connection_policy,
     keepalive_socket_options,
 )
@@ -44,6 +47,43 @@ def test_keepalive_idle_option_is_set_on_platforms_that_have_one() -> None:
     assert values == [TCP_KEEPALIVE_IDLE_SECONDS]
 
 
+def test_the_macos_spelling_of_the_idle_option_is_used_when_it_is_the_only_one(
+    monkeypatch: pytest.MonkeyPatch,
+) -> None:
+    """Linux calls it TCP_KEEPIDLE, macOS calls the same thing TCP_KEEPALIVE."""
+    monkeypatch.delattr(socket, 'TCP_KEEPIDLE', raising=False)
+    monkeypatch.setattr(socket, 'TCP_KEEPALIVE', 0x10, raising=False)
+
+    assert (socket.IPPROTO_TCP, 0x10, TCP_KEEPALIVE_IDLE_SECONDS) in keepalive_socket_options()
+
+
+def test_a_platform_missing_an_option_still_gets_the_rest(monkeypatch: pytest.MonkeyPatch) -> None:
+    for name in ('TCP_KEEPIDLE', 'TCP_KEEPALIVE', 'TCP_KEEPINTVL', 'TCP_KEEPCNT'):
+        monkeypatch.delattr(socket, name, raising=False)
+
+    assert (socket.SOL_SOCKET, socket.SO_KEEPALIVE, 1) in keepalive_socket_options()
+
+
+def test_a_platform_without_keepalive_leaves_the_defaults_alone(monkeypatch: pytest.MonkeyPatch) -> None:
+    monkeypatch.delattr(socket, 'SO_KEEPALIVE', raising=False)
+
+    assert keepalive_socket_options() == list(HTTPConnection.default_socket_options)
+
+
+def test_a_connection_class_without_default_socket_options(monkeypatch: pytest.MonkeyPatch) -> None:
+    """Under Pyodide `urllib3` swaps in an Emscripten connection that has no defaults at all."""
+
+    class EmscriptenLikeConnection:
+        pass
+
+    monkeypatch.setattr('urllib3.connection.HTTPConnection', EmscriptenLikeConnection)
+    options = keepalive_socket_options()
+
+    assert (socket.SOL_SOCKET, socket.SO_KEEPALIVE, 1) in options
+    for default in HTTPConnection.default_socket_options:
+        assert default not in options
+
+
 def test_adapter_passes_socket_options_to_the_pool() -> None:
     adapter = LogfireHTTPAdapter()
     assert (socket.SOL_SOCKET, socket.SO_KEEPALIVE, 1) in adapter.poolmanager.connection_pool_kw['socket_options']
@@ -79,7 +119,7 @@ def not_dropped(conn: Any) -> bool:
 
     # Mock connections are not real sockets, so urllib3's own liveness check must stand aside.
     monkeypatch.setattr('urllib3.connectionpool.is_connection_dropped', not_dropped)
-    pool = _pool_classes(idle_recycle_seconds)['https']('example.com', maxsize=5)
+    pool: Any = _recycling_pool_class(HTTPSConnectionPool, idle_recycle_seconds)('example.com', maxsize=5)
     # urllib3 pre-fills the pool with `None` placeholders; drain them so seeded connections are
     # not discarded as "pool is full".
     while not pool.pool.empty():
@@ -148,6 +188,54 @@ def test_recycle_window_is_configurable(monkeypatch: pytest.MonkeyPatch, clock:
     conn.close.assert_called_once()
 
 
+def test_connections_urllib3_never_pooled_are_passed_through(pool: Any) -> None:
+    """`urllib3` puts `None` back after a failed request, and builds fresh connections lazily."""
+    pool._put_conn(None)
+
+    conn = pool._get_conn()
+
+    assert not hasattr(conn, '_logfire_idle_since')
+
+
+def test_recycling_pools_are_derived_from_the_classes_the_manager_already_uses() -> None:
+    """A SOCKS proxy manager brings pool classes of its own, which must not be replaced."""
+
+    class CustomPool(HTTPSConnectionPool):
+        pass
+
+    manager = PoolManager()
+    manager.pool_classes_by_scheme = {'https': CustomPool}  # pyright: ignore[reportAttributeAccessIssue]
+
+    _install_recycling_pools(manager, IDLE_CONNECTION_RECYCLE_SECONDS)
+
+    installed = manager.pool_classes_by_scheme['https']
+    assert issubclass(installed, CustomPool)
+    assert issubclass(installed, _IdleRecyclingPoolMixin)
+
+
+def test_proxied_requests_get_the_policy_too() -> None:
+    """`requests` builds a separate manager per proxy, which `init_poolmanager` never sees."""
+    adapter = LogfireHTTPAdapter(idle_recycle_seconds=13)
+
+    manager = adapter.proxy_manager_for('http://proxy.example.com')
+
+    assert (socket.SOL_SOCKET, socket.SO_KEEPALIVE, 1) in manager.connection_pool_kw['socket_options']
+    pool_class = manager.pool_classes_by_scheme['https']
+    assert issubclass(pool_class, _IdleRecyclingPoolMixin)
+    assert pool_class.idle_recycle_seconds == 13
+
+
+def test_a_reused_proxy_manager_is_not_wrapped_again() -> None:
+    """`requests` caches proxy managers, so re-applying the policy must be a no-op."""
+    adapter = LogfireHTTPAdapter()
+    proxy = 'http://proxy.example.com'
+
+    first = adapter.proxy_manager_for(proxy).pool_classes_by_scheme['https']
+    second = adapter.proxy_manager_for(proxy).pool_classes_by_scheme['https']
+
+    assert first is second
+
+
 def test_install_connection_policy_mounts_both_schemes() -> None:
     session = requests.Session()
     install_connection_policy(session)
@@ -163,6 +251,7 @@ def test_adapter_survives_pickling() -> None:
     restored = pickle.loads(pickle.dumps(LogfireHTTPAdapter(idle_recycle_seconds=7)))
 
     assert isinstance(restored, LogfireHTTPAdapter)
+    assert restored._idle_recycle_seconds == 7  # pyright: ignore[reportPrivateUsage]
     assert restored.poolmanager.pool_classes_by_scheme['https'].idle_recycle_seconds == 7