From 115c10ea76cebd50a61b083beef2a71dd0df39e7 Mon Sep 17 00:00:00 2001 From: Apeksha_2022 <70309060+Apeksha9389@users.noreply.github.com> Date: Sat, 17 Oct 2026 18:59:52 +0530 Subject: [PATCH] Scan Redis cache keys incrementally during maintenance --- src/cachier/cores/redis.py | 128 ++++++++++-------- tests/redis_tests/clients.py | 19 +++ tests/redis_tests/test_redis_core.py | 12 +- .../redis_tests/test_redis_core_exceptions.py | 12 +- tests/redis_tests/test_scan_maintenance.py | 87 ++++++++++++ 5 files changed, 192 insertions(+), 66 deletions(-) create mode 100644 tests/redis_tests/test_scan_maintenance.py diff --git a/src/cachier/cores/redis.py b/src/cachier/cores/redis.py index df6eea00..71dcf321 100644 --- a/src/cachier/cores/redis.py +++ b/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,15 +341,42 @@ 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) @@ -356,11 +384,9 @@ def clear_cache(self) -> None: 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) @@ -368,12 +394,9 @@ async def aclear_cache(self) -> None: 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) diff --git a/tests/redis_tests/clients.py b/tests/redis_tests/clients.py index 6a5d2ec1..c71eaacd 100644 --- a/tests/redis_tests/clients.py +++ b/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") diff --git a/tests/redis_tests/test_redis_core.py b/tests/redis_tests/test_redis_core.py index 6b2d9f39..1b181b64 100644 --- a/tests/redis_tests/test_redis_core.py +++ b/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) diff --git a/tests/redis_tests/test_redis_core_exceptions.py b/tests/redis_tests/test_redis_core_exceptions.py index 7d51eb11..924461b5 100644 --- a/tests/redis_tests/test_redis_core_exceptions.py +++ b/tests/redis_tests/test_redis_core_exceptions.py @@ -92,25 +92,25 @@ 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"): @@ -118,7 +118,7 @@ def test_delete_stale_entries_timestamp_parse_exception(self, core, mock_redis): 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 diff --git a/tests/redis_tests/test_scan_maintenance.py b/tests/redis_tests/test_scan_maintenance.py new file mode 100644 index 00000000..54196018 --- /dev/null +++ b/tests/redis_tests/test_scan_maintenance.py @@ -0,0 +1,87 @@ +"""Redis maintenance must traverse scan pages without blocking KEYS calls.""" + +from datetime import datetime, timedelta +from unittest.mock import AsyncMock, NonCallableMock, call + +import pytest + +from cachier.cores.redis import _RedisCore + + +def _identity(value: str) -> str: + """Return a value for cache function identity. + + A Python callable provides the inspectable signature required by the core. + + """ + return value + + +@pytest.mark.redis +@pytest.mark.parametrize("operation", ["clear_cache", "clear_being_calculated", "delete_stale_entries"]) +def test_maintenance_scans_every_page(operation: str) -> None: + """Visit empty intermediate pages and the final nonempty page. + + Keep deletes and pipeline writes confined to each returned batch. + + """ + pytest.importorskip("redis") + client = NonCallableMock() + client.scan.side_effect = [(7, [b"first"]), (19, []), (0, [b"last"])] + client.keys.side_effect = AssertionError("KEYS must not be used") + client.hget.return_value = (datetime.now() - timedelta(days=2)).isoformat() + core = _RedisCore(hash_func=None, redis_client=client) + core.set_func(_identity) + args = (timedelta(days=1),) if operation == "delete_stale_entries" else () + getattr(core, operation)(*args) + pattern = f"{core.key_prefix}:{core._func_str}:*" + assert client.scan.call_args_list == [ + call(cursor=0, match=pattern, count=100), + call(cursor=7, match=pattern, count=100), + call(cursor=19, match=pattern, count=100), + ] + client.keys.assert_not_called() + if operation == "clear_being_calculated": + assert client.pipeline.return_value.hset.call_args_list == [ + call(b"first", "processing", "false"), + call(b"last", "processing", "false"), + ] + assert client.pipeline.return_value.execute.call_count == 2 + else: + assert client.delete.call_args_list == [call(b"first"), call(b"last")] + + +@pytest.mark.redis +@pytest.mark.asyncio +@pytest.mark.parametrize("operation", ["aclear_cache", "aclear_being_calculated", "adelete_stale_entries"]) +async def test_async_maintenance_scans_every_page(operation: str) -> None: + """Traverse asynchronous scan pages without stopping on an empty page. + + Process the final page before the zero cursor terminates iteration. + + """ + pytest.importorskip("redis") + client = NonCallableMock() + client.scan = AsyncMock(side_effect=[(7, [b"first"]), (19, []), (0, [b"last"])]) + client.keys = AsyncMock(side_effect=AssertionError("KEYS must not be used")) + client.hget = AsyncMock(return_value=(datetime.now() - timedelta(days=2)).isoformat()) + client.hset = AsyncMock() + client.delete = AsyncMock() + core = _RedisCore(hash_func=None, redis_client=client) + core.set_func(_identity) + args = (timedelta(days=1),) if operation == "adelete_stale_entries" else () + await getattr(core, operation)(*args) + pattern = f"{core.key_prefix}:{core._func_str}:*" + assert client.scan.await_args_list == [ + call(cursor=0, match=pattern, count=100), + call(cursor=7, match=pattern, count=100), + call(cursor=19, match=pattern, count=100), + ] + client.keys.assert_not_called() + if operation == "aclear_being_calculated": + assert client.hset.await_args_list == [ + call(b"first", "processing", "false"), + call(b"last", "processing", "false"), + ] + else: + assert client.delete.await_args_list == [call(b"first"), call(b"last")]