From 9d01e9d7d0baa7708c8c6313c868101904f387bb Mon Sep 17 00:00:00 2001 From: ump45nose Date: Sat, 7 Nov 2026 03:24:09 +0800 Subject: [PATCH 1/2] fix(filestorage): close owned httpx clients after remote sync AzureBlobObjectStore and VercelBlobObjectStore built a fresh httpx.Client per call and never closed it, so repeated run_remote_sync leaked FDs until cyclic GC. Add ObjectStore.close(), implement it for Azure/Vercel (honoring _owns_client), no-op for S3-family/GCS, and call it from run_remote_sync in a finally block. Fixes #6989 --- infrastructure/filestorage/contracts.py | 8 ++++ infrastructure/filestorage/operations.py | 25 ++++++----- .../filestorage/providers/_s3_shared.py | 3 ++ infrastructure/filestorage/providers/azure.py | 7 +++ infrastructure/filestorage/providers/gcs.py | 3 ++ .../filestorage/providers/vercel.py | 6 +++ tests/filestorage/test_azure_provider.py | 24 ++++++++++ .../test_community_provider_template.py | 3 ++ .../test_multi_provider_support.py | 3 ++ tests/filestorage/test_push_concurrency.py | 3 ++ tests/filestorage/test_remote_sync.py | 3 ++ .../filestorage/test_service_and_registry.py | 44 +++++++++++++++++++ tests/filestorage/test_vercel_provider.py | 23 ++++++++++ 13 files changed, 144 insertions(+), 11 deletions(-) diff --git a/infrastructure/filestorage/contracts.py b/infrastructure/filestorage/contracts.py index 0348ca6c87..702eb6c297 100644 --- a/infrastructure/filestorage/contracts.py +++ b/infrastructure/filestorage/contracts.py @@ -55,5 +55,13 @@ def put_object(self, key: str, data: bytes) -> None: def describe(self) -> str: """Short human-readable destination, for logs and CLI output.""" + def close(self) -> None: + """Release owned network resources (for example an ``httpx.Client``). + + Providers that do not own a client may no-op. Callers that build a + fresh store per operation should invoke this in a ``finally`` so + connection pools are not left to cyclic GC. + """ + __all__ = ["ObjectStore", "RemoteObject"] diff --git a/infrastructure/filestorage/operations.py b/infrastructure/filestorage/operations.py index 4cbbf8f1f0..2051a95837 100644 --- a/infrastructure/filestorage/operations.py +++ b/infrastructure/filestorage/operations.py @@ -173,18 +173,21 @@ def run_remote_sync( return None roots = syncable_roots() store = build_object_store(config) - return _owned_report( - run_sync( - store, - direction=resolved, - roots=roots, - exclusions=config.exclude, - dry_run=dry_run, - on_progress=on_progress, - # The backend, not the engine, knows how hard it can be pushed. - max_parallel_uploads=max_parallel_uploads_for_provider(config.provider), + try: + return _owned_report( + run_sync( + store, + direction=resolved, + roots=roots, + exclusions=config.exclude, + dry_run=dry_run, + on_progress=on_progress, + # The backend, not the engine, knows how hard it can be pushed. + max_parallel_uploads=max_parallel_uploads_for_provider(config.provider), + ) ) - ) + finally: + store.close() __all__ = [ diff --git a/infrastructure/filestorage/providers/_s3_shared.py b/infrastructure/filestorage/providers/_s3_shared.py index adac4f2999..4fc888a15e 100644 --- a/infrastructure/filestorage/providers/_s3_shared.py +++ b/infrastructure/filestorage/providers/_s3_shared.py @@ -46,6 +46,9 @@ def describe(self) -> str: """Human-facing ``scheme://bucket/prefix`` target for error messages.""" raise NotImplementedError + def close(self) -> None: + """S3-family stores do not own an ``httpx.Client``; nothing to release.""" + def list_objects(self, prefix: str) -> list[RemoteObject]: full_prefix = ( self._config.key_for(prefix) if prefix else f"{self._config.prefix.rstrip('/')}/" diff --git a/infrastructure/filestorage/providers/azure.py b/infrastructure/filestorage/providers/azure.py index db6f066b37..6887decb19 100644 --- a/infrastructure/filestorage/providers/azure.py +++ b/infrastructure/filestorage/providers/azure.py @@ -51,6 +51,13 @@ def __init__( ) self._token = token if token is not None else _get_azure_access_token() self._client = client if client is not None else httpx.Client(timeout=30.0) + self._owns_client = client is None + + def close(self) -> None: + """Close the owned ``httpx.Client``; no-op when the caller injected one.""" + if self._owns_client: + self._client.close() + self._owns_client = False def describe(self) -> str: return f"azure-blob://{self._account_name}/{self._config.bucket}/{self._config.prefix}" diff --git a/infrastructure/filestorage/providers/gcs.py b/infrastructure/filestorage/providers/gcs.py index 497829f10d..9912168204 100644 --- a/infrastructure/filestorage/providers/gcs.py +++ b/infrastructure/filestorage/providers/gcs.py @@ -66,6 +66,9 @@ def __init__(self, config: RemoteSyncConfig, *, session: Any | None = None) -> N self._config = config self._session = session if session is not None else _build_session() + def close(self) -> None: + """GCS uses a requests session managed elsewhere; nothing required here.""" + def describe(self) -> str: return f"gs://{self._config.bucket}/{self._config.prefix}" diff --git a/infrastructure/filestorage/providers/vercel.py b/infrastructure/filestorage/providers/vercel.py index 533dfff5d9..6a3938e297 100644 --- a/infrastructure/filestorage/providers/vercel.py +++ b/infrastructure/filestorage/providers/vercel.py @@ -76,6 +76,12 @@ def __init__( self._client = client if client is not None else httpx.Client(timeout=_TIMEOUT) self._owns_client = client is None + def close(self) -> None: + """Close the owned ``httpx.Client``; no-op when the caller injected one.""" + if self._owns_client: + self._client.close() + self._owns_client = False + def describe(self) -> str: return f"vercel-blob://{self._config.bucket}/{self._config.prefix}" diff --git a/tests/filestorage/test_azure_provider.py b/tests/filestorage/test_azure_provider.py index f101726cbc..f6bdf88376 100644 --- a/tests/filestorage/test_azure_provider.py +++ b/tests/filestorage/test_azure_provider.py @@ -181,3 +181,27 @@ def handle_request(self, request: httpx.Request) -> httpx.Response: with pytest.raises(RemoteSyncUnavailableError) as caught_put: store.put_object("test.json", b"data") assert "403" in str(caught_put.value) + + +def test_close_releases_owned_httpx_client(monkeypatch: pytest.MonkeyPatch) -> None: + """Owned clients must close; injected clients must stay open (#6989).""" + closed: list[httpx.Client] = [] + real_close = httpx.Client.close + + def _tracking_close(self: httpx.Client) -> None: + closed.append(self) + real_close(self) + + monkeypatch.setattr(httpx.Client, "close", _tracking_close) + owned = AzureBlobObjectStore(_config(), token="fake") + owned.close() + assert closed == [owned._client] + # Idempotent: a second close must not touch the already-closed client again. + owned.close() + assert closed == [owned._client] + + injected = httpx.Client(transport=_FakeTransport()) + store = AzureBlobObjectStore(_config(), token="fake", client=injected) + store.close() + assert injected not in closed + injected.close() diff --git a/tests/filestorage/test_community_provider_template.py b/tests/filestorage/test_community_provider_template.py index cfd93f279d..dbb15c1d7e 100644 --- a/tests/filestorage/test_community_provider_template.py +++ b/tests/filestorage/test_community_provider_template.py @@ -138,6 +138,9 @@ def put_object(self, key: str, data: bytes) -> None: def describe(self) -> str: return f"{_PROVIDER_NAME}://template/{self._config.prefix}" + def close(self) -> None: + return + @pytest.fixture def fake_provider() -> Iterator[dict[str, tuple[bytes, datetime]]]: diff --git a/tests/filestorage/test_multi_provider_support.py b/tests/filestorage/test_multi_provider_support.py index 7d2fd507b5..38d5104e2d 100644 --- a/tests/filestorage/test_multi_provider_support.py +++ b/tests/filestorage/test_multi_provider_support.py @@ -67,6 +67,9 @@ def put_object(self, key: str, data: bytes) -> None: def describe(self) -> str: return f"{self.label}://test" + def close(self) -> None: + return + @pytest.fixture def roots(tmp_path: Path) -> tuple[SyncRoot, ...]: diff --git a/tests/filestorage/test_push_concurrency.py b/tests/filestorage/test_push_concurrency.py index bda4c8d8ba..16c310cb20 100644 --- a/tests/filestorage/test_push_concurrency.py +++ b/tests/filestorage/test_push_concurrency.py @@ -78,6 +78,9 @@ def put_object(self, key: str, data: bytes) -> None: def describe(self) -> str: return "recording://bucket" + def close(self) -> None: + return + class FailingStore(RecordingStore): """Fails the named keys, to pin which error a caller sees.""" diff --git a/tests/filestorage/test_remote_sync.py b/tests/filestorage/test_remote_sync.py index 45a8ec4c3d..c2d693ff50 100644 --- a/tests/filestorage/test_remote_sync.py +++ b/tests/filestorage/test_remote_sync.py @@ -74,6 +74,9 @@ def put_object(self, key: str, data: bytes) -> None: def describe(self) -> str: return "fake://bucket" + def close(self) -> None: + return + @pytest.fixture def home(tmp_path: Path) -> Path: diff --git a/tests/filestorage/test_service_and_registry.py b/tests/filestorage/test_service_and_registry.py index 9d1a78c314..3b8efeb34e 100644 --- a/tests/filestorage/test_service_and_registry.py +++ b/tests/filestorage/test_service_and_registry.py @@ -67,6 +67,9 @@ def put_object(self, key: str, data: bytes) -> None: def describe(self) -> str: return "mem://t" + def close(self) -> None: + return + @pytest.fixture(autouse=True) def _restore_registry() -> Iterator[None]: @@ -334,3 +337,44 @@ def test_run_remote_sync_uses_scoped_roots(tmp_path: Path, monkeypatch: pytest.M report2 = run_remote_sync() assert report2 is not None assert report2 is not report + + +def test_run_remote_sync_closes_store(tmp_path: Path, monkeypatch: pytest.MonkeyPatch) -> None: + """``run_remote_sync`` must close the store it built (#6989).""" + from infrastructure.filestorage import operations as sync_service + + class _ClosingStore(_MemStore): + def __init__(self) -> None: + super().__init__() + self.closed = False + + def close(self) -> None: + self.closed = True + + stores: list[_ClosingStore] = [] + + def _factory(_cfg: RemoteSyncConfig) -> _ClosingStore: + store = _ClosingStore() + stores.append(store) + return store + + register_object_store("close-wiring", _factory) + try: + sessions = tmp_path / "sessions" + memory = tmp_path / "memory" + sessions.mkdir() + memory.mkdir() + (sessions / "x.jsonl").write_text("{}\n", encoding="utf-8") + roots = ( + SyncRoot(name=SyncRootName.SESSIONS, path=sessions), + SyncRoot(name=SyncRootName.MEMORY, path=memory), + ) + monkeypatch.setenv(REMOTE_SYNC_ENV, "1") + monkeypatch.setenv(REMOTE_SYNC_BUCKET_ENV, "b") + monkeypatch.setenv(REMOTE_SYNC_PROVIDER_ENV, "close-wiring") + monkeypatch.setattr(sync_service, "syncable_roots", lambda: roots) + report = run_remote_sync() + assert report is not None + assert stores and stores[0].closed is True + finally: + unregister_object_store("close-wiring") diff --git a/tests/filestorage/test_vercel_provider.py b/tests/filestorage/test_vercel_provider.py index 2dca7e8b08..0e1476cd61 100644 --- a/tests/filestorage/test_vercel_provider.py +++ b/tests/filestorage/test_vercel_provider.py @@ -327,3 +327,26 @@ def test_engine_push_restore_via_registry(tmp_path_factory: pytest.TempPathFacto assert (other_memory / "fact.md").read_text(encoding="utf-8") == "remembered\n" # Sanity: content_tag matches Vercel etags for single-part puts. assert content_tag(b'{"turn": 1}\n') == _md5(b'{"turn": 1}\n') + + +def test_close_releases_owned_httpx_client(monkeypatch: pytest.MonkeyPatch) -> None: + """Owned clients must close; injected clients must stay open (#6989).""" + closed: list[httpx.Client] = [] + real_close = httpx.Client.close + + def _tracking_close(self: httpx.Client) -> None: + closed.append(self) + real_close(self) + + monkeypatch.setattr(httpx.Client, "close", _tracking_close) + owned = VercelBlobObjectStore(_config(), token=_TOKEN) + owned.close() + assert closed == [owned._client] + owned.close() + assert closed == [owned._client] + + injected = httpx.Client(transport=_FakeTransport()) + store = VercelBlobObjectStore(_config(), token=_TOKEN, client=injected) + store.close() + assert injected not in closed + injected.close() From 438a4845b5031de40fc3965adb19da955409d33c Mon Sep 17 00:00:00 2001 From: ump45nose Date: Sat, 7 Nov 2026 05:21:35 +0800 Subject: [PATCH 2/2] fix(filestorage): close ObjectStore fakes used by sync-on-exit CI cli-runtime-3 failed because _FakeObjectStore lacked close(), so the finally in run_remote_sync raised AttributeError and masked the expected RuntimeError. Add the no-op on the fake and tolerate a missing close on legacy stores so a sync failure stays visible. --- infrastructure/filestorage/operations.py | 6 +++++- tests/cli/test_sync_on_exit.py | 3 +++ 2 files changed, 8 insertions(+), 1 deletion(-) diff --git a/infrastructure/filestorage/operations.py b/infrastructure/filestorage/operations.py index 2051a95837..6481ddb2ea 100644 --- a/infrastructure/filestorage/operations.py +++ b/infrastructure/filestorage/operations.py @@ -187,7 +187,11 @@ def run_remote_sync( ) ) finally: - store.close() + # Protocol requires close(); getattr keeps a sync failure visible if a + # legacy/community store omitted the no-op method. + closer = getattr(store, "close", None) + if closer is not None: + closer() __all__ = [ diff --git a/tests/cli/test_sync_on_exit.py b/tests/cli/test_sync_on_exit.py index 407a78adb7..8d0e9dc8ef 100644 --- a/tests/cli/test_sync_on_exit.py +++ b/tests/cli/test_sync_on_exit.py @@ -46,6 +46,9 @@ def put_object(self, key: str, data: bytes) -> None: def describe(self) -> str: return "fake://sync-on-exit" + def close(self) -> None: + return + class _FailingObjectStore(_FakeObjectStore): def list_objects(self, _prefix: str) -> list[RemoteObject]: