From 57cc6e3b8e19c50f44b0a1b419087c58e78dbfc5 Mon Sep 17 00:00:00 2001
From: Vishal Bala <vishalbala.1994@gmail.com>
Date: Thu, 6 Aug 2026 19:19:12 +0200
Subject: [PATCH 1/5] fix(cache): drain every primary when clearing on Redis
 Cluster

`BaseCache.clear`/`aclear` could spin forever on Redis Cluster, leaving
the cache populated. `RedisCluster.scan` broadcasts SCAN to every primary
and replies with a `{node_name: cursor}` mapping, so `cursor_int == 0`
was never true, the `Mapping` branch only broke once every node reported
0, and the `else` that advanced the cursor was unreachable. The cursor
stayed 0 and `SCAN 0` was re-issued indefinitely.

Re-issuing `SCAN 0` accidentally makes progress when every key in the DB
matches the cache prefix: each round deletes the first page, the keyspace
shrinks, and the loop drains. That is why small-cache tests passed. The
genuine hang needs keys that do NOT match the prefix -- the normal case,
since redisvl shares a keyspace between index docs, caches and app data.
Then a `SCAN 0` page can match nothing, nothing is deleted, and the loop
makes zero progress. Measured against a real 3-primary cluster with
50k unrelated keys and 200 cache keys: 20,000 SCAN calls without
terminating, 197 of 200 cache keys orphaned.

Cursors are node-local, so the mapping cannot be handed back to
`scan(cursor=...)` (redis-py raises DataError) and a single value cannot
be broadcast. Rather than hand-roll the per-node walk, delegate to
redis-py's `scan_iter`, which already drives each primary on its own
cursor via `target_nodes`. That drops the sync/async duplication and all
six `# type: ignore`s, and leaves cluster cursor semantics to redis-py.

Deletes are batched at CLEAR_BATCH_SIZE rather than one DEL per SCAN
page. Also documents what callers actually get: SCAN is not a
point-in-time snapshot, so this is a best-effort sweep, not an atomic
flush. Those docstrings are published API reference text, since
docs/api/cache.rst autodocs SemanticCache and EmbeddingsCache with
`:inherited-members:`.

Affects `SemanticCache` and `EmbeddingsCache`, neither of which overrides
`BaseCache`.
---
 redisvl/extensions/cache/base.py              | 109 ++++---
 tests/unit/test_cache_clear_cluster_cursor.py | 296 ++++++++++++++++++
 2 files changed, 364 insertions(+), 41 deletions(-)
 create mode 100644 tests/unit/test_cache_clear_cluster_cursor.py

diff --git a/redisvl/extensions/cache/base.py b/redisvl/extensions/cache/base.py
index 8ff3566d..f14abb86 100644
--- a/redisvl/extensions/cache/base.py
+++ b/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."""
diff --git a/tests/unit/test_cache_clear_cluster_cursor.py b/tests/unit/test_cache_clear_cluster_cursor.py
new file mode 100644
index 00000000..79e053a6
--- /dev/null
+++ b/tests/unit/test_cache_clear_cluster_cursor.py
@@ -0,0 +1,296 @@
+"""Unit tests for BaseCache.clear/aclear key enumeration, especially on cluster.
+
+On a cluster client, ``scan`` is broadcast to every primary and replies with a
+``{node_name: cursor}`` mapping. Those cursors are node-local: they can neither
+be fed back as a single cursor nor broadcast to the other primaries. The clear
+loop used to leave its cursor at 0 in that case, so it re-issued ``SCAN 0``
+forever and never made progress once the first page stopped yielding matches.
+
+``clear`` now delegates enumeration to redis-py's ``scan_iter``, which drives
+each primary on its own cursor via ``target_nodes``. The fakes below bind the
+real upstream ``scan_iter`` onto themselves, so these tests exercise the actual
+library loop rather than a reimplementation of it, and assert that every
+follow-up ``SCAN`` carries the cursor the previous reply returned for that node.
+"""
+
+import pytest
+from redis.asyncio.client import Redis as AsyncRedis
+from redis.asyncio.cluster import RedisCluster as AsyncRedisCluster
+from redis.client import Redis
+from redis.cluster import RedisCluster
+
+from redisvl.extensions.cache.base import CLEAR_BATCH_SIZE, BaseCache
+from redisvl.extensions.cache.embeddings import EmbeddingsCache
+
+PREFIX = "clear_cursor_test"
+MATCH = f"{PREFIX}:*"
+
+
+class MockClusterClient:
+    """Stand-in for RedisCluster SCAN/DEL, driven by a per-node key layout.
+
+    Keys live on named primaries and each primary pages through its own keys on
+    its own cursor, which is what makes the cluster contract testable. ``scan``
+    raises if a node-local cursor is ever broadcast, and if the call count runs
+    away -- so a loop that fails to advance fails the test loudly instead of
+    hanging the suite (``pytest-timeout`` is not installed).
+    """
+
+    # Exercise the real upstream loop instead of imitating it.
+    scan_iter = RedisCluster.scan_iter
+
+    def __init__(self, keys_by_node, page=2, missing_nodes=frozenset(), max_calls=200):
+        self.keys_by_node = {n: list(k) for n, k in keys_by_node.items()}
+        self.page = page
+        self.missing_nodes = frozenset(missing_nodes)
+        self.max_calls = max_calls
+        self.scan_calls = []
+        self.deleted = []
+
+    def _page(self, node, cursor):
+        keys = self.keys_by_node[node]
+        nxt = cursor + self.page
+        return (0 if nxt >= len(keys) else nxt), keys[cursor : cursor + self.page]
+
+    def _next_scan(self, cursor, match, count, target_nodes):
+        self.scan_calls.append(
+            {"cursor": cursor, "match": match, "target_nodes": target_nodes}
+        )
+        if len(self.scan_calls) > self.max_calls:
+            raise AssertionError(
+                f"{len(self.scan_calls)} SCAN calls -- the cursor is not "
+                f"advancing: {self.scan_calls[:6]}..."
+            )
+        if target_nodes is None:
+            # Broadcast to all primaries. Only cursor 0 may be broadcast; a
+            # node-local cursor sent here would resume other nodes mid-keyspace.
+            assert cursor == 0, f"node-local cursor {cursor!r} broadcast to primaries"
+            pages = {n: self._page(n, 0) for n in self.keys_by_node}
+            return (
+                {n: c for n, (c, _) in pages.items()},
+                [k for _, ks in pages.values() for k in ks],
+            )
+        node = str(target_nodes).removeprefix("node-object:")
+        nxt, keys = self._page(node, cursor)
+        return {node: nxt}, keys
+
+    def scan(self, cursor=0, match=None, count=None, target_nodes=None, **kwargs):
+        return self._next_scan(cursor, match, count, target_nodes)
+
+    def delete(self, *keys):
+        self.deleted.extend(keys)
+        return len(keys)
+
+    def get_node(self, host=None, port=None, node_name=None):
+        if node_name in self.missing_nodes:
+            return None
+        return f"node-object:{node_name}"
+
+
+class MockAsyncClusterClient(MockClusterClient):
+    """Async variant, bound to the async cluster client's ``scan_iter``."""
+
+    scan_iter = AsyncRedisCluster.scan_iter
+
+    async def scan(self, cursor=0, match=None, count=None, target_nodes=None, **kwargs):
+        return self._next_scan(cursor, match, count, target_nodes)
+
+    async def delete(self, *keys):
+        self.deleted.extend(keys)
+        return len(keys)
+
+
+class MockStandaloneClient(MockClusterClient):
+    """Standalone client: one integer cursor, no per-node mapping."""
+
+    scan_iter = Redis.scan_iter
+
+    def _next_scan(self, cursor, match, count, target_nodes):
+        self.scan_calls.append(
+            {"cursor": cursor, "match": match, "target_nodes": target_nodes}
+        )
+        if len(self.scan_calls) > self.max_calls:
+            raise AssertionError(f"{len(self.scan_calls)} SCAN calls -- not advancing")
+        keys = self.keys_by_node["single"]
+        cursor = int(cursor)
+        nxt = cursor + self.page
+        return (0 if nxt >= len(keys) else nxt), keys[cursor : cursor + self.page]
+
+
+class MockAsyncStandaloneClient(MockStandaloneClient):
+    """Async standalone client."""
+
+    scan_iter = AsyncRedis.scan_iter
+
+    async def scan(self, cursor=0, match=None, count=None, target_nodes=None, **kwargs):
+        return self._next_scan(cursor, match, count, target_nodes)
+
+    async def delete(self, *keys):
+        self.deleted.extend(keys)
+        return len(keys)
+
+
+def _layout():
+    """Three primaries with uneven scan depth, one done after the broadcast."""
+    return {
+        "node-1": [f"{PREFIX}:a{i}" for i in range(6)],  # 3 rounds
+        "node-2": [f"{PREFIX}:b{i}" for i in range(2)],  # done on the broadcast
+        "node-3": [f"{PREFIX}:c{i}" for i in range(4)],  # 2 rounds
+    }
+
+
+def _all_keys(layout):
+    return sorted(k for ks in layout.values() for k in ks)
+
+
+def _assert_cursors_advanced(client):
+    """Every targeted SCAN must carry a non-zero, node-local cursor."""
+    targeted = [c for c in client.scan_calls if c["target_nodes"] is not None]
+    assert targeted, "no per-node continuation happened; test proves nothing"
+    assert all(
+        c["cursor"] != 0 for c in targeted
+    ), f"a continuation restarted at cursor 0: {targeted}"
+    assert all(c["match"] == MATCH for c in client.scan_calls)
+
+
+class TestClearOnCluster:
+    """clear() must drain every primary and terminate."""
+
+    def test_drains_all_primaries_with_uneven_depth(self):
+        layout = _layout()
+        client = MockClusterClient(layout)
+        cache = BaseCache(name=PREFIX, redis_client=client)
+
+        cache.clear()
+
+        assert sorted(client.deleted) == _all_keys(layout)
+        _assert_cursors_advanced(client)
+        # node-2 finished on the broadcast and is never targeted again.
+        assert not any(
+            c["target_nodes"] == "node-object:node-2" for c in client.scan_calls
+        )
+
+    def test_single_node_cluster_is_drained(self):
+        layout = {"node-1": [f"{PREFIX}:{i}" for i in range(5)]}
+        client = MockClusterClient(layout)
+
+        BaseCache(name=PREFIX, redis_client=client).clear()
+
+        assert sorted(client.deleted) == _all_keys(layout)
+        _assert_cursors_advanced(client)
+
+    def test_empty_cache_issues_no_delete(self):
+        client = MockClusterClient({"node-1": [], "node-2": []})
+
+        BaseCache(name=PREFIX, redis_client=client).clear()
+
+        assert client.deleted == []
+        assert len(client.scan_calls) == 1
+
+    def test_tolerates_duplicate_keys_across_pages(self):
+        # SCAN may return the same key more than once. DEL is idempotent, so
+        # clear() must not choke -- and nothing here counts keys.
+        dup = f"{PREFIX}:dup"
+        client = MockClusterClient({"node-1": [dup, dup, f"{PREFIX}:other", dup]})
+
+        BaseCache(name=PREFIX, redis_client=client).clear()
+
+        assert sorted(set(client.deleted)) == [dup, f"{PREFIX}:other"]
+
+    def test_deletes_are_batched(self):
+        n = CLEAR_BATCH_SIZE * 2 + 7
+        layout = {"node-1": [f"{PREFIX}:{i}" for i in range(n)]}
+        client = MockClusterClient(layout, page=CLEAR_BATCH_SIZE)
+        calls = []
+
+        real_delete = client.delete
+        client.delete = lambda *keys: (calls.append(len(keys)), real_delete(*keys))[1]
+
+        BaseCache(name=PREFIX, redis_client=client).clear()
+
+        assert sorted(client.deleted) == _all_keys(layout)
+        assert calls == [CLEAR_BATCH_SIZE, CLEAR_BATCH_SIZE, 7]
+
+
+class TestAsyncClearOnCluster:
+    """aclear() must match clear() behavior exactly."""
+
+    @pytest.mark.asyncio
+    async def test_drains_all_primaries_with_uneven_depth(self):
+        layout = _layout()
+        client = MockAsyncClusterClient(layout)
+        cache = BaseCache(name=PREFIX, async_redis_client=client)
+
+        await cache.aclear()
+
+        assert sorted(client.deleted) == _all_keys(layout)
+        _assert_cursors_advanced(client)
+
+    @pytest.mark.asyncio
+    async def test_empty_cache_issues_no_delete(self):
+        client = MockAsyncClusterClient({"node-1": [], "node-2": []})
+
+        await BaseCache(name=PREFIX, async_redis_client=client).aclear()
+
+        assert client.deleted == []
+
+    @pytest.mark.asyncio
+    async def test_tolerates_duplicate_keys_across_pages(self):
+        dup = f"{PREFIX}:dup"
+        client = MockAsyncClusterClient({"node-1": [dup, dup, f"{PREFIX}:other", dup]})
+
+        await BaseCache(name=PREFIX, async_redis_client=client).aclear()
+
+        assert sorted(set(client.deleted)) == [dup, f"{PREFIX}:other"]
+
+
+class TestClearOnStandalone:
+    """The standalone path must keep working unchanged."""
+
+    def test_advances_single_cursor(self):
+        keys = [f"{PREFIX}:{i}" for i in range(7)]
+        client = MockStandaloneClient({"single": keys})
+
+        BaseCache(name=PREFIX, redis_client=client).clear()
+
+        assert sorted(client.deleted) == sorted(keys)
+        cursors = [c["cursor"] for c in client.scan_calls]
+        assert cursors[0] in (0, "0")
+        assert [int(c) for c in cursors] == [0, 2, 4, 6]
+
+    @pytest.mark.asyncio
+    async def test_async_advances_single_cursor(self):
+        keys = [f"{PREFIX}:{i}" for i in range(7)]
+        client = MockAsyncStandaloneClient({"single": keys})
+
+        await BaseCache(name=PREFIX, async_redis_client=client).aclear()
+
+        assert sorted(client.deleted) == sorted(keys)
+        assert [int(c["cursor"]) for c in client.scan_calls] == [0, 2, 4, 6]
+
+
+class TestConcreteCachesClearOnCluster:
+    """The fix must hold through the cache classes users actually instantiate.
+
+    Also pins that each subclass's effective key prefix really is ``<name>:*``,
+    which clear() depends on and nothing else covers.
+    """
+
+    def test_embeddings_cache_drains_cluster(self):
+        layout = _layout()
+        client = MockClusterClient(layout)
+
+        EmbeddingsCache(name=PREFIX, redis_client=client).clear()
+
+        assert sorted(client.deleted) == _all_keys(layout)
+        assert all(c["match"] == MATCH for c in client.scan_calls)
+
+    @pytest.mark.asyncio
+    async def test_embeddings_cache_drains_cluster_async(self):
+        layout = _layout()
+        client = MockAsyncClusterClient(layout)
+
+        await EmbeddingsCache(name=PREFIX, async_redis_client=client).aclear()
+
+        assert sorted(client.deleted) == _all_keys(layout)
+        assert all(c["match"] == MATCH for c in client.scan_calls)

From 38988b406ef83740cfb7160cc1e48c7be5602463 Mon Sep 17 00:00:00 2001
From: Vishal Bala <vishalbala.1994@gmail.com>
Date: Thu, 6 Aug 2026 19:19:34 +0200
Subject: [PATCH 2/5] fix(migration): SCAN key enumeration crashes on Redis
 Cluster

Six migration SCAN loops fed the previous reply's cursor straight back
into `client.scan(cursor=...)`. On a cluster client that reply is a
`{node_name: cursor}` mapping, so `cursor == 0` is never true and the
second iteration passes a dict as the cursor. Verified against a real
3-primary cluster: `DataError: Invalid input of type: 'dict'`. Unlike the
cache bug this is a hard crash on the second SCAN, and it fires for any
cluster keyspace.

Reachability, most to least exposed:

  validation._count_index_keys        every validate run
  async_validation._count_index_keys  every validate run
  async_planner._async_sample_keys    count=max(limit, 10) rarely fills
                                      the limit on the first page
  planner._sample_keys                count=max(limit, 1000) usually
                                      returns early, but not when the
                                      keyspace is smaller than the limit
  executor._enumerate_with_scan       fallback paths only
  async_executor._enumerate_with_scan fallback paths only

All six are plain "enumerate keys matching a pattern", so each collapses
to `scan_iter`, which handles the per-node cursors upstream. The sample
sites gain a small bonus: a generator lets the sample limit stop us
mid-page instead of draining the page first.

Also widens `scan_by_pattern` from `Redis` to `SyncRedisClient` -- it was
already cluster-correct via `scan_iter`, but its annotation said
otherwise.

The migration test doubles define `scan` but not `scan_iter`, so they now
bind redis-py's real `scan_iter` and drive their own `scan` through the
actual library loop.
---
 redisvl/migration/async_executor.py        | 24 +++++++-----------
 redisvl/migration/async_planner.py         | 26 ++++++++-----------
 redisvl/migration/async_validation.py      | 15 ++++++-----
 redisvl/migration/executor.py              | 24 +++++++-----------
 redisvl/migration/planner.py               | 29 +++++++++++-----------
 redisvl/migration/validation.py            | 18 ++++++--------
 redisvl/utils/utils.py                     | 12 ++++++---
 tests/unit/test_async_migration_planner.py |  7 +++++-
 tests/unit/test_batch_migration.py         |  7 +++++-
 tests/unit/test_migration_planner.py       |  7 +++++-
 10 files changed, 84 insertions(+), 85 deletions(-)

diff --git a/redisvl/migration/async_executor.py b/redisvl/migration/async_executor.py
index 149ae0e9..6a7326eb 100644
--- a/redisvl/migration/async_executor.py
+++ b/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,
diff --git a/redisvl/migration/async_planner.py b/redisvl/migration/async_planner.py
index 6c75efda..60edc41a 100644
--- a/redisvl/migration/async_planner.py
+++ b/redisvl/migration/async_planner.py
@@ -272,21 +272,17 @@ 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,
-                    match=match_pattern,
-                    count=max(self.key_sample_limit, 10),
-                )
-                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
+            # See the note in the sync planner's _sample_keys on why this
+            # delegates to scan_iter rather than driving the cursor by hand.
+            async for key in client.scan_iter(
+                match=match_pattern,
+                count=max(self.key_sample_limit, 10),
+            ):
+                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 write_plan(self, plan: MigrationPlan, plan_out: str) -> None:
diff --git a/redisvl/migration/async_validation.py b/redisvl/migration/async_validation.py
index ce742a3d..02146bfb 100644
--- a/redisvl/migration/async_validation.py
+++ b/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(
diff --git a/redisvl/migration/executor.py b/redisvl/migration/executor.py
index a4f8ae3d..06b8f3f0 100644
--- a/redisvl/migration/executor.py
+++ b/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,
diff --git a/redisvl/migration/planner.py b/redisvl/migration/planner.py
index 4c09fe04..f0419aca 100644
--- a/redisvl/migration/planner.py
+++ b/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(
diff --git a/redisvl/migration/validation.py b/redisvl/migration/validation.py
index f8735a44..9f68564b 100644
--- a/redisvl/migration/validation.py
+++ b/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(
diff --git a/redisvl/utils/utils.py b/redisvl/utils/utils.py
index 85f74397..e80165cf 100644
--- a/redisvl/utils/utils.py
+++ b/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:
diff --git a/tests/unit/test_async_migration_planner.py b/tests/unit/test_async_migration_planner.py
index 93ce3d49..1591b90b 100644
--- a/tests/unit/test_async_migration_planner.py
+++ b/tests/unit/test_async_migration_planner.py
@@ -7,6 +7,7 @@
 
 import pytest
 import yaml
+from redis.asyncio.client import Redis as AsyncRedis
 
 from redisvl.migration import AsyncMigrationPlanner, MigrationPlanner
 from redisvl.schema.schema import IndexSchema
@@ -15,10 +16,14 @@
 class AsyncDummyClient:
     """Async mock Redis client for testing."""
 
+    # Bind redis-py's real scan_iter so key enumeration under test goes through
+    # the actual library loop rather than a stand-in for it.
+    scan_iter = AsyncRedis.scan_iter
+
     def __init__(self, keys):
         self.keys = keys
 
-    async def scan(self, cursor=0, match=None, count=None):
+    async def scan(self, cursor=0, match=None, count=None, _type=None, **kwargs):
         matched = []
         for key in self.keys:
             decoded_key = key.decode() if isinstance(key, bytes) else str(key)
diff --git a/tests/unit/test_batch_migration.py b/tests/unit/test_batch_migration.py
index 210c2cb4..f9dcfee4 100644
--- a/tests/unit/test_batch_migration.py
+++ b/tests/unit/test_batch_migration.py
@@ -14,6 +14,7 @@
 
 import pytest
 import yaml
+from redis.client import Redis
 
 from redisvl.migration import (
     BatchMigrationExecutor,
@@ -33,6 +34,10 @@
 class MockRedisClient:
     """Mock Redis client for batch migration tests."""
 
+    # Bind redis-py's real scan_iter so key enumeration under test goes through
+    # the actual library loop rather than a stand-in for it.
+    scan_iter = Redis.scan_iter
+
     def __init__(self, indexes: List[str] = None, keys: Dict[str, List[str]] = None):
         self.indexes = indexes or []
         self.keys = keys or {}
@@ -43,7 +48,7 @@ def execute_command(self, *args, **kwargs):
             return [idx.encode() for idx in self.indexes]
         raise NotImplementedError(f"Command not mocked: {args}")
 
-    def scan(self, cursor=0, match=None, count=None):
+    def scan(self, cursor=0, match=None, count=None, _type=None, **kwargs):
         matched = []
         all_keys = []
         for prefix_keys in self.keys.values():
diff --git a/tests/unit/test_migration_planner.py b/tests/unit/test_migration_planner.py
index b07f9df9..092c472c 100644
--- a/tests/unit/test_migration_planner.py
+++ b/tests/unit/test_migration_planner.py
@@ -2,6 +2,7 @@
 from unittest.mock import MagicMock
 
 import yaml
+from redis.client import Redis
 
 from redisvl.migration import MigrationPlanner
 from redisvl.migration.executor import _extract_prefixes_from_info
@@ -17,10 +18,14 @@
 
 
 class DummyClient:
+    # Bind redis-py's real scan_iter so key enumeration under test goes through
+    # the actual library loop rather than a stand-in for it.
+    scan_iter = Redis.scan_iter
+
     def __init__(self, keys):
         self.keys = keys
 
-    def scan(self, cursor=0, match=None, count=None):
+    def scan(self, cursor=0, match=None, count=None, _type=None, **kwargs):
         matched = []
         for key in self.keys:
             decoded_key = key.decode() if isinstance(key, bytes) else str(key)

From 410de85bcb91726f25a0697d9155fea6dac29253 Mon Sep 17 00:00:00 2001
From: Vishal Bala <vishalbala.1994@gmail.com>
Date: Thu, 6 Aug 2026 19:19:55 +0200
Subject: [PATCH 3/5] test(cluster): repair and extend EmbeddingsCache cluster
 coverage

`test_embeddings_cache_cluster_sync`/`_async` passed `text=` to
`EmbeddingsCache.set`/`aset`, whose parameter is `content`. They raised
TypeError on their first statement and had never executed, which is part
of why the cluster clear hang shipped. Both also called `clear()` with no
assertion afterward, so even once repaired they would not have caught it.

Adds a dedicated multi-page regression test, sync and async. It seeds both
unrelated keys and a cache larger than one SCAN page, because a cache
whose keys are the entire keyspace clears fine even with the bug present
-- 100 keys, what the existing test used, is exactly the size that passes
either way. It bounds the SCAN calls clear() may issue, since a
regression hangs rather than fails and pytest-timeout is not installed,
and asserts paging actually happened so the test cannot pass vacuously.

Counts keys with `scan_iter`, never KEYS or DBSIZE: those are routed to a
single node on a cluster and silently report roughly one shard's worth
(measured: 201 of 600).

The async test seeds with `aset` rather than `amset`, because `amset`
silently writes nothing on an async cluster client -- it awaits the
pipeline object returned by the queueing call, which drains the queue
before `execute()`. That is a separate bug, filed separately; this test is
about `aclear`.

Note these tests only run under `--run-cluster-tests`, which no CI
workflow passes today.
---
 .../integration/test_redis_cluster_support.py | 184 +++++++++++++++++-
 1 file changed, 180 insertions(+), 4 deletions(-)

diff --git a/tests/integration/test_redis_cluster_support.py b/tests/integration/test_redis_cluster_support.py
index fc01922c..fb5c1ea9 100644
--- a/tests/integration/test_redis_cluster_support.py
+++ b/tests/integration/test_redis_cluster_support.py
@@ -1,5 +1,7 @@
 """Tests for Redis Cluster support in RedisVL."""
 
+import asyncio
+
 import pytest
 from redis import Redis
 from redis.asyncio.cluster import RedisCluster as AsyncRedisCluster
@@ -164,7 +166,7 @@ async def test_embeddings_cache_cluster_async(redis_cluster_url, redis_test_name
 
     try:
         await cache.aset(
-            text="hey",
+            content="hey",
             model_name="test",
             embedding=[1, 2, 3],
         )
@@ -172,6 +174,7 @@ async def test_embeddings_cache_cluster_async(redis_cluster_url, redis_test_name
         assert result is not None
         assert result["embedding"] == [1, 2, 3]
         await cache.aclear()
+        assert await cache.aget("hey", "test") is None
     finally:
         # Manually close the cluster client to prevent connection leaks
         await cluster_client.aclose()
@@ -187,7 +190,7 @@ def test_embeddings_cache_cluster_sync(redis_cluster_url, redis_test_name):
 
     for i in range(100):
         cache.set(
-            text=f"hey_{i}",
+            content=f"hey_{i}",
             model_name="test",
             embedding=[1, 2, 3],
         )
@@ -195,11 +198,12 @@ def test_embeddings_cache_cluster_sync(redis_cluster_url, redis_test_name):
     assert result is not None
     assert result["embedding"] == [1, 2, 3]
     cache.clear()
+    assert cache.get("hey_0", "test") is None
 
     cache.mset(
         [
-            {"text": "hey_0", "model_name": "test", "embedding": [1, 2, 3]},
-            {"text": "hey_1", "model_name": "test", "embedding": [1, 2, 3]},
+            {"content": "hey_0", "model_name": "test", "embedding": [1, 2, 3]},
+            {"content": "hey_1", "model_name": "test", "embedding": [1, 2, 3]},
         ]
     )
     result = cache.mget(["hey_0", "hey_1"], "test")
@@ -208,6 +212,7 @@ def test_embeddings_cache_cluster_sync(redis_cluster_url, redis_test_name):
     assert result[0]["embedding"] == [1, 2, 3]
     assert result[1]["embedding"] == [1, 2, 3]
     cache.clear()
+    assert cache.mget(["hey_0", "hey_1"], "test") == [None, None]
 
 
 @pytest.mark.requires_cluster
@@ -247,3 +252,174 @@ def test_semantic_router_cluster_client(
 
     if router._index and router._index.exists():
         router._index.delete(drop=True)
+
+
+# =============================================================================
+# BaseCache.clear/aclear on a cluster keyspace larger than one SCAN page.
+#
+# On a cluster, SCAN is broadcast to all primaries and replies with a
+# {node_name: cursor} mapping of node-local cursors. The clear loop used to
+# leave its cursor at 0, so it re-issued SCAN 0 forever. It accidentally made
+# progress when every key in the DB matched the cache prefix -- deleting the
+# first page shrank the keyspace -- which is why small-cache tests passed. The
+# genuine hang needs unrelated keys in the DB, the normal case for redisvl,
+# where index docs and caches share a keyspace: then a SCAN 0 page can match
+# nothing, nothing gets deleted, and the loop spins with zero progress forever.
+#
+# These tests therefore seed BOTH unrelated keys and a multi-page cache, and
+# bound the SCAN count during clear() so a regression fails loudly instead of
+# hanging the suite (pytest-timeout is not installed).
+#
+# Note: count keys with scan_iter, never KEYS or DBSIZE. Those are routed to a
+# single node on a cluster, so they silently report roughly one shard's worth.
+# =============================================================================
+
+CLEAR_NOISE_KEYS = 2000
+CLEAR_CACHE_KEYS = 600
+MAX_CLEAR_SCANS = 400
+
+
+def _count_keys(client, pattern):
+    return sum(1 for _ in client.scan_iter(match=pattern, count=500))
+
+
+async def _acount_keys(client, pattern):
+    total = 0
+    async for _ in client.scan_iter(match=pattern, count=500):
+        total += 1
+    return total
+
+
+def _drop_keys(client, pattern):
+    keys = list(client.scan_iter(match=pattern, count=500))
+    if keys:
+        client.delete(*keys)
+
+
+async def _adrop_keys(client, pattern):
+    keys = [k async for k in client.scan_iter(match=pattern, count=500)]
+    if keys:
+        await client.delete(*keys)
+
+
+@pytest.mark.requires_cluster
+def test_embeddings_cache_clear_multipage_cluster(redis_cluster_url, redis_test_name):
+    """clear() empties a multi-page cache on a cluster and spares other keys."""
+    cluster_client = RedisCluster.from_url(redis_cluster_url)
+    name = redis_test_name("clear_multipage")
+    noise_prefix = redis_test_name("clear_noise")
+    cache = EmbeddingsCache(name=name, redis_client=cluster_client)
+
+    try:
+        # Unrelated keys: without these the buggy loop accidentally terminates.
+        pipe = cluster_client.pipeline()
+        for i in range(CLEAR_NOISE_KEYS):
+            pipe.set(f"{noise_prefix}:{i}", "keep")
+        pipe.execute()
+
+        cache.mset(
+            [
+                {
+                    "content": f"content-{i}",
+                    "model_name": "test",
+                    "embedding": [0.1, 0.2, 0.3],
+                }
+                for i in range(CLEAR_CACHE_KEYS)
+            ]
+        )
+        assert _count_keys(cluster_client, f"{name}:*") == CLEAR_CACHE_KEYS
+
+        # Bound the SCAN calls clear() itself issues: the pre-fix loop never
+        # terminates, and pytest-timeout is not installed.
+        real_scan = cluster_client.scan
+        calls: list = []
+
+        def counting_scan(*args, **kwargs):
+            calls.append(kwargs.get("cursor", args[0] if args else None))
+            if len(calls) > MAX_CLEAR_SCANS:
+                raise AssertionError(
+                    f"clear() issued {len(calls)} SCAN calls without finishing; "
+                    f"first cursors: {calls[:8]}"
+                )
+            return real_scan(*args, **kwargs)
+
+        cluster_client.scan = counting_scan
+        try:
+            cache.clear()
+        finally:
+            cluster_client.scan = real_scan
+
+        # The real-world bug: keys past the first page survived.
+        assert _count_keys(cluster_client, f"{name}:*") == 0
+        # clear() must not touch anything outside its own prefix.
+        assert _count_keys(cluster_client, f"{noise_prefix}:*") == CLEAR_NOISE_KEYS
+        # Guard against a vacuous pass: paging must actually have happened.
+        assert len(calls) > 1, "one SCAN sufficed; raise CLEAR_CACHE_KEYS"
+    finally:
+        _drop_keys(cluster_client, f"{name}:*")
+        _drop_keys(cluster_client, f"{noise_prefix}:*")
+        cluster_client.close()
+
+
+@pytest.mark.requires_cluster
+@pytest.mark.asyncio
+async def test_embeddings_cache_aclear_multipage_cluster(
+    redis_cluster_url, redis_test_name
+):
+    """aclear() must match clear(): aclear is a separate code path."""
+    cluster_client = RedisConnectionFactory.get_async_redis_cluster_connection(
+        redis_cluster_url
+    )
+    name = redis_test_name("aclear_multipage")
+    noise_prefix = redis_test_name("aclear_noise")
+    cache = EmbeddingsCache(name=name, async_redis_client=cluster_client)
+
+    try:
+        pipe = cluster_client.pipeline()
+        for i in range(CLEAR_NOISE_KEYS):
+            pipe.set(f"{noise_prefix}:{i}", "keep")
+        await pipe.execute()
+
+        # Seeded with aset, not amset: amset silently writes nothing on an
+        # async cluster client (it awaits the pipeline returned by the queueing
+        # call, which drains the queue). That is a separate bug; this test is
+        # about aclear, so don't let it depend on the broken path.
+        await asyncio.gather(
+            *(
+                cache.aset(
+                    content=f"content-{i}",
+                    model_name="test",
+                    embedding=[0.1, 0.2, 0.3],
+                )
+                for i in range(CLEAR_CACHE_KEYS)
+            )
+        )
+        assert await _acount_keys(cluster_client, f"{name}:*") == CLEAR_CACHE_KEYS
+
+        real_scan = cluster_client.scan
+        calls: list = []
+
+        async def counting_scan(*args, **kwargs):
+            calls.append(kwargs.get("cursor", args[0] if args else None))
+            if len(calls) > MAX_CLEAR_SCANS:
+                raise AssertionError(
+                    f"aclear() issued {len(calls)} SCAN calls without finishing; "
+                    f"first cursors: {calls[:8]}"
+                )
+            return await real_scan(*args, **kwargs)
+
+        cluster_client.scan = counting_scan
+        try:
+            await cache.aclear()
+        finally:
+            cluster_client.scan = real_scan
+
+        assert await _acount_keys(cluster_client, f"{name}:*") == 0
+        assert (
+            await _acount_keys(cluster_client, f"{noise_prefix}:*") == CLEAR_NOISE_KEYS
+        )
+        assert len(calls) > 1, "one SCAN sufficed; raise CLEAR_CACHE_KEYS"
+    finally:
+        await _adrop_keys(cluster_client, f"{name}:*")
+        await _adrop_keys(cluster_client, f"{noise_prefix}:*")
+        await cluster_client.aclose()

From 0cea98e1daaaf132eb0e809daafcf2bbc9a3061a Mon Sep 17 00:00:00 2001
From: Vishal Bala <vishalbala.1994@gmail.com>
Date: Fri, 7 Aug 2026 10:14:55 +0200
Subject: [PATCH 4/5] fix(migration): close scan_iter when key sampling stops
 early

`_async_sample_keys` returns from inside `async for` once the sample limit
is reached. An async generator abandoned that way is not closed until loop
shutdown, which surfaced as `RuntimeWarning: coroutine method 'aclose' of
'AsyncScanCommands.scan_iter' was never awaited` during the migration
tests. Introduced when the hand-rolled SCAN loop became `scan_iter` --
the old `while` loop held no generator.

Wraps the iteration in `contextlib.aclosing`. The sync planner has the
same early return, but a plain generator is closed deterministically by
refcounting on CPython and emits no warning, so it is left alone.
---
 redisvl/migration/async_planner.py | 31 ++++++++++++++++++++----------
 1 file changed, 21 insertions(+), 10 deletions(-)

diff --git a/redisvl/migration/async_planner.py b/redisvl/migration/async_planner.py
index 60edc41a..7bd9a154 100644
--- a/redisvl/migration/async_planner.py
+++ b/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 (
@@ -274,15 +275,25 @@ async def _async_sample_keys(
                 match_pattern = f"{prefix}{key_separator}*"
             # See the note in the sync planner's _sample_keys on why this
             # delegates to scan_iter rather than driving the cursor by hand.
-            async for key in client.scan_iter(
-                match=match_pattern,
-                count=max(self.key_sample_limit, 10),
-            ):
-                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
+            # 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),
+                ),
+            )
+            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
         return key_sample
 
     def write_plan(self, plan: MigrationPlan, plan_out: str) -> None:

From efbd3d073edba930410909b7c7f0b7e011f31ca5 Mon Sep 17 00:00:00 2001
From: Vishal Bala <vishalbala.1994@gmail.com>
Date: Fri, 7 Aug 2026 10:15:01 +0200
Subject: [PATCH 5/5] test: cut cache clear tests to what is load-bearing,
 cover migration

Mutation-tested the new suite. Of 12 cache clear tests only 3 killed
anything no other test killed, and two killed nothing at all. Now that
the cursor walk lives in redis-py rather than in `clear`, most of what
those tests asserted was upstream's behavior.

Cut 12 tests to 5, with a strictly larger kill set:

- Dropped the single-node, duplicate-key and concrete-EmbeddingsCache
  cases: kill sets were subsets of, or identical to, the multi-primary
  drain test. The duplicate-key test also asserted `sorted(set(...))`,
  discarding the duplication it claimed to check.
- Dropped the standalone cursor-arithmetic tests. They pinned
  `Redis.scan_iter`'s internals, down to its habit of seeding the cursor
  with the string "0". Standalone clear/aclear is covered end to end
  against a real Redis by tests/integration/test_llmcache.py.
- Dropped the `cursor != 0` assertion on continuations. With `scan_iter`
  driving, no version of our 9-line `clear` can violate it.
- Kept `test_deletes_are_batched`: it is the only thing standing between
  us and `client.delete(*list(client.scan_iter(...)))`, which would OOM
  on a large cache. Now patches CLEAR_BATCH_SIZE instead of depending on
  its value, so tuning the constant does not touch the test.
- Fixed `test_empty_cache_issues_no_delete`, which previously asserted
  nothing that could fail. The fake now rejects a zero-argument DEL, the
  way real Redis does ("wrong number of arguments for 'del'"), making
  these two the only guard on dropping the `if batch:` flush guard.

Adds tests/unit/test_migration_cluster_scan.py. Reverting all six
migration modules left 65/65 existing tests green -- that fix had no
regression coverage at all. The fake replies with per-node cursors and
raises DataError on a dict cursor, so the old loop fails there exactly as
it fails against a real cluster. Covers the four always-reachable sites;
the two executor sites sit behind index-info mocking and are the same
one-line pattern, so they are left to integration.

Also tightens the three migration test doubles: `_type=None` alone is
what upstream passes, so `**kwargs` is dropped and an unexpected future
kwarg now fails loudly instead of being swallowed.
---
 tests/unit/test_async_migration_planner.py    |   2 +-
 tests/unit/test_batch_migration.py            |   2 +-
 tests/unit/test_cache_clear_cluster_cursor.py | 256 +++++-------------
 tests/unit/test_migration_cluster_scan.py     | 146 ++++++++++
 tests/unit/test_migration_planner.py          |   2 +-
 5 files changed, 222 insertions(+), 186 deletions(-)
 create mode 100644 tests/unit/test_migration_cluster_scan.py

diff --git a/tests/unit/test_async_migration_planner.py b/tests/unit/test_async_migration_planner.py
index 1591b90b..005e4f65 100644
--- a/tests/unit/test_async_migration_planner.py
+++ b/tests/unit/test_async_migration_planner.py
@@ -23,7 +23,7 @@ class AsyncDummyClient:
     def __init__(self, keys):
         self.keys = keys
 
-    async def scan(self, cursor=0, match=None, count=None, _type=None, **kwargs):
+    async def scan(self, cursor=0, match=None, count=None, _type=None):
         matched = []
         for key in self.keys:
             decoded_key = key.decode() if isinstance(key, bytes) else str(key)
diff --git a/tests/unit/test_batch_migration.py b/tests/unit/test_batch_migration.py
index f9dcfee4..3ee2dcef 100644
--- a/tests/unit/test_batch_migration.py
+++ b/tests/unit/test_batch_migration.py
@@ -48,7 +48,7 @@ def execute_command(self, *args, **kwargs):
             return [idx.encode() for idx in self.indexes]
         raise NotImplementedError(f"Command not mocked: {args}")
 
-    def scan(self, cursor=0, match=None, count=None, _type=None, **kwargs):
+    def scan(self, cursor=0, match=None, count=None, _type=None):
         matched = []
         all_keys = []
         for prefix_keys in self.keys.values():
diff --git a/tests/unit/test_cache_clear_cluster_cursor.py b/tests/unit/test_cache_clear_cluster_cursor.py
index 79e053a6..1dba05fa 100644
--- a/tests/unit/test_cache_clear_cluster_cursor.py
+++ b/tests/unit/test_cache_clear_cluster_cursor.py
@@ -1,26 +1,29 @@
-"""Unit tests for BaseCache.clear/aclear key enumeration, especially on cluster.
+"""Unit tests for BaseCache.clear/aclear enumeration on Redis Cluster.
 
 On a cluster client, ``scan`` is broadcast to every primary and replies with a
-``{node_name: cursor}`` mapping. Those cursors are node-local: they can neither
-be fed back as a single cursor nor broadcast to the other primaries. The clear
-loop used to leave its cursor at 0 in that case, so it re-issued ``SCAN 0``
-forever and never made progress once the first page stopped yielding matches.
-
-``clear`` now delegates enumeration to redis-py's ``scan_iter``, which drives
-each primary on its own cursor via ``target_nodes``. The fakes below bind the
-real upstream ``scan_iter`` onto themselves, so these tests exercise the actual
-library loop rather than a reimplementation of it, and assert that every
-follow-up ``SCAN`` carries the cursor the previous reply returned for that node.
+``{node_name: cursor}`` mapping of node-local cursors. The clear loop used to
+leave its cursor at 0 in that case, so it re-issued ``SCAN 0`` forever and made
+no progress once the first page stopped yielding matches.
+
+``clear`` now delegates enumeration to redis-py's ``scan_iter``, so the cursor
+walk itself is upstream's code and not worth re-asserting here. What these tests
+pin is our part: every primary gets drained, the match pattern stays scoped to
+the cache prefix, and deletes are batched rather than issued per page or all at
+once. The fakes bind the real upstream ``scan_iter`` so the drain runs through
+the actual library loop.
+
+Standalone clear/aclear is covered end-to-end against a real Redis by
+tests/integration/test_llmcache.py; the cluster path is covered against a real
+cluster by tests/integration/test_redis_cluster_support.py, which only runs
+under --run-cluster-tests.
 """
 
 import pytest
-from redis.asyncio.client import Redis as AsyncRedis
 from redis.asyncio.cluster import RedisCluster as AsyncRedisCluster
-from redis.client import Redis
 from redis.cluster import RedisCluster
 
-from redisvl.extensions.cache.base import CLEAR_BATCH_SIZE, BaseCache
-from redisvl.extensions.cache.embeddings import EmbeddingsCache
+from redisvl.extensions.cache import base as cache_base
+from redisvl.extensions.cache.base import BaseCache
 
 PREFIX = "clear_cursor_test"
 MATCH = f"{PREFIX}:*"
@@ -30,29 +33,35 @@ class MockClusterClient:
     """Stand-in for RedisCluster SCAN/DEL, driven by a per-node key layout.
 
     Keys live on named primaries and each primary pages through its own keys on
-    its own cursor, which is what makes the cluster contract testable. ``scan``
-    raises if a node-local cursor is ever broadcast, and if the call count runs
-    away -- so a loop that fails to advance fails the test loudly instead of
-    hanging the suite (``pytest-timeout`` is not installed).
+    its own cursor, which is what makes the cluster contract testable.
+
+    Two built-in guards, so failures are loud rather than silent:
+    ``scan`` raises if a node-local cursor is ever broadcast or if the call
+    count runs away (a non-advancing cursor would otherwise hang the suite --
+    pytest-timeout is not installed), and ``delete`` raises on a zero-argument
+    call, which real Redis rejects with "wrong number of arguments".
     """
 
     # Exercise the real upstream loop instead of imitating it.
     scan_iter = RedisCluster.scan_iter
 
-    def __init__(self, keys_by_node, page=2, missing_nodes=frozenset(), max_calls=200):
+    def __init__(self, keys_by_node, page=2, max_calls=200):
         self.keys_by_node = {n: list(k) for n, k in keys_by_node.items()}
         self.page = page
-        self.missing_nodes = frozenset(missing_nodes)
         self.max_calls = max_calls
         self.scan_calls = []
-        self.deleted = []
+        self.delete_batches = []
+
+    @property
+    def deleted(self):
+        return [key for batch in self.delete_batches for key in batch]
 
     def _page(self, node, cursor):
         keys = self.keys_by_node[node]
         nxt = cursor + self.page
         return (0 if nxt >= len(keys) else nxt), keys[cursor : cursor + self.page]
 
-    def _next_scan(self, cursor, match, count, target_nodes):
+    def _next_scan(self, cursor, match, target_nodes):
         self.scan_calls.append(
             {"cursor": cursor, "match": match, "target_nodes": target_nodes}
         )
@@ -74,16 +83,19 @@ def _next_scan(self, cursor, match, count, target_nodes):
         nxt, keys = self._page(node, cursor)
         return {node: nxt}, keys
 
-    def scan(self, cursor=0, match=None, count=None, target_nodes=None, **kwargs):
-        return self._next_scan(cursor, match, count, target_nodes)
+    def _record_delete(self, keys):
+        if not keys:
+            raise AssertionError("DEL issued with no keys; real Redis rejects this")
+        self.delete_batches.append(list(keys))
+        return len(keys)
+
+    def scan(self, cursor=0, match=None, count=None, target_nodes=None, _type=None):
+        return self._next_scan(cursor, match, target_nodes)
 
     def delete(self, *keys):
-        self.deleted.extend(keys)
-        return len(keys)
+        return self._record_delete(keys)
 
     def get_node(self, host=None, port=None, node_name=None):
-        if node_name in self.missing_nodes:
-            return None
         return f"node-object:{node_name}"
 
 
@@ -92,42 +104,13 @@ class MockAsyncClusterClient(MockClusterClient):
 
     scan_iter = AsyncRedisCluster.scan_iter
 
-    async def scan(self, cursor=0, match=None, count=None, target_nodes=None, **kwargs):
-        return self._next_scan(cursor, match, count, target_nodes)
-
-    async def delete(self, *keys):
-        self.deleted.extend(keys)
-        return len(keys)
-
-
-class MockStandaloneClient(MockClusterClient):
-    """Standalone client: one integer cursor, no per-node mapping."""
-
-    scan_iter = Redis.scan_iter
-
-    def _next_scan(self, cursor, match, count, target_nodes):
-        self.scan_calls.append(
-            {"cursor": cursor, "match": match, "target_nodes": target_nodes}
-        )
-        if len(self.scan_calls) > self.max_calls:
-            raise AssertionError(f"{len(self.scan_calls)} SCAN calls -- not advancing")
-        keys = self.keys_by_node["single"]
-        cursor = int(cursor)
-        nxt = cursor + self.page
-        return (0 if nxt >= len(keys) else nxt), keys[cursor : cursor + self.page]
-
-
-class MockAsyncStandaloneClient(MockStandaloneClient):
-    """Async standalone client."""
-
-    scan_iter = AsyncRedis.scan_iter
-
-    async def scan(self, cursor=0, match=None, count=None, target_nodes=None, **kwargs):
-        return self._next_scan(cursor, match, count, target_nodes)
+    async def scan(
+        self, cursor=0, match=None, count=None, target_nodes=None, _type=None
+    ):
+        return self._next_scan(cursor, match, target_nodes)
 
     async def delete(self, *keys):
-        self.deleted.extend(keys)
-        return len(keys)
+        return self._record_delete(keys)
 
 
 def _layout():
@@ -139,92 +122,60 @@ def _layout():
     }
 
 
-def _all_keys(layout):
-    return sorted(k for ks in layout.values() for k in ks)
-
-
-def _assert_cursors_advanced(client):
-    """Every targeted SCAN must carry a non-zero, node-local cursor."""
-    targeted = [c for c in client.scan_calls if c["target_nodes"] is not None]
-    assert targeted, "no per-node continuation happened; test proves nothing"
-    assert all(
-        c["cursor"] != 0 for c in targeted
-    ), f"a continuation restarted at cursor 0: {targeted}"
+def _assert_drained(client, layout):
+    """Every primary emptied, nothing scanned outside the cache prefix."""
+    assert sorted(client.deleted) == sorted(k for ks in layout.values() for k in ks)
+    # Scoping the match pattern is what keeps clear() from wiping the whole DB.
     assert all(c["match"] == MATCH for c in client.scan_calls)
+    # Anti-vacuity: a single broadcast page would prove nothing about paging.
+    assert any(
+        c["target_nodes"] is not None for c in client.scan_calls
+    ), "no per-node continuation happened; test proves nothing"
+    # node-2 finished on the broadcast and must not be revisited.
+    assert not any(c["target_nodes"] == "node-object:node-2" for c in client.scan_calls)
 
 
 class TestClearOnCluster:
-    """clear() must drain every primary and terminate."""
-
-    def test_drains_all_primaries_with_uneven_depth(self):
+    def test_drains_every_primary(self):
         layout = _layout()
         client = MockClusterClient(layout)
-        cache = BaseCache(name=PREFIX, redis_client=client)
-
-        cache.clear()
-
-        assert sorted(client.deleted) == _all_keys(layout)
-        _assert_cursors_advanced(client)
-        # node-2 finished on the broadcast and is never targeted again.
-        assert not any(
-            c["target_nodes"] == "node-object:node-2" for c in client.scan_calls
-        )
-
-    def test_single_node_cluster_is_drained(self):
-        layout = {"node-1": [f"{PREFIX}:{i}" for i in range(5)]}
-        client = MockClusterClient(layout)
 
         BaseCache(name=PREFIX, redis_client=client).clear()
 
-        assert sorted(client.deleted) == _all_keys(layout)
-        _assert_cursors_advanced(client)
+        _assert_drained(client, layout)
 
     def test_empty_cache_issues_no_delete(self):
         client = MockClusterClient({"node-1": [], "node-2": []})
 
         BaseCache(name=PREFIX, redis_client=client).clear()
 
-        assert client.deleted == []
-        assert len(client.scan_calls) == 1
-
-    def test_tolerates_duplicate_keys_across_pages(self):
-        # SCAN may return the same key more than once. DEL is idempotent, so
-        # clear() must not choke -- and nothing here counts keys.
-        dup = f"{PREFIX}:dup"
-        client = MockClusterClient({"node-1": [dup, dup, f"{PREFIX}:other", dup]})
-
-        BaseCache(name=PREFIX, redis_client=client).clear()
-
-        assert sorted(set(client.deleted)) == [dup, f"{PREFIX}:other"]
-
-    def test_deletes_are_batched(self):
-        n = CLEAR_BATCH_SIZE * 2 + 7
-        layout = {"node-1": [f"{PREFIX}:{i}" for i in range(n)]}
-        client = MockClusterClient(layout, page=CLEAR_BATCH_SIZE)
-        calls = []
+        assert client.delete_batches == []
 
-        real_delete = client.delete
-        client.delete = lambda *keys: (calls.append(len(keys)), real_delete(*keys))[1]
+    def test_deletes_are_batched(self, monkeypatch):
+        # Patch the batch size rather than asserting against its real value, so
+        # tuning CLEAR_BATCH_SIZE in production doesn't touch this test.
+        monkeypatch.setattr(cache_base, "CLEAR_BATCH_SIZE", 3)
+        keys = [f"{PREFIX}:{i}" for i in range(8)]
+        client = MockClusterClient({"node-1": keys}, page=3)
 
         BaseCache(name=PREFIX, redis_client=client).clear()
 
-        assert sorted(client.deleted) == _all_keys(layout)
-        assert calls == [CLEAR_BATCH_SIZE, CLEAR_BATCH_SIZE, 7]
+        # Bounded batches, and the trailing remainder is still flushed.
+        assert [len(b) for b in client.delete_batches] == [3, 3, 2]
+        assert sorted(client.deleted) == sorted(keys)
 
 
 class TestAsyncClearOnCluster:
-    """aclear() must match clear() behavior exactly."""
+    """aclear is written separately from clear, so it needs its own coverage."""
 
     @pytest.mark.asyncio
-    async def test_drains_all_primaries_with_uneven_depth(self):
+    async def test_drains_every_primary(self):
         layout = _layout()
         client = MockAsyncClusterClient(layout)
-        cache = BaseCache(name=PREFIX, async_redis_client=client)
 
-        await cache.aclear()
+        await BaseCache(name=PREFIX, async_redis_client=client).aclear()
 
-        assert sorted(client.deleted) == _all_keys(layout)
-        _assert_cursors_advanced(client)
+        _assert_drained(client, layout)
 
     @pytest.mark.asyncio
     async def test_empty_cache_issues_no_delete(self):
@@ -232,65 +183,4 @@ async def test_empty_cache_issues_no_delete(self):
 
         await BaseCache(name=PREFIX, async_redis_client=client).aclear()
 
-        assert client.deleted == []
-
-    @pytest.mark.asyncio
-    async def test_tolerates_duplicate_keys_across_pages(self):
-        dup = f"{PREFIX}:dup"
-        client = MockAsyncClusterClient({"node-1": [dup, dup, f"{PREFIX}:other", dup]})
-
-        await BaseCache(name=PREFIX, async_redis_client=client).aclear()
-
-        assert sorted(set(client.deleted)) == [dup, f"{PREFIX}:other"]
-
-
-class TestClearOnStandalone:
-    """The standalone path must keep working unchanged."""
-
-    def test_advances_single_cursor(self):
-        keys = [f"{PREFIX}:{i}" for i in range(7)]
-        client = MockStandaloneClient({"single": keys})
-
-        BaseCache(name=PREFIX, redis_client=client).clear()
-
-        assert sorted(client.deleted) == sorted(keys)
-        cursors = [c["cursor"] for c in client.scan_calls]
-        assert cursors[0] in (0, "0")
-        assert [int(c) for c in cursors] == [0, 2, 4, 6]
-
-    @pytest.mark.asyncio
-    async def test_async_advances_single_cursor(self):
-        keys = [f"{PREFIX}:{i}" for i in range(7)]
-        client = MockAsyncStandaloneClient({"single": keys})
-
-        await BaseCache(name=PREFIX, async_redis_client=client).aclear()
-
-        assert sorted(client.deleted) == sorted(keys)
-        assert [int(c["cursor"]) for c in client.scan_calls] == [0, 2, 4, 6]
-
-
-class TestConcreteCachesClearOnCluster:
-    """The fix must hold through the cache classes users actually instantiate.
-
-    Also pins that each subclass's effective key prefix really is ``<name>:*``,
-    which clear() depends on and nothing else covers.
-    """
-
-    def test_embeddings_cache_drains_cluster(self):
-        layout = _layout()
-        client = MockClusterClient(layout)
-
-        EmbeddingsCache(name=PREFIX, redis_client=client).clear()
-
-        assert sorted(client.deleted) == _all_keys(layout)
-        assert all(c["match"] == MATCH for c in client.scan_calls)
-
-    @pytest.mark.asyncio
-    async def test_embeddings_cache_drains_cluster_async(self):
-        layout = _layout()
-        client = MockAsyncClusterClient(layout)
-
-        await EmbeddingsCache(name=PREFIX, async_redis_client=client).aclear()
-
-        assert sorted(client.deleted) == _all_keys(layout)
-        assert all(c["match"] == MATCH for c in client.scan_calls)
+        assert client.delete_batches == []
diff --git a/tests/unit/test_migration_cluster_scan.py b/tests/unit/test_migration_cluster_scan.py
new file mode 100644
index 00000000..984f7e4a
--- /dev/null
+++ b/tests/unit/test_migration_cluster_scan.py
@@ -0,0 +1,146 @@
+"""Regression tests for migration SCAN key enumeration on Redis Cluster.
+
+Six migration call sites used to feed the previous reply's cursor straight back
+into ``client.scan(cursor=...)``. On a cluster client that reply is a
+``{node_name: cursor}`` mapping, so ``cursor == 0`` is never true and the second
+iteration passes a dict as the cursor -- redis-py rejects it with ``DataError``.
+That is a hard crash on any cluster, not a silent under-count.
+
+The fake below reproduces exactly that: it replies with per-node cursors like a
+real cluster and raises ``DataError`` if handed a non-integer cursor, so the old
+loop fails here the same way it fails against real Redis.
+
+Covers the four always-reachable sites (validator key counts, planner key
+sampling; sync and async). The two executor sites
+(``_enumerate_with_scan``) are the same one-line ``scan_iter`` pattern but sit
+behind index-info mocking, and they only run on fallback paths; they are left to
+the migration integration tests.
+"""
+
+import pytest
+from redis.asyncio.cluster import RedisCluster as AsyncRedisCluster
+from redis.cluster import RedisCluster
+from redis.exceptions import DataError
+
+from redisvl.migration.async_planner import AsyncMigrationPlanner
+from redisvl.migration.async_validation import AsyncMigrationValidator
+from redisvl.migration.planner import MigrationPlanner
+from redisvl.migration.validation import MigrationValidator
+
+PREFIX = "doc"
+NODES = {
+    "node-1": [f"{PREFIX}:a{i}".encode() for i in range(6)],
+    "node-2": [f"{PREFIX}:b{i}".encode() for i in range(2)],
+    "node-3": [f"{PREFIX}:c{i}".encode() for i in range(4)],
+}
+ALL_KEYS = sorted(k.decode() for ks in NODES.values() for k in ks)
+
+
+class MockClusterClient:
+    """Cluster-shaped SCAN: per-node cursors, and DataError on a dict cursor."""
+
+    scan_iter = RedisCluster.scan_iter
+
+    def __init__(self, keys_by_node=None, page=2, max_calls=200):
+        self.keys_by_node = {n: list(k) for n, k in (keys_by_node or NODES).items()}
+        self.page = page
+        self.max_calls = max_calls
+        self.scan_calls = 0
+
+    def _page(self, node, cursor):
+        keys = self.keys_by_node[node]
+        nxt = cursor + self.page
+        return (0 if nxt >= len(keys) else nxt), keys[cursor : cursor + self.page]
+
+    def _next_scan(self, cursor, target_nodes):
+        self.scan_calls += 1
+        if self.scan_calls > self.max_calls:
+            raise AssertionError(f"{self.scan_calls} SCAN calls -- not advancing")
+        # What redis-py's encoder does when the old loop hands back the mapping.
+        if not isinstance(cursor, (int, str, bytes)):
+            raise DataError(
+                f"Invalid input of type: {type(cursor).__name__!r}. "
+                "Convert to a bytes, string, int or float first."
+            )
+        cursor = int(cursor)
+        if target_nodes is None:
+            assert cursor == 0, f"node-local cursor {cursor!r} broadcast to primaries"
+            pages = {n: self._page(n, 0) for n in self.keys_by_node}
+            return (
+                {n: c for n, (c, _) in pages.items()},
+                [k for _, ks in pages.values() for k in ks],
+            )
+        node = str(target_nodes).removeprefix("node-object:")
+        nxt, keys = self._page(node, cursor)
+        return {node: nxt}, keys
+
+    def scan(self, cursor=0, match=None, count=None, target_nodes=None, _type=None):
+        return self._next_scan(cursor, target_nodes)
+
+    def get_node(self, host=None, port=None, node_name=None):
+        return f"node-object:{node_name}"
+
+
+class MockAsyncClusterClient(MockClusterClient):
+    scan_iter = AsyncRedisCluster.scan_iter
+
+    async def scan(
+        self, cursor=0, match=None, count=None, target_nodes=None, _type=None
+    ):
+        return self._next_scan(cursor, target_nodes)
+
+
+class _Index:
+    """Minimal SearchIndex stand-in for the validators' key-count path."""
+
+    def __init__(self, client):
+        self.client = client
+        self.schema = type(
+            "S",
+            (),
+            {"index": type("I", (), {"prefix": PREFIX, "key_separator": ":"})()},
+        )()
+
+
+class TestValidatorCountsKeysOnCluster:
+    """_count_index_keys runs on every validate, so this is the hottest path."""
+
+    def test_counts_across_all_primaries(self):
+        client = MockClusterClient()
+        validator = MigrationValidator.__new__(MigrationValidator)
+
+        assert validator._count_index_keys(_Index(client)) == len(ALL_KEYS)
+
+    @pytest.mark.asyncio
+    async def test_counts_across_all_primaries_async(self):
+        client = MockAsyncClusterClient()
+        validator = AsyncMigrationValidator.__new__(AsyncMigrationValidator)
+
+        assert await validator._count_index_keys(_Index(client)) == len(ALL_KEYS)
+
+
+class TestPlannerSamplesKeysOnCluster:
+    """_sample_keys returns early at the sample limit, mid-page."""
+
+    def test_samples_across_primaries(self):
+        client = MockClusterClient()
+        planner = MigrationPlanner.__new__(MigrationPlanner)
+        planner.key_sample_limit = len(ALL_KEYS)
+
+        sample = planner._sample_keys(
+            client=client, prefixes=[PREFIX], key_separator=":"
+        )
+
+        assert sorted(sample) == ALL_KEYS
+
+    @pytest.mark.asyncio
+    async def test_samples_across_primaries_async(self):
+        client = MockAsyncClusterClient()
+        planner = AsyncMigrationPlanner.__new__(AsyncMigrationPlanner)
+        planner.key_sample_limit = len(ALL_KEYS)
+
+        sample = await planner._async_sample_keys(
+            client=client, prefixes=[PREFIX], key_separator=":"
+        )
+
+        assert sorted(sample) == ALL_KEYS
diff --git a/tests/unit/test_migration_planner.py b/tests/unit/test_migration_planner.py
index 092c472c..5bd52b26 100644
--- a/tests/unit/test_migration_planner.py
+++ b/tests/unit/test_migration_planner.py
@@ -25,7 +25,7 @@ class DummyClient:
     def __init__(self, keys):
         self.keys = keys
 
-    def scan(self, cursor=0, match=None, count=None, _type=None, **kwargs):
+    def scan(self, cursor=0, match=None, count=None, _type=None):
         matched = []
         for key in self.keys:
             decoded_key = key.decode() if isinstance(key, bytes) else str(key)