perf: reuse one aiobotocore client per credential set

Creating a client per request built a fresh aiohttp connector + SSLContext that
loads the whole CA store each time -- memray showed ~9.5M allocations in
_create_connector and repeated load_default_certs. Clients are pool-safe and
meant to be long-lived, so cache one per (endpoint, key, region) and close them
on shutdown. Big CPU/allocation win; modest (~5MB) memory.

Author: ServerSideHannes

s3proxy/app.py

@@ -22,7 +22,7 @@
 from structlog.stdlib import BoundLogger
 
 from . import concurrency
-from .client import SigV4Verifier
+from .client import SigV4Verifier, close_cached_clients
 from .config import Settings
 from .errors import S3Error, get_s3_error_code
 from .handlers import S3ProxyHandler
@@ -200,6 +200,7 @@ async def lifespan(app: FastAPI) -> AsyncIterator[None]:
         await stats_store.aclose()  # flush buffered samples before Redis closes
         await close_redis()
         await close_http_client()
+        await close_cached_clients()
         logger.info("Shutting down")
 
     return lifespan

s3proxy/client/__init__.py

@@ -1,6 +1,6 @@
 """S3 client layer - credentials, verification, and API wrapper."""
 
-from .s3 import S3Client, get_shared_session
+from .s3 import S3Client, close_cached_clients, get_shared_session
 from .types import ParsedRequest, S3Credentials
 from .verifier import CLOCK_SKEW_TOLERANCE, SigV4Verifier, _derive_signing_key
 
@@ -11,5 +11,6 @@
     "S3Credentials",
     "SigV4Verifier",
     "_derive_signing_key",
+    "close_cached_clients",
     "get_shared_session",
 ]

s3proxy/client/s3.py

@@ -2,6 +2,8 @@
 
 from __future__ import annotations
 
+import asyncio
+import contextlib
 import time
 from typing import TYPE_CHECKING, Any
 
@@ -29,6 +31,33 @@ def get_shared_session() -> aioboto3.Session:
     return _shared_session
 
 
+# Long-lived aiobotocore clients, keyed by (endpoint, access_key, region). Creating
+# a client per request builds a fresh aiohttp connector + SSLContext that loads the
+# whole CA store each time -- large native (OpenSSL) allocations invisible to
+# tracemalloc, which pile up under concurrency and dominate RSS (memray: millions
+# of allocations in _create_connector / load_default_certs). aiobotocore clients
+# are pool-safe and meant to be reused, so we cache one per credential set and
+# keep it open for the app's lifetime (closed via close_cached_clients on shutdown).
+_client_cache: dict[tuple, tuple[Any, Any]] = {}
+_client_cache_lock: asyncio.Lock | None = None
+
+
+def _cache_lock() -> asyncio.Lock:
+    global _client_cache_lock
+    if _client_cache_lock is None:
+        _client_cache_lock = asyncio.Lock()
+    return _client_cache_lock
+
+
+async def close_cached_clients() -> None:
+    """Close all cached aiobotocore clients (call on app shutdown)."""
+    async with _cache_lock():
+        for ctx, _client in _client_cache.values():
+            with contextlib.suppress(Exception):
+                await ctx.__aexit__(None, None, None)
+        _client_cache.clear()
+
+
 def _add_optional_kwargs(kwargs: dict[str, Any], **optional: Any) -> None:
     """Add non-None optional kwargs to the dict."""
     for key, value in optional.items():
@@ -63,28 +92,41 @@ def __init__(self, settings: Settings, credentials: S3Credentials):
         self._client_context = None
 
     async def __aenter__(self):
-        """Enter async context - create client from shared session."""
-        # Use shared session to avoid loading JSON service models repeatedly
-        # Each new session costs ~30-150MB for botocore service definitions
-        session = get_shared_session()
-        self._client_context = session.client(
-            "s3",
-            endpoint_url=self.settings.s3_endpoint,
-            config=self._config,
-            aws_access_key_id=self.credentials.access_key,
-            aws_secret_access_key=self.credentials.secret_key,
-            region_name=self.credentials.region,
-        )
-        self._cached_client = await self._client_context.__aenter__()
+        """Enter async context - reuse a long-lived client for these credentials."""
+        self._cached_client = await self._get_or_create_client()
         return self
 
+    async def _get_or_create_client(self):
+        key = (
+            self.settings.s3_endpoint,
+            self.credentials.access_key,
+            self.credentials.secret_key,
+            self.credentials.region,
+        )
+        cached = _client_cache.get(key)
+        if cached is not None:
+            return cached[1]
+        async with _cache_lock():
+            cached = _client_cache.get(key)  # double-checked under lock
+            if cached is not None:
+                return cached[1]
+            # Shared session avoids reloading botocore JSON models per client.
+            ctx = get_shared_session().client(
+                "s3",
+                endpoint_url=self.settings.s3_endpoint,
+                config=self._config,
+                aws_access_key_id=self.credentials.access_key,
+                aws_secret_access_key=self.credentials.secret_key,
+                region_name=self.credentials.region,
+            )
+            client = await ctx.__aenter__()
+            _client_cache[key] = (ctx, client)
+            return client
+
     async def __aexit__(self, exc_type, exc_val, exc_tb):
-        """Exit async context - clean up client."""
-        if self._client_context is not None:
-            await self._client_context.__aexit__(exc_type, exc_val, exc_tb)
-            self._cached_client = None
-            self._client_context = None
-        logger.debug("Cleaned up S3 client context")
+        """Exit - the client is cached and shared, so it stays open (closed on
+        app shutdown via close_cached_clients). Nothing to tear down per request."""
+        self._cached_client = None
 
     async def get_object(
         self,

tests/unit/test_client_reuse.py

@@ -0,0 +1,92 @@
+"""S3Client must reuse one aiobotocore client per credential set.
+
+Creating a client per request builds a fresh aiohttp connector + SSLContext that
+loads the whole CA store each time -- large native allocations (invisible to
+tracemalloc) that pile up under concurrency and drive RSS. memray showed millions
+of allocations in _create_connector / load_default_certs. Clients are pool-safe
+and meant to be long-lived, so the wrapper caches one per (endpoint, key, region)
+and reuses it. These tests pin that: same creds => one underlying client; distinct
+creds => distinct clients; and they're closed on shutdown.
+"""
+
+import pytest
+
+from s3proxy.client import s3
+from s3proxy.client.types import S3Credentials
+
+
+class _FakeCtx:
+    def __init__(self, client):
+        self._client = client
+
+    async def __aenter__(self):
+        return self._client
+
+    async def __aexit__(self, *a):
+        self._client.closed = True
+        return False
+
+
+class _FakeClient:
+    def __init__(self, n):
+        self.n = n
+        self.closed = False
+
+
+class _FakeSession:
+    def __init__(self):
+        self.calls = 0
+
+    def client(self, *a, **k):
+        self.calls += 1
+        return _FakeCtx(_FakeClient(self.calls))
+
+
+class _Settings:
+    s3_endpoint = "http://minio:9000"
+
+
+def _creds(key="AKIA1"):
+    return S3Credentials(access_key=key, secret_key="s", region="us-east-1")
+
+
+@pytest.fixture(autouse=True)
+async def _clean(monkeypatch):
+    fake = _FakeSession()
+    monkeypatch.setattr(s3, "get_shared_session", lambda: fake)
+    await s3.close_cached_clients()
+    yield fake
+    await s3.close_cached_clients()
+
+
+@pytest.mark.asyncio
+async def test_same_credentials_reuse_one_client(_clean):
+    settings = _Settings()
+    async with s3.S3Client(settings, _creds()) as c1:
+        first = c1._cached_client
+    async with s3.S3Client(settings, _creds()) as c2:
+        second = c2._cached_client
+    assert first is second  # reused
+    assert _clean.calls == 1  # session.client() called once, not per request
+    assert first.closed is False  # not torn down between requests
+
+
+@pytest.mark.asyncio
+async def test_distinct_credentials_distinct_clients(_clean):
+    settings = _Settings()
+    async with s3.S3Client(settings, _creds("AKIA1")) as c1:
+        a = c1._cached_client
+    async with s3.S3Client(settings, _creds("AKIA2")) as c2:
+        b = c2._cached_client
+    assert a is not b
+    assert _clean.calls == 2
+
+
+@pytest.mark.asyncio
+async def test_close_cached_clients_tears_down(_clean):
+    settings = _Settings()
+    async with s3.S3Client(settings, _creds()) as c:
+        client = c._cached_client
+    await s3.close_cached_clients()
+    assert client.closed is True
+    assert len(s3._client_cache) == 0