From d4b02661925adf261a49ba4a45ee20702aa94e69 Mon Sep 17 00:00:00 2001
From: Yuneng Jiang <yuneng@berri.ai>
Date: Wed, 2 Sep 2026 13:39:14 -0700
Subject: [PATCH] fix(vector_stores): release MongoDB clients built on closed
 event loops

The async client cache is keyed per event loop, and pymongo's AsyncMongoClient
holds a reference to the loop it was built on, so an entry for a closed loop
kept that client and its sockets alive for the life of the process. A script
that calls asyncio.run once per search fills the cache to its cap this way and
then stops caching entirely. Measured live against Atlas over 40 loops: 32
pinned clients and 212 open descriptors before, 1 cached client and no
monotonic descriptor growth after.
---
 litellm/llms/mongodb/common_utils.py          | 13 ++++++++++
 .../test_mongodb_transformation.py            | 24 +++++++++++++++++++
 2 files changed, 37 insertions(+)

diff --git a/litellm/llms/mongodb/common_utils.py b/litellm/llms/mongodb/common_utils.py
index 8ac02552ecb..460a2903c60 100644
--- a/litellm/llms/mongodb/common_utils.py
+++ b/litellm/llms/mongodb/common_utils.py
@@ -109,6 +109,18 @@ def get_sync_client(key: MongoClientKey, client_class: SyncClientFactory | None
     return client
 
 
+def _purge_dead_loops() -> None:
+    """The cached client holds its loop object alive, so a closed loop's entry would otherwise pin
+    that client and its sockets for the life of the process. Callers that run one loop per search
+    (``asyncio.run`` in a script) reach the cap this way and never release what is behind it."""
+    for stale in tuple(
+        cache_key
+        for cache_key, (loop_ref, _) in _async_clients.items()
+        if (cached_loop := loop_ref()) is None or cached_loop.is_closed()
+    ):
+        del _async_clients[stale]
+
+
 def get_async_client(key: MongoClientKey, client_class: AsyncClientFactory | None = None) -> "AsyncMongoClient":
     """Async clients bind to the loop that created them, so the cache is keyed per loop."""
     loop: Final = asyncio.get_running_loop()
@@ -116,6 +128,7 @@ def get_async_client(key: MongoClientKey, client_class: AsyncClientFactory | Non
     cached: Final = _async_clients.get(loop_key)
     if cached is not None and cached[0]() is loop:
         return cached[1]
+    _purge_dead_loops()
     build: Final = client_class if client_class is not None else import_async_mongo_client()
     client: Final = build(key.connection_string, **_client_kwargs(key))
     if len(_async_clients) < _MAX_CACHED_CLIENTS or loop_key in _async_clients:
diff --git a/tests/test_litellm/llms/mongodb/vector_stores/test_mongodb_transformation.py b/tests/test_litellm/llms/mongodb/vector_stores/test_mongodb_transformation.py
index 140efd53449..6d932c05065 100644
--- a/tests/test_litellm/llms/mongodb/vector_stores/test_mongodb_transformation.py
+++ b/tests/test_litellm/llms/mongodb/vector_stores/test_mongodb_transformation.py
@@ -10,6 +10,8 @@ import pytest
 
 from litellm.exceptions import BadRequestError, Timeout
 from litellm.llms.mongodb.common_utils import (
+    _MAX_CACHED_CLIENTS,
+    _async_clients,
     MongoClientKey,
     index_not_ready_error,
     missing_index_error,
@@ -655,6 +657,28 @@ class TestClientCache:
         ]
         assert stale == [], f"{len(stale)} of 20 loops were handed a client built on a closed loop"
 
+    def test_the_cache_releases_clients_built_on_closed_loops(self):
+        """pymongo's AsyncMongoClient keeps a reference to the loop it was built on, so an entry
+        for a closed loop holds that client, and its sockets, for the life of the process. A
+        script calling asyncio.run per search fills the cache to its cap that way: measured live
+        against Atlas at 32 pinned clients and 212 open descriptors after 40 loops."""
+
+        class LoopHoldingClient:
+            def __init__(self, *args, **kwargs):
+                self.loop = asyncio.get_running_loop()
+
+        key = self._key()
+
+        async def fetch():
+            return get_async_client(key, LoopHoldingClient)
+
+        for _ in range(_MAX_CACHED_CLIENTS + 8):
+            loop = asyncio.new_event_loop()
+            loop.run_until_complete(fetch())
+            loop.close()
+
+        assert len(_async_clients) == 1, f"{len(_async_clients)} closed-loop clients are still cached"
+
 
 class TestClientKeyDerivation:
     def test_no_timeout_uses_the_bounded_defaults(self):