feat(index): owns_client for explicit Redis client ownership handover

pyproject.toml

@@ -169,9 +169,6 @@ exclude = '''
 [tool.pytest.ini_options]
 log_cli = true
 asyncio_mode = "auto"
-filterwarnings = [
-    "ignore:get_async_redis_connection will become async in the next major release:DeprecationWarning",
-]
 
 [tool.mypy]
 warn_unused_configs = true

redisvl/extensions/cache/base.py

@@ -4,6 +4,7 @@
 specific cache types such as LLM caches and embedding caches.
 """
 
+import asyncio
 from collections.abc import Mapping
 from typing import Any, cast
 
@@ -59,7 +60,13 @@ def __init__(
         # Initialize Redis clients
         self._async_redis_client = async_redis_client
         self._redis_client = redis_client
+        # Guards lazy async client creation, which suspends on an await and so
+        # cannot rely on a bare check-then-set. Mirrors AsyncSearchIndex._lock.
+        self._async_client_lock = asyncio.Lock()
 
+        # Caches never close a caller-supplied client and register no GC
+        # finalizer, so the index's owns_client handover has no cache
+        # equivalent by design.
         if redis_client or async_redis_client:
             self._owns_redis_client = False
         else:
@@ -128,22 +135,26 @@ async def _get_async_redis_client(self) -> AsyncRedisClient:
         Returns:
             AsyncRedisClient: An async Redis client instance.
         """
-        if not hasattr(self, "_async_redis_client") or self._async_redis_client is None:
-            client = self.redis_kwargs.get("redis_client")
-
-            if client and isinstance(client, (Redis, RedisCluster)):
-                self._async_redis_client = RedisConnectionFactory.sync_to_async_redis(
-                    client
-                )
-            else:
-                url = cast(str | None, self.redis_kwargs["redis_url"])
-                kwargs = cast(dict[str, Any], self.redis_kwargs["connection_kwargs"])
-                self._async_redis_client = (
-                    RedisConnectionFactory.get_async_redis_connection(
-                        redis_url=url, **kwargs
-                    )
-                )
-        return self._async_redis_client
+        client = getattr(self, "_async_redis_client", None)
+        if client is None:
+            async with self._async_client_lock:
+                # Double-check: another task may have created the client while
+                # this one waited on the lock or on the factory's round trip.
+                client = getattr(self, "_async_redis_client", None)
+                if client is None:
+                    provided = self.redis_kwargs.get("redis_client")
+                    if provided and isinstance(provided, (Redis, RedisCluster)):
+                        client = RedisConnectionFactory.sync_to_async_redis(provided)
+                    else:
+                        url = cast(str | None, self.redis_kwargs["redis_url"])
+                        kwargs = cast(
+                            dict[str, Any], self.redis_kwargs["connection_kwargs"]
+                        )
+                        client = await RedisConnectionFactory._get_aredis_connection(
+                            redis_url=url, **kwargs
+                        )
+                    self._async_redis_client = client
+        return client
 
     def expire(self, key: str, ttl: int | None = None) -> None:
         """Set or refresh the expiration time for a key in the cache.

redisvl/extensions/router/semantic.py

@@ -192,7 +192,11 @@ def from_existing(
                 **factory_kwargs,
             )
             index_kwargs["_client_validated"] = True
-            index_kwargs["_owns_redis_client"] = True
+            # index_kwargs wins the merge below, so only claim ownership when
+            # the caller has not already answered. Matches the setdefault in
+            # SearchIndex.from_existing, where an explicit value also wins.
+            if "owns_client" not in init_kwargs:
+                index_kwargs["owns_client"] = True
             if lib_name is not None:
                 index_kwargs["lib_name"] = lib_name
             created_redis_client = True

redisvl/index/index.py

@@ -748,7 +748,12 @@ def from_dict(cls, schema_dict: dict[str, Any], **kwargs):
         return cls(schema=schema, **kwargs)
 
     def disconnect(self):
-        """Disconnect from the Redis database."""
+        """Close the Redis client if this index owns it.
+
+        Always invalidates the cached SQL schema. When the index does not own
+        the client (see ``owns_client``), the client is left open and the
+        index remains usable.
+        """
         raise NotImplementedError("This method should be implemented by subclasses.")
 
     def key(self, id: str) -> str:
@@ -816,11 +821,11 @@ def __init__(
         redis_url: str | None = None,
         connection_kwargs: dict[str, Any] | None = None,
         validate_on_load: bool = False,
+        owns_client: bool | None = None,
         **kwargs,
     ):
-        """Initialize the RedisVL search index with a schema, Redis client
-        (or URL string with other connection args), connection_args, and other
-        kwargs.
+        """Initialize the RedisVL search index with a schema and either a Redis
+        client or a URL string with other connection kwargs.
 
         Args:
             schema (IndexSchema): Index schema object.
@@ -832,6 +837,12 @@ def __init__(
                 args.
             validate_on_load (bool, optional): Whether to validate data against schema
                 when loading. Defaults to False.
+            owns_client (Optional[bool], optional): Whether the index closes
+                the Redis client when the index is disconnected or garbage
+                collected. Defaults to None, meaning the index owns a client
+                only if it created one itself. Pass True to hand over a client
+                you created, or False to keep one the index would otherwise
+                close, in which case closing it becomes your responsibility.
         """
         if "connection_args" in kwargs:
             connection_kwargs = kwargs.pop("connection_args")
@@ -851,7 +862,18 @@ def __init__(
         self._sql_executors: dict[str, Any] = {}
 
         self._validated_client = kwargs.pop("_client_validated", False)
-        self._owns_redis_client = kwargs.pop("_owns_redis_client", redis_client is None)
+        if "_owns_redis_client" in kwargs:
+            # Underscore-prefixed kwargs are forwarded verbatim by
+            # _split_from_existing_kwargs, so this would otherwise be dropped
+            # in silence and leak the connection it used to control.
+            raise TypeError(
+                "_owns_redis_client is no longer accepted; use owns_client instead"
+            )
+        # Must be assigned before _register_client_finalizer, which gates on
+        # this flag.
+        self._owns_redis_client = (
+            redis_client is None if owns_client is None else bool(owns_client)
+        )
         self._client_finalizer = None
         # Close the owned client when this index is garbage collected. When
         # the client is created lazily, registration happens at creation time
@@ -861,10 +883,15 @@ def __init__(
     _finalizer_close_client = staticmethod(_close_owned_sync_client)
 
     def disconnect(self):
-        """Disconnect from the Redis database."""
+        """Close the Redis client if this index owns it.
+
+        Always invalidates the cached SQL schema. When the index does not own
+        the client (see ``owns_client``), the client is left open and the
+        index remains usable.
+        """
         self.invalidate_sql_schema_cache()
-        if self._owns_redis_client is False:
-            logger.info("Index does not own client, not disconnecting")
+        if not self._owns_redis_client:
+            logger.info("Index does not own its client; leaving it open")
             return
         self._detach_client_finalizer()
         if self.__redis_client:
@@ -888,6 +915,9 @@ def from_existing(
                 instantiated redis client.
             redis_url (Optional[str]): The URL of the Redis server to
                 connect to.
+            owns_client (Optional[bool], optional): Whether the index closes
+                the client. Defaults to True when this method created the
+                client from `redis_url`, and False when you supplied one.
 
         Raises:
             ValueError: If redis_url or redis_client is not provided.
@@ -923,7 +953,7 @@ def from_existing(
         schema_dict = convert_index_info_to_schema(index_info)
         schema = IndexSchema.from_dict(schema_dict)
         if created_redis_client:
-            init_kwargs["_owns_redis_client"] = True
+            init_kwargs.setdefault("owns_client", True)
             return cls(
                 schema,
                 redis_client=redis_client,
@@ -2131,6 +2161,7 @@ def __init__(
         redis_client: AsyncRedisClient | None = None,
         connection_kwargs: dict[str, Any] | None = None,
         validate_on_load: bool = False,
+        owns_client: bool | None = None,
         **kwargs,
     ):
         """Initialize the RedisVL async search index with a schema.
@@ -2145,6 +2176,12 @@ def __init__(
                 args.
             validate_on_load (bool, optional): Whether to validate data against schema
                 when loading. Defaults to False.
+            owns_client (Optional[bool], optional): Whether the index closes
+                the Redis client when the index is disconnected or garbage
+                collected. Defaults to None, meaning the index owns a client
+                only if it created one itself. Pass True to hand over a client
+                you created, or False to keep one the index would otherwise
+                close, in which case closing it becomes your responsibility.
         """
         if "redis_kwargs" in kwargs:
             connection_kwargs = kwargs.pop("redis_kwargs")
@@ -2157,15 +2194,30 @@ def __init__(
         self._validate_on_load = validate_on_load
         self._lib_name: str | None = kwargs.pop("lib_name", None)
 
-        # Store connection parameters
+        # Store connection parameters. Note the asymmetry with SearchIndex:
+        # there, _redis_client is a property that lazily creates the client,
+        # whereas here it is a plain attribute that stays None until
+        # _get_client() creates one. Read it through _get_client(), not
+        # directly.
         self._redis_client = redis_client
         self._redis_url = redis_url
         self._connection_kwargs = connection_kwargs or {}
         self._lock = asyncio.Lock()
         self._sql_executors: dict[str, Any] = {}
 
         self._validated_client = kwargs.pop("_client_validated", False)
-        self._owns_redis_client = kwargs.pop("_owns_redis_client", redis_client is None)
+        if "_owns_redis_client" in kwargs:
+            # Underscore-prefixed kwargs are forwarded verbatim by
+            # _split_from_existing_kwargs, so this would otherwise be dropped
+            # in silence and leak the connection it used to control.
+            raise TypeError(
+                "_owns_redis_client is no longer accepted; use owns_client instead"
+            )
+        # Must be assigned before _register_client_finalizer, which gates on
+        # this flag.
+        self._owns_redis_client = (
+            redis_client is None if owns_client is None else bool(owns_client)
+        )
         self._client_finalizer = None
         # Close the owned client when this index is garbage collected. When
         # the client is created lazily, registration happens at creation time
@@ -2191,6 +2243,9 @@ async def from_existing(
                 instantiated redis client.
             redis_url (Optional[str]): The URL of the Redis server to
                 connect to.
+            owns_client (Optional[bool], optional): Whether the index closes
+                the client. Defaults to True when this method created the
+                client from `redis_url`, and False when you supplied one.
         """
         if not redis_url and not redis_client:
             raise ValueError(
@@ -2231,7 +2286,7 @@ async def from_existing(
         schema_dict = convert_index_info_to_schema(index_info)
         schema = IndexSchema.from_dict(schema_dict)
         if created_redis_client:
-            init_kwargs["_owns_redis_client"] = True
+            init_kwargs.setdefault("owns_client", True)
             return cls(
                 schema,
                 redis_client=redis_client,
@@ -3316,16 +3371,27 @@ async def info(self, name: str | None = None) -> dict[str, Any]:
         return await self._info(index_name, client)
 
     async def disconnect(self):
+        """Close the Redis client if this index owns it.
+
+        Always invalidates the cached SQL schema. When the index does not own
+        the client (see ``owns_client``), the client is left open and the
+        index remains usable.
+        """
         self.invalidate_sql_schema_cache()
-        if self._owns_redis_client is False:
+        if not self._owns_redis_client:
             return
         self._detach_client_finalizer()
         if self._redis_client is not None:
             await self._redis_client.aclose()
         self._redis_client = None
 
     def disconnect_sync(self):
-        if self._redis_client is None or self._owns_redis_client is False:
+        """Close an owned Redis client from synchronous code.
+
+        For callers outside an event loop, such as ``__del__`` or a shutdown
+        hook. Honours ``owns_client`` exactly as :meth:`disconnect` does.
+        """
+        if self._redis_client is None or not self._owns_redis_client:
             return
         sync_wrapper(self.disconnect)()
 

redisvl/mcp/server.py

@@ -643,11 +643,9 @@ async def _load_effective_schema(
     @staticmethod
     def _make_index(schema: IndexSchema, client: Any) -> AsyncSearchIndex:
         """Bind an inspected schema and Redis client into an async index."""
-        index = AsyncSearchIndex(schema=schema, redis_client=client)
         # The server acquired this client explicitly during startup, so hand
         # ownership to the index for a single shutdown path.
-        index._owns_redis_client = True
-        return index
+        return AsyncSearchIndex(schema=schema, redis_client=client, owns_client=True)
 
     async def _initialize_vectorizer(
         self, binding: MCPIndexBindingConfig, schema: IndexSchema, timeout: int

redisvl/migration/async_executor.py

@@ -867,8 +867,6 @@ def _notify(step: str, detail: Optional[str] = None) -> None:
 
         try:
             client = await source_index._get_client()
-            if client is None:
-                raise ValueError("Failed to get Redis client from source index")
             aof_enabled = await self._detect_aof_enabled(client)
             disk_estimate = estimate_disk_space(plan, aof_enabled=aof_enabled)
             if disk_estimate.has_quantization:

redisvl/migration/async_planner.py

@@ -235,9 +235,7 @@ async def snapshot_source(
         prefixes = index.schema.index.prefix
         prefix_list = prefixes if isinstance(prefixes, list) else [prefixes]
 
-        client = index.client
-        if client is None:
-            raise ValueError("Failed to get Redis client from index")
+        client = await index._get_client()
 
         return SourceSnapshot(
             index_name=index_name,

redisvl/migration/async_validation.py

@@ -74,13 +74,10 @@ async def validate(
         validation.doc_count_match = source_total == target_total
 
         key_sample = plan.source.keyspace.key_sample
-        client = target_index.client
         if not key_sample:
             validation.key_sample_exists = True
-        elif client is None:
-            validation.key_sample_exists = False
-            validation.errors.append("Failed to get Redis client for key sample check")
         else:
+            client = await target_index._get_client()
             # Handle prefix change: transform key_sample to use new prefix.
             # Must match the executor's RENAME logic exactly:
             #   new_key = new_prefix + key[len(old_prefix):]
@@ -140,9 +137,7 @@ async def validate(
 
     async def _count_index_keys(self, index: AsyncSearchIndex) -> int:
         """Count keys matching the target index prefixes with SCAN."""
-        client = index.client
-        if client is None:
-            raise ValueError("Redis client is required to count index keys")
+        client = await index._get_client()
 
         prefixes = index.schema.index.prefix
         prefix_list = prefixes if isinstance(prefixes, list) else [prefixes]
@@ -181,25 +176,16 @@ async def _run_query_checks(
                 )
             )
 
-        client = target_index.client
+        client = await target_index._get_client()
         for key in query_checks.get("keys_exist", []):
-            if client is None:
-                results.append(
-                    QueryCheckResult(
-                        name=f"key:{key}",
-                        passed=False,
-                        details="Failed to get Redis client",
-                    )
-                )
-            else:
-                exists = bool(await client.exists(key))
-                results.append(
-                    QueryCheckResult(
-                        name=f"key:{key}",
-                        passed=exists,
-                        details="Key exists" if exists else "Key not found",
-                    )
+            exists = bool(await client.exists(key))
+            results.append(
+                QueryCheckResult(
+                    name=f"key:{key}",
+                    passed=exists,
+                    details="Key exists" if exists else "Key not found",
                 )
+            )
 
         return results