perf(redis): fetch get_cache_data in chunked, non-transactional pipelines

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

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(