Use incremental Redis scans for cache maintenance

src/cachier/cores/redis.py

@@ -3,6 +3,7 @@
 import pickle
 import time
 import warnings
+from collections.abc import AsyncIterator, Iterator
 from datetime import datetime, timedelta
 from typing import TYPE_CHECKING, Any, Awaitable, Callable, Optional, Tuple, Union
 
@@ -340,40 +341,62 @@ def wait_on_entry_calc(self, key: str) -> Any:
                 return entry.value
             self.check_calc_timeout(time_spent)
 
+    def _scan_key_batches(self, client: Any) -> Iterator[list[Union[str, bytes]]]:
+        """Yield matching cache keys in incremental Redis SCAN pages.
+
+        Empty pages do not terminate iteration while the cursor is nonzero.
+
+        """
+        pattern = f"{self.key_prefix}:{self._func_str}:*"
+        cursor = 0
+        while True:
+            cursor, keys = client.scan(cursor=cursor, match=pattern, count=100)
+            if keys:
+                yield keys
+            if cursor == 0:
+                break
+
+    async def _ascan_key_batches(self, client: Any) -> AsyncIterator[list[Union[str, bytes]]]:
+        """Yield matching cache keys in asynchronous Redis SCAN pages.
+
+        Empty pages do not terminate iteration while the cursor is nonzero.
+
+        """
+        pattern = f"{self.key_prefix}:{self._func_str}:*"
+        cursor = 0
+        while True:
+            cursor, keys = await client.scan(cursor=cursor, match=pattern, count=100)
+            if keys:
+                yield keys
+            if cursor == 0:
+                break
+
     def clear_cache(self) -> None:
         """Clear the cache of this core."""
         redis_client = self._resolve_redis_client()
-        pattern = f"{self.key_prefix}:{self._func_str}:*"
 
         try:
-            # Find all keys matching the pattern
-            keys = redis_client.keys(pattern)
-            if keys:
+            for keys in self._scan_key_batches(redis_client):
                 redis_client.delete(*keys)
         except Exception as e:
             warnings.warn(f"Redis clear_cache failed: {e}", stacklevel=2)
 
     async def aclear_cache(self) -> None:
         """Clear the cache of this core asynchronously."""
         redis_client = await self._resolve_redis_client_async()
-        pattern = f"{self.key_prefix}:{self._func_str}:*"
 
         try:
-            keys = await redis_client.keys(pattern)
-            if keys:
+            async for keys in self._ascan_key_batches(redis_client):
                 await redis_client.delete(*keys)
         except Exception as e:
             warnings.warn(f"Redis clear_cache failed: {e}", stacklevel=2)
 
     def clear_being_calculated(self) -> None:
         """Mark all entries in this cache as not being calculated."""
         redis_client = self._resolve_redis_client()
-        pattern = f"{self.key_prefix}:{self._func_str}:*"
 
         try:
-            # Find all keys matching the pattern
-            keys = redis_client.keys(pattern)
-            if keys:
+            for keys in self._scan_key_batches(redis_client):
                 # Use pipeline for efficiency
                 pipe = redis_client.pipeline()
                 for key in keys:
@@ -385,68 +408,65 @@ def clear_being_calculated(self) -> None:
     async def aclear_being_calculated(self) -> None:
         """Mark all entries in this cache as not being calculated asynchronously."""
         redis_client = await self._resolve_redis_client_async()
-        pattern = f"{self.key_prefix}:{self._func_str}:*"
 
         try:
-            keys = await redis_client.keys(pattern)
-            for key in keys or []:
-                await redis_client.hset(key, "processing", "false")
+            async for keys in self._ascan_key_batches(redis_client):
+                for key in keys:
+                    await redis_client.hset(key, "processing", "false")
         except Exception as e:
             warnings.warn(f"Redis clear_being_calculated failed: {e}", stacklevel=2)
 
     def delete_stale_entries(self, stale_after: timedelta) -> None:
         """Remove stale entries from the Redis cache."""
         redis_client = self._resolve_redis_client()
-        pattern = f"{self.key_prefix}:{self._func_str}:*"
         try:
-            keys = redis_client.keys(pattern)
             threshold = datetime.now() - stale_after
-            for key in keys:
-                ts = redis_client.hget(key, "timestamp")
-                if ts is None:
-                    continue
-                # ts may be bytes or str depending on client configuration
-                if isinstance(ts, bytes):
+            for keys in self._scan_key_batches(redis_client):
+                for key in keys:
+                    ts = redis_client.hget(key, "timestamp")
+                    if ts is None:
+                        continue
+                    # ts may be bytes or str depending on client configuration
+                    if isinstance(ts, bytes):
+                        try:
+                            ts_s = ts.decode("utf-8")
+                        except Exception:
+                            ts_s = ts.decode("latin-1", errors="ignore")
+                    else:
+                        ts_s = str(ts)
                     try:
-                        ts_s = ts.decode("utf-8")
-                    except Exception:
-                        ts_s = ts.decode("latin-1", errors="ignore")
-                else:
-                    ts_s = str(ts)
-                try:
-                    ts_val = datetime.fromisoformat(ts_s)
-                except Exception as exc:
-                    warnings.warn(f"Redis timestamp parse failed: {exc}", stacklevel=2)
-                    continue
-                if ts_val < threshold:
-                    redis_client.delete(key)
+                        ts_val = datetime.fromisoformat(ts_s)
+                    except Exception as exc:
+                        warnings.warn(f"Redis timestamp parse failed: {exc}", stacklevel=2)
+                        continue
+                    if ts_val < threshold:
+                        redis_client.delete(key)
         except Exception as e:
             warnings.warn(f"Redis delete_stale_entries failed: {e}", stacklevel=2)
 
     async def adelete_stale_entries(self, stale_after: timedelta) -> None:
         """Remove stale entries from the Redis cache asynchronously."""
         redis_client = await self._resolve_redis_client_async()
-        pattern = f"{self.key_prefix}:{self._func_str}:*"
         try:
-            keys = await redis_client.keys(pattern)
             threshold = datetime.now() - stale_after
-            for key in keys or []:
-                ts = await redis_client.hget(key, "timestamp")
-                if ts is None:
-                    continue
-                if isinstance(ts, bytes):
+            async for keys in self._ascan_key_batches(redis_client):
+                for key in keys:
+                    ts = await redis_client.hget(key, "timestamp")
+                    if ts is None:
+                        continue
+                    if isinstance(ts, bytes):
+                        try:
+                            ts_s = ts.decode("utf-8")
+                        except Exception:
+                            ts_s = ts.decode("latin-1", errors="ignore")
+                    else:
+                        ts_s = str(ts)
                     try:
-                        ts_s = ts.decode("utf-8")
-                    except Exception:
-                        ts_s = ts.decode("latin-1", errors="ignore")
-                else:
-                    ts_s = str(ts)
-                try:
-                    ts_val = datetime.fromisoformat(ts_s)
-                except Exception as exc:
-                    warnings.warn(f"Redis timestamp parse failed: {exc}", stacklevel=2)
-                    continue
-                if ts_val < threshold:
-                    await redis_client.delete(key)
+                        ts_val = datetime.fromisoformat(ts_s)
+                    except Exception as exc:
+                        warnings.warn(f"Redis timestamp parse failed: {exc}", stacklevel=2)
+                        continue
+                    if ts_val < threshold:
+                        await redis_client.delete(key)
         except Exception as e:
             warnings.warn(f"Redis delete_stale_entries failed: {e}", stacklevel=2)

tests/redis_tests/clients.py

@@ -1,5 +1,6 @@
 """Redis test client implementations shared across Redis test modules."""
 
+from collections.abc import Sequence
 from fnmatch import fnmatch
 
 
@@ -59,6 +60,14 @@ def keys(self, pattern):
         print(f"DEBUG: keys({pattern}) = {result}")
         return result
 
+    def scan(self, cursor: int = 0, match: str = "*", count: int = 100) -> tuple[int, Sequence[str]]:
+        """Return matching keys as a single scan page.
+
+        The in-memory client does not need multiple cursor pages.
+
+        """
+        return 0, [key for key in self.data if fnmatch(key, match)]
+
     def delete(self, *keys):
         for key in keys:
             self.data.pop(key, None)
@@ -140,6 +149,16 @@ async def keys(self, pattern: str) -> list[str]:
             raise Exception("keys failed")
         return [key for key in self._data if fnmatch(key, pattern)]
 
+    async def scan(self, cursor: int = 0, match: str = "*", count: int = 100) -> tuple[int, Sequence[str]]:
+        """Return matching keys as a single asynchronous scan page.
+
+        Reuse the key-scan failure flag for error-path tests.
+
+        """
+        if self.fail_keys:
+            raise Exception("scan failed")
+        return 0, [key for key in self._data if fnmatch(key, match)]
+
     async def delete(self, *keys: str):
         if self.fail_delete:
             raise Exception("delete failed")

tests/redis_tests/test_redis_core.py

@@ -351,7 +351,7 @@ def test_redis_core_exceptions():
     # Configure all methods to raise exceptions
     mock_client.hgetall = MagicMock(side_effect=Exception("Redis connection error"))
     mock_client.hset = MagicMock(side_effect=Exception("Redis write error"))
-    mock_client.keys = MagicMock(side_effect=Exception("Redis keys error"))
+    mock_client.scan = MagicMock(side_effect=Exception("Redis keys error"))
     mock_client.delete = MagicMock(side_effect=Exception("Redis delete error"))
 
     core = _RedisCore(hash_func=None, redis_client=mock_client, wait_for_calc_timeout=10)
@@ -449,7 +449,7 @@ def mock_func():
     delete_mock_client = MagicMock()
 
     # Set up keys method
-    delete_mock_client.keys = MagicMock(return_value=[b"key1", b"key2", b"key3"])
+    delete_mock_client.scan = MagicMock(return_value=(0, [b"key1", b"key2", b"key3"]))
 
     now = datetime.now()
     old_timestamp = (now - timedelta(hours=2)).isoformat()
@@ -482,7 +482,7 @@ def mock_func():
 
     # Test exception during timestamp parsing
     mock_client.reset_mock()
-    mock_client.keys.return_value = [b"key4"]
+    mock_client.scan.return_value = (0, [b"key4"])
     mock_client.hget.return_value = b"invalid-timestamp"
 
     # Need to mock _resolve_redis_client for the original core as well
@@ -496,7 +496,7 @@ def mock_func():
 
     # Test exception during keys operation
     mock_client.reset_mock()
-    mock_client.keys.side_effect = Exception("Redis keys error")
+    mock_client.scan.side_effect = Exception("Redis keys error")
 
     with warnings.catch_warnings(record=True) as w2:
         warnings.simplefilter("always")
@@ -565,7 +565,7 @@ def test_redis_clear_being_calculated_with_pipeline():
     pipeline_mock = MagicMock()
 
     # Set up keys to return 3 keys
-    pipeline_mock_client.keys = MagicMock(return_value=[b"key1", b"key2", b"key3"])
+    pipeline_mock_client.scan = MagicMock(return_value=(0, [b"key1", b"key2", b"key3"]))
 
     # Set up pipeline
     pipeline_mock_client.pipeline = MagicMock(return_value=pipeline_mock)
@@ -598,7 +598,7 @@ def mock_func():
 def test_redis_clear_being_calculated_without_keys():
     """Test clear_being_calculated when no keys are returned."""
     mock_client = MagicMock()
-    mock_client.keys = MagicMock(return_value=[])
+    mock_client.scan = MagicMock(return_value=(0, []))
     mock_client.pipeline = MagicMock()
 
     core = _RedisCore(hash_func=None, redis_client=mock_client, wait_for_calc_timeout=10)

tests/redis_tests/test_redis_core_exceptions.py

@@ -92,33 +92,33 @@ def test_mark_entry_not_calculated_exceptions(self, core, mock_redis):
 
     def test_clear_cache_exceptions(self, core, mock_redis):
         """Test clear_cache Redis keys exception handling."""
-        mock_redis.keys.side_effect = Exception("Redis error")
+        mock_redis.scan.side_effect = Exception("Redis error")
         with pytest.warns(UserWarning, match="Redis clear_cache failed"):
             core.clear_cache()
 
     def test_clear_being_calculated_exceptions(self, core, mock_redis):
         """Test clear_being_calculated Redis keys exception handling."""
-        mock_redis.keys.side_effect = Exception("Redis error")
+        mock_redis.scan.side_effect = Exception("Redis error")
         with pytest.warns(UserWarning, match="Redis clear_being_calculated failed"):
             core.clear_being_calculated()
 
     def test_delete_stale_entries_keys_exception(self, core, mock_redis):
         """Test delete_stale_entries Redis keys exception handling."""
-        mock_redis.keys.side_effect = Exception("Redis error")
+        mock_redis.scan.side_effect = Exception("Redis error")
         with pytest.warns(UserWarning, match="Redis delete_stale_entries failed"):
             core.delete_stale_entries(timedelta(seconds=1))
 
     def test_delete_stale_entries_timestamp_parse_exception(self, core, mock_redis):
         """Test delete_stale_entries timestamp parsing exception handling."""
-        mock_redis.keys.return_value = [b"key1"]
+        mock_redis.scan.return_value = (0, [b"key1"])
         mock_redis.hget.return_value = b"invalid_timestamp"
 
         with pytest.warns(UserWarning, match="Redis timestamp parse failed"):
             core.delete_stale_entries(timedelta(seconds=1))
 
     def test_delete_stale_entries_latin1_fallback(self, core, mock_redis):
         """Test delete_stale_entries uses latin-1 for invalid utf-8."""
-        mock_redis.keys.return_value = [b"key1"]
+        mock_redis.scan.return_value = (0, [b"key1"])
         # b'\xff' is invalid utf-8 start byte
         mock_redis.hget.return_value = b"\xff"
 
@@ -128,7 +128,7 @@ def test_delete_stale_entries_latin1_fallback(self, core, mock_redis):
 
     def test_delete_stale_entries_str_timestamp(self, core, mock_redis):
         """Test delete_stale_entries handles string timestamps (not bytes)."""
-        mock_redis.keys.return_value = [b"key1"]
+        mock_redis.scan.return_value = (0, [b"key1"])
         now = datetime.now()
         old_time = now - timedelta(hours=1)
         # Return a string, not bytes