fix: cache clear() and migration key scans hang or crash on Redis Cluster

redisvl/extensions/cache/base.py

@@ -4,7 +4,6 @@
 specific cache types such as LLM caches and embedding caches.
 """
 
-from collections.abc import Mapping
 from typing import Any, cast
 
 from redis import Redis  # For backwards compatibility in type checking
@@ -13,6 +12,10 @@
 from redisvl.redis.connection import RedisConnectionFactory
 from redisvl.types import AsyncRedisClient, SyncRedisClient
 
+# Keys deleted per DEL when clearing. Also the SCAN count hint, so one page of
+# keys maps to one delete round-trip.
+CLEAR_BATCH_SIZE = 500
+
 
 class BaseCache:
     """Base abstract cache interface for all RedisVL caches.
@@ -182,50 +185,74 @@ async def aexpire(self, key: str, ttl: int | None = None) -> None:
             await client.expire(key, _ttl)
 
     def clear(self) -> None:
-        """Clear the cache of all keys."""
+        """Clear the cache of all keys.
+
+        Deletes every Redis key under the cache's prefix (``<name>:``) with
+        ``SCAN`` + ``DEL``. The cache object itself stays usable for future
+        writes.
+
+        Note:
+            ``SCAN`` is not a point-in-time snapshot, so this is a best-effort
+            sweep rather than an atomic flush:
+
+            - Keys written by other clients while the sweep is in progress may
+              or may not be deleted, so the cache is not guaranteed to be empty
+              when this returns. Quiesce writers first if you need that.
+            - ``SCAN`` may return the same key on more than one page. ``DEL``
+              on an already-deleted key is a no-op, so this is harmless.
+            - Deletion is not atomic across keys. If the call raises partway
+              through, some keys are already gone. The operation is idempotent,
+              so retrying is safe and converges.
+        """
         client = self._get_redis_client()
-        prefix = self._get_prefix()
-
-        # Scan for all keys with our prefix
-        cursor = 0  # Start with cursor 0
-        while True:
-            cursor_int, keys = client.scan(cursor=cursor, match=f"{prefix}*", count=100)  # type: ignore
-            if keys:
-                client.delete(*keys)
-            if cursor_int == 0:  # Redis returns 0 when scan is complete
-                break
-            # Cluster returns a dict of cursor values. We need to stop if these all
-            # come back as 0.
-            elif isinstance(cursor_int, Mapping):
-                cursor_values = list(cursor_int.values())
-                if all(v == 0 for v in cursor_values):
-                    break
-            else:
-                cursor = cursor_int  # Update cursor for next iteration
+        # scan_iter, not a hand-rolled SCAN loop: on a cluster client SCAN is
+        # broadcast to every primary and replies with a {node_name: cursor}
+        # mapping, and those cursors are node-local -- they can neither be fed
+        # back as a single cursor nor broadcast. redis-py's scan_iter already
+        # drives each primary on its own cursor via target_nodes.
+        batch: list[Any] = []
+        for key in client.scan_iter(
+            match=f"{self._get_prefix()}*", count=CLEAR_BATCH_SIZE
+        ):
+            batch.append(key)
+            if len(batch) >= CLEAR_BATCH_SIZE:
+                client.delete(*batch)
+                batch.clear()
+        if batch:
+            client.delete(*batch)
 
     async def aclear(self) -> None:
-        """Async clear the cache of all keys."""
+        """Asynchronously clear the cache of all keys.
+
+        Deletes every Redis key under the cache's prefix (``<name>:``) with
+        ``SCAN`` + ``DEL``. The cache object itself stays usable for future
+        writes.
+
+        Note:
+            ``SCAN`` is not a point-in-time snapshot, so this is a best-effort
+            sweep rather than an atomic flush:
+
+            - Keys written by other clients while the sweep is in progress may
+              or may not be deleted, so the cache is not guaranteed to be empty
+              when this returns. Quiesce writers first if you need that.
+            - ``SCAN`` may return the same key on more than one page. ``DEL``
+              on an already-deleted key is a no-op, so this is harmless.
+            - Deletion is not atomic across keys. If the call raises partway
+              through, some keys are already gone. The operation is idempotent,
+              so retrying is safe and converges.
+        """
         client = await self._get_async_redis_client()
-        prefix = self._get_prefix()
-
-        # Scan for all keys with our prefix
-        cursor = 0  # Start with cursor 0
-        while True:
-            cursor_int, keys = await client.scan(
-                cursor=cursor, match=f"{prefix}*", count=100
-            )  # type: ignore
-            if keys:
-                await client.delete(*keys)
-            if cursor_int == 0:  # Redis returns 0 when scan is complete
-                break
-            # Cluster returns a dict of cursor values. We need to stop if these all
-            # come back as 0.
-            elif isinstance(cursor_int, Mapping):
-                cursor_values = list(cursor_int.values())
-                if all(v == 0 for v in cursor_values):
-                    break
-            else:
-                cursor = cursor_int  # Update cursor for next iteration
+        # See the note in clear() on why this delegates to scan_iter.
+        batch: list[Any] = []
+        async for key in client.scan_iter(
+            match=f"{self._get_prefix()}*", count=CLEAR_BATCH_SIZE
+        ):
+            batch.append(key)
+            if len(batch) >= CLEAR_BATCH_SIZE:
+                await client.delete(*batch)
+                batch.clear()
+        if batch:
+            await client.delete(*batch)
 
     def disconnect(self) -> None:
         """Disconnect from Redis."""

redisvl/migration/async_executor.py

@@ -229,21 +229,15 @@ async def _enumerate_with_scan(
         for match_pattern in build_scan_match_patterns(
             normalized_prefixes, key_separator
         ):
-            cursor: int = 0
-            while True:
-                cursor, keys = await client.scan(
-                    cursor=cursor,
-                    match=match_pattern,
-                    count=batch_size,
-                )
-                for key in keys:
-                    key_str = key.decode() if isinstance(key, bytes) else str(key)
-                    if key_str not in seen_keys:
-                        seen_keys.add(key_str)
-                        yield key_str
-
-                if cursor == 0:
-                    break
+            # scan_iter, not a hand-rolled SCAN loop: a cluster client replies
+            # with a {node_name: cursor} mapping, which cannot be fed back as a
+            # cursor (redis-py raises DataError). scan_iter drives each primary
+            # on its own cursor.
+            async for key in client.scan_iter(match=match_pattern, count=batch_size):
+                key_str = key.decode() if isinstance(key, bytes) else str(key)
+                if key_str not in seen_keys:
+                    seen_keys.add(key_str)
+                    yield key_str
 
     async def _rename_keys(
         self,

redisvl/migration/async_planner.py

@@ -1,6 +1,7 @@
 from __future__ import annotations
 
-from typing import Any, List, Optional
+from contextlib import aclosing
+from typing import Any, AsyncGenerator, List, Optional, cast
 
 from redisvl.index import AsyncSearchIndex
 from redisvl.migration.models import (
@@ -272,21 +273,27 @@ async def _async_sample_keys(
                 match_pattern = f"{prefix}*"
             else:
                 match_pattern = f"{prefix}{key_separator}*"
-            cursor: int = 0
-            while True:
-                cursor, keys = await client.scan(
-                    cursor=cursor,
+            # See the note in the sync planner's _sample_keys on why this
+            # delegates to scan_iter rather than driving the cursor by hand.
+            # aclosing because we return mid-iteration once the sample limit is
+            # hit: an async generator abandoned that way is only closed at loop
+            # shutdown, which warns that scan_iter's aclose was never awaited.
+            # scan_iter is annotated AsyncIterator, which does not advertise
+            # aclose, but every implementation of it is an async generator.
+            scanner = cast(
+                AsyncGenerator[Any, None],
+                client.scan_iter(
                     match=match_pattern,
                     count=max(self.key_sample_limit, 10),
-                )
-                for key in keys:
+                ),
+            )
+            async with aclosing(scanner) as keys:
+                async for key in keys:
                     decoded_key = key.decode() if isinstance(key, bytes) else str(key)
                     if decoded_key not in key_sample:
                         key_sample.append(decoded_key)
                     if len(key_sample) >= self.key_sample_limit:
                         return key_sample
-                if cursor == 0:
-                    break
         return key_sample
 
     def write_plan(self, plan: MigrationPlan, plan_out: str) -> None:

redisvl/migration/async_validation.py

@@ -149,14 +149,13 @@ async def _count_index_keys(self, index: AsyncSearchIndex) -> int:
         key_separator = index.schema.index.key_separator
         seen_keys: set[str] = set()
         for match_pattern in build_scan_match_patterns(prefix_list, key_separator):
-            cursor = 0
-            while True:
-                cursor, keys = await client.scan(cursor=cursor, match=match_pattern)
-                for key in keys:
-                    key_str = key.decode() if isinstance(key, bytes) else str(key)
-                    seen_keys.add(key_str)
-                if cursor == 0:
-                    break
+            # scan_iter, not a hand-rolled SCAN loop: a cluster client replies
+            # with a {node_name: cursor} mapping, which cannot be fed back as a
+            # cursor (redis-py raises DataError). scan_iter drives each primary
+            # on its own cursor.
+            async for key in client.scan_iter(match=match_pattern):
+                key_str = key.decode() if isinstance(key, bytes) else str(key)
+                seen_keys.add(key_str)
         return len(seen_keys)
 
     async def _run_query_checks(

redisvl/migration/executor.py

@@ -386,21 +386,15 @@ def _enumerate_with_scan(
         for match_pattern in build_scan_match_patterns(
             normalized_prefixes, key_separator
         ):
-            cursor = 0
-            while True:
-                cursor, keys = client.scan(  # type: ignore[misc]
-                    cursor=cursor,
-                    match=match_pattern,
-                    count=batch_size,
-                )
-                for key in keys:
-                    key_str = key.decode() if isinstance(key, bytes) else str(key)
-                    if key_str not in seen_keys:
-                        seen_keys.add(key_str)
-                        yield key_str
-
-                if cursor == 0:
-                    break
+            # scan_iter, not a hand-rolled SCAN loop: a cluster client replies
+            # with a {node_name: cursor} mapping, which cannot be fed back as a
+            # cursor (redis-py raises DataError). scan_iter drives each primary
+            # on its own cursor.
+            for key in client.scan_iter(match=match_pattern, count=batch_size):
+                key_str = key.decode() if isinstance(key, bytes) else str(key)
+                if key_str not in seen_keys:
+                    seen_keys.add(key_str)
+                    yield key_str
 
     def _rename_keys(
         self,

redisvl/migration/planner.py

@@ -664,21 +664,20 @@ def _sample_keys(
                 # key_separator — a PREFIX of "doc" must match "doc:1",
                 # "doca:1", etc., exactly like FT.CREATE does.
                 match_pattern = f"{prefix}*"
-            cursor = 0
-            while True:
-                cursor, keys = client.scan(
-                    cursor=cursor,
-                    match=match_pattern,
-                    count=max(self.key_sample_limit, 1000),
-                )
-                for key in keys:
-                    decoded_key = key.decode() if isinstance(key, bytes) else str(key)
-                    if decoded_key not in key_sample:
-                        key_sample.append(decoded_key)
-                    if len(key_sample) >= self.key_sample_limit:
-                        return key_sample
-                if cursor == 0:
-                    break
+            # scan_iter, not a hand-rolled SCAN loop: a cluster client replies
+            # with a {node_name: cursor} mapping, which cannot be fed back as a
+            # cursor (redis-py raises DataError). scan_iter drives each primary
+            # on its own cursor. It also lets the sample limit below stop us
+            # mid-page instead of draining the whole page first.
+            for key in client.scan_iter(
+                match=match_pattern,
+                count=max(self.key_sample_limit, 1000),
+            ):
+                decoded_key = key.decode() if isinstance(key, bytes) else str(key)
+                if decoded_key not in key_sample:
+                    key_sample.append(decoded_key)
+                if len(key_sample) >= self.key_sample_limit:
+                    return key_sample
         return key_sample
 
     def _detect_possible_field_renames(

redisvl/migration/validation.py

@@ -138,17 +138,13 @@ def _count_index_keys(self, index: SearchIndex) -> int:
         key_separator = index.schema.index.key_separator
         seen_keys: set[str] = set()
         for match_pattern in build_scan_match_patterns(prefix_list, key_separator):
-            cursor = 0
-            while True:
-                cursor, keys = cast(
-                    tuple[int, list[Any]],
-                    client.scan(cursor=cursor, match=match_pattern),
-                )
-                for key in keys:
-                    key_str = key.decode() if isinstance(key, bytes) else str(key)
-                    seen_keys.add(key_str)
-                if cursor == 0:
-                    break
+            # scan_iter, not a hand-rolled SCAN loop: a cluster client replies
+            # with a {node_name: cursor} mapping, which cannot be fed back as a
+            # cursor (redis-py raises DataError). scan_iter drives each primary
+            # on its own cursor.
+            for key in client.scan_iter(match=match_pattern):
+                key_str = key.decode() if isinstance(key, bytes) else str(key)
+                seen_keys.add(key_str)
         return len(seen_keys)
 
     def _run_query_checks(

redisvl/utils/utils.py

@@ -12,9 +12,10 @@
 from warnings import warn
 
 from pydantic import BaseModel
-from redis import Redis
 from ulid import ULID
 
+from redisvl.types import SyncRedisClient
+
 T = TypeVar("T")
 
 
@@ -283,14 +284,19 @@ def norm_l2_distance(value: float) -> float:
 
 
 def scan_by_pattern(
-    redis_client: Redis,
+    redis_client: SyncRedisClient,
     pattern: str,
 ) -> Sequence[str]:
     """
     Scan the Redis database for keys matching a specific pattern.
 
+    Uses scan_iter, so this is correct for both standalone and cluster clients:
+    on a cluster, SCAN is broadcast to every primary and each one has to be
+    iterated on its own node-local cursor.
+
     Args:
-        redis (Redis): The Redis client instance.
+        redis_client (SyncRedisClient): The Redis client instance. Standalone
+            or cluster.
         pattern (str): The pattern to match keys against.
 
     Returns: