diff --git a/fastapi_cachex/backends/redis.py b/fastapi_cachex/backends/redis.py index 02f17e7..b9ab815 100644 --- a/fastapi_cachex/backends/redis.py +++ b/fastapi_cachex/backends/redis.py @@ -460,28 +460,29 @@ async def get_cache_data(self) -> dict[str, tuple[CacheEntry, float | None]]: all_keys = await self.get_all_keys() cache_data: dict[str, tuple[CacheEntry, float | None]] = {} - if not all_keys: - return cache_data - - # Fetch every value and its remaining lifetime in a single pipeline - # round-trip instead of 2N commands. - pipe = self.client.pipeline() - for key in all_keys: - redis_key = self._make_key(key) - pipe.get(redis_key) - pipe.pttl(redis_key) - replies: list[Any] = await pipe.execute() - now = time.time() - - for key, raw, pttl in zip(all_keys, replies[::2], replies[1::2], strict=True): - # -2: the key expired or was deleted between SCAN and this fetch. - if pttl == _PTTL_MISSING: - continue - value = decode_entry(raw) - if value is None: - continue - expiry = None if pttl == _PTTL_NO_EXPIRY else now + pttl / 1000 - cache_data[key] = (value, expiry) + # Fetch values and remaining lifetimes in pipelines of _BATCH_SIZE + # keys instead of 2N round trips. The pipelines are not transactional: + # a MULTI/EXEC over the whole keyspace would block the server for its + # duration, and a consistent snapshot is not needed for monitoring. + for start in range(0, len(all_keys), _BATCH_SIZE): + chunk = all_keys[start : start + _BATCH_SIZE] + pipe = self.client.pipeline(transaction=False) + for key in chunk: + redis_key = self._make_key(key) + pipe.get(redis_key) + pipe.pttl(redis_key) + replies: list[Any] = await pipe.execute() + now = time.time() + + for key, raw, pttl in zip(chunk, replies[::2], replies[1::2], strict=True): + # -2: the key expired or was deleted between SCAN and this fetch. + if pttl == _PTTL_MISSING: + continue + value = decode_entry(raw) + if value is None: + continue + expiry = None if pttl == _PTTL_NO_EXPIRY else now + pttl / 1000 + cache_data[key] = (value, expiry) logger.debug("Redis GET_CACHE_DATA; keys=%s", len(cache_data)) return cache_data diff --git a/tests/backends/test_redis.py b/tests/backends/test_redis.py index 993e3e9..4fd5642 100644 --- a/tests/backends/test_redis.py +++ b/tests/backends/test_redis.py @@ -874,6 +874,44 @@ async def test_redis_get_cache_data_reports_absolute_expiry( assert data["forever"][1] is None +@requires_redis +@pytest.mark.asyncio +async def test_redis_get_cache_data_uses_chunked_non_transactional_pipelines( + async_redis_backend: AsyncRedisCacheBackend, + monkeypatch: pytest.MonkeyPatch, +) -> None: + """One MULTI/EXEC over the whole keyspace would block the server (#171).""" + total = _BATCH_SIZE * 2 + 10 + for index in range(total): + await async_redis_backend.set( + f"chunk:{index}", CacheEntry(fingerprint="e", content=b"v"), ttl=60 + ) + + real_pipeline = async_redis_backend.client.pipeline + pipelines: list[tuple[bool, int]] = [] + + def recording_pipeline(*args: Any, **kwargs: Any) -> Any: + pipe = real_pipeline(*args, **kwargs) + real_execute = pipe.execute + + async def execute(*e_args: Any, **e_kwargs: Any) -> Any: + pipelines.append((kwargs.get("transaction", True), len(pipe))) + return await real_execute(*e_args, **e_kwargs) + + monkeypatch.setattr(pipe, "execute", execute) + return pipe + + monkeypatch.setattr(async_redis_backend.client, "pipeline", recording_pipeline) + + data = await async_redis_backend.get_cache_data() + + assert len(data) == total + assert all(expiry is not None for _entry, expiry in data.values()) + assert [transaction for transaction, _ in pipelines] == [False] * 3 + assert sum(size for _, size in pipelines) == 2 * total + assert max(size for _, size in pipelines) <= 2 * _BATCH_SIZE + + @requires_redis @pytest.mark.asyncio async def test_redis_get_cache_data_skips_keys_gone_after_scan(