fix: do not close a caller-provided client passed to set_client()

redisvl/index/index.py

@@ -947,10 +947,22 @@ def connect(self, redis_url: str | None = None, **kwargs):
             ModuleNotFoundError: If required Redis modules are not installed.
         """
         self.invalidate_sql_schema_cache()
-        self.__redis_client = RedisConnectionFactory.get_redis_connection(
+        client = RedisConnectionFactory.get_redis_connection(
             redis_url=redis_url, **kwargs
         )
-        self._register_client_finalizer(self.__redis_client)
+        # Release the client this index owned before, if any. Registering a
+        # finalizer for the new client detaches the old client's finalizer, so
+        # without closing it here its connections would leak.
+        if self._owns_redis_client:
+            self.disconnect()
+        with self._lock:
+            self.__redis_client = client
+            # This index created the client, so it owns and must close it. The
+            # index may have been holding a caller-provided client until now
+            # (constructor injection or set_client), in which case ownership
+            # has to be taken back or nothing would ever close this one.
+            self._owns_redis_client = True
+            self._register_client_finalizer(client)
 
     @deprecated_function("set_client", "Pass connection parameters in __init__.")
     def set_client(self, redis_client: SyncRedisClient, **kwargs):
@@ -969,8 +981,19 @@ def set_client(self, redis_client: SyncRedisClient, **kwargs):
         """
         RedisConnectionFactory.validate_sync_redis(redis_client)
         self.invalidate_sql_schema_cache()
-        self.__redis_client = redis_client
-        self._register_client_finalizer(redis_client)
+        # Release the client this index created for itself, if any. Skipped when
+        # the current client is not ours, where disconnect() would do nothing
+        # beyond logging that it is not disconnecting.
+        if self._owns_redis_client:
+            self.disconnect()
+        # Swap the client and its ownership together so no other thread can
+        # observe the new client while ownership still describes the old one.
+        with self._lock:
+            self.__redis_client = redis_client
+            # The caller owns the client they passed in, so this index must
+            # never close it. Matches __init__(redis_client=...) semantics.
+            self._owns_redis_client = False
+            self._detach_client_finalizer()
         return self
 
     def _check_svs_support(self) -> None:
@@ -2171,20 +2194,46 @@ async def connect(self, redis_url: str | None = None, **kwargs):
         client = await RedisConnectionFactory._get_aredis_connection(
             redis_url=redis_url, **kwargs
         )
-        await self.set_client(client)
+        # This index created the client, so it owns and must close it.
+        await self._swap_client(client, owns=True)
 
     @deprecated_function("set_client", "Pass connection parameters in __init__.")
     async def set_client(self, redis_client: AsyncRedisClient | SyncRedisClient):
         """
         [DEPRECATED] Manually set the Redis client to use with the search index.
         This method is deprecated; please provide connection parameters in __init__.
         """
-        redis_client = await self._validate_client(redis_client)
+        # The caller owns the client they passed in, so this index must never
+        # close it. This matches __init__(redis_client=...) semantics.
+        return await self._swap_client(redis_client, owns=False)
+
+    async def _swap_client(
+        self, redis_client: AsyncRedisClient | SyncRedisClient, *, owns: bool
+    ):
+        """Replace the active client, releasing the previous one if owned.
+
+        Args:
+            redis_client: The client to start using.
+            owns: Whether this index owns the new client and is therefore
+                responsible for closing it. Callers passing their own client
+                must use False so it is never closed on their behalf.
+        """
+        validated_client = await self._validate_client(redis_client)
         self.invalidate_sql_schema_cache()
-        await self.disconnect()
+        # Release the client this index created for itself, if any. Skipped when
+        # the current client is not ours, where disconnect() is a no-op.
+        if self._owns_redis_client:
+            await self.disconnect()
+        # Swap the client and its ownership together. Setting ownership outside
+        # the lock would leave a window where another coroutine could see the
+        # new client while ownership still describes the old one, and close a
+        # caller-provided client on that basis.
         async with self._lock:
-            self._redis_client = redis_client
-        self._register_client_finalizer(redis_client)
+            self._redis_client = validated_client
+            self._owns_redis_client = owns
+            # No-op when owns is False; also detaches any finalizer left over
+            # from a previously owned client.
+            self._register_client_finalizer(validated_client)
         return self
 
     async def _get_client(self) -> AsyncRedisClient:

tests/integration/test_async_search_index.py

@@ -225,9 +225,15 @@ async def test_search_index_set_client(client, redis_url, index_schema):
             await async_index.set_client(client)
             assert isinstance(async_index.client, AsyncRedis)
 
-            if async_index.client:
-                await async_index.disconnect()
-            assert async_index.client is None
+            # The caller supplied this client, so the index does not own it.
+            # disconnect() must therefore leave it in place and open, exactly
+            # as when a client is passed to __init__. The converted async
+            # wrapper shares the caller's connection pool, so closing it here
+            # would tear down connections the caller still relies on.
+            assert async_index._owns_redis_client is False
+            await async_index.disconnect()
+            assert async_index.client is not None
+            assert await async_index.client.ping() is True
 
 
 @pytest.mark.asyncio

tests/integration/test_index_client_ownership_integration.py

@@ -0,0 +1,188 @@
+"""Client ownership semantics against a real Redis.
+
+Verifies with live connections that a client handed to an index via the
+deprecated `set_client()` is never closed by that index, while a client the
+index creates for itself (via the deprecated `connect()`) still is.
+
+See tests/unit/test_index_client_ownership.py for the mocked-client coverage.
+"""
+
+import asyncio
+import gc
+import warnings
+import weakref
+
+import pytest
+
+from redisvl.index import AsyncSearchIndex, SearchIndex
+
+fields = [{"name": "tag", "type": "tag"}, {"name": "num", "type": "numeric"}]
+
+
+def collect():
+    for _ in range(3):
+        gc.collect()
+
+
+@pytest.fixture
+def schema_dict(redis_test_name):
+    name = redis_test_name("ownership")
+    return {
+        "index": {"name": name, "prefix": name, "storage_type": "hash"},
+        "fields": fields,
+    }
+
+
+def pool_sockets_closed(sync_client) -> bool:
+    pool = sync_client.connection_pool
+    conns = list(pool._available_connections) + list(pool._in_use_connections)
+    return all(getattr(conn, "_sock", None) is None for conn in conns)
+
+
+class TestSetClientLeavesCallerClientOpen:
+    def test_sync_caller_client_usable_after_index_collected(
+        self, redis_url, schema_dict, client
+    ):
+        # Index owns a client of its own first, then the caller swaps theirs in.
+        index = SearchIndex.from_dict(schema_dict, redis_url=redis_url)
+        assert index.exists() is False
+        index_own_client = index.client
+
+        with warnings.catch_warnings():
+            warnings.simplefilter("ignore")
+            index.set_client(client)
+
+        assert index._owns_redis_client is False
+        # The index's own client was released when it was replaced.
+        assert pool_sockets_closed(index_own_client)
+
+        ref = weakref.ref(index)
+        del index
+        collect()
+
+        assert ref() is None, "index was not collected"
+        assert client.ping() is True, "caller's client was closed by the index"
+
+    def test_sync_disconnect_leaves_caller_client_open(
+        self, redis_url, schema_dict, client
+    ):
+        index = SearchIndex.from_dict(schema_dict, redis_url=redis_url)
+        with warnings.catch_warnings():
+            warnings.simplefilter("ignore")
+            index.set_client(client)
+
+        index.disconnect()
+
+        assert client.ping() is True, "disconnect() closed the caller's client"
+
+    async def test_async_caller_client_usable_after_index_collected(
+        self, redis_url, schema_dict, async_client
+    ):
+        index = AsyncSearchIndex.from_dict(schema_dict, redis_url=redis_url)
+        assert await index.exists() is False
+
+        with warnings.catch_warnings():
+            warnings.simplefilter("ignore")
+            await index.set_client(async_client)
+
+        assert index._owns_redis_client is False
+
+        ref = weakref.ref(index)
+        del index
+        collect()
+        # Give any (incorrectly) scheduled close a chance to run.
+        await asyncio.sleep(0)
+
+        assert ref() is None, "async index was not collected"
+        assert (
+            await async_client.ping() is True
+        ), "caller's async client was closed by the index"
+
+    async def test_async_disconnect_leaves_caller_client_open(
+        self, redis_url, schema_dict, async_client
+    ):
+        index = AsyncSearchIndex.from_dict(schema_dict, redis_url=redis_url)
+        with warnings.catch_warnings():
+            warnings.simplefilter("ignore")
+            await index.set_client(async_client)
+
+        await index.disconnect()
+
+        assert (
+            await async_client.ping() is True
+        ), "disconnect() closed the caller's async client"
+
+
+class TestConnectStillOwnsItsClient:
+    """The deprecated connect() creates the client, so the index must still
+    close it. This guards against over-correcting the ownership fix."""
+
+    def test_sync_connect_created_client_closed_on_collection(
+        self, redis_url, schema_dict
+    ):
+        index = SearchIndex.from_dict(schema_dict)
+        with warnings.catch_warnings():
+            warnings.simplefilter("ignore")
+            index.connect(redis_url=redis_url)
+
+        assert index._owns_redis_client is True
+        created = index.client
+        assert created is not None
+        assert created.ping() is True
+
+        del index
+        collect()
+
+        assert pool_sockets_closed(
+            created
+        ), "client created by connect() was not closed on collection"
+
+    def test_async_connect_created_client_closed_on_collection(
+        self, redis_url, schema_dict
+    ):
+        aclose_calls = []
+
+        async def build():
+            index = AsyncSearchIndex.from_dict(schema_dict)
+            with warnings.catch_warnings():
+                warnings.simplefilter("ignore")
+                await index.connect(redis_url=redis_url)
+            assert index._owns_redis_client is True
+            created = index.client
+            assert await created.ping() is True
+
+            original = created.aclose
+
+            async def recording_aclose():
+                aclose_calls.append(1)
+                await original()
+
+            created.aclose = recording_aclose
+            return index
+
+        index = asyncio.run(build())
+        ref = weakref.ref(index)
+        del index
+        collect()
+
+        assert ref() is None
+        assert aclose_calls == [
+            1
+        ], "client created by connect() was not closed exactly once"
+
+
+class TestInjectedClientAtConstruction:
+    """Baseline: __init__ already got this right. Kept so the three entry
+    points (constructor, set_client, connect) are covered together."""
+
+    def test_sync_constructor_injected_client_survives(self, schema_dict, client):
+        from redisvl.schema import IndexSchema
+
+        index = SearchIndex(IndexSchema.from_dict(schema_dict), redis_client=client)
+        assert index._owns_redis_client is False
+        assert index.exists() is False
+
+        del index
+        collect()
+
+        assert client.ping() is True