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

@@ -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:

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):