fix: retry the batched Redis value read and test the batching helper

The default iteration retries every read through `get_value`, but the
batched HMGET in the Redis override was not retried. It now returns a
list under the same `retry_on_error` as the other Redis reads, matching
the SQL client.

Add a direct unit test for `batch_records_by_size`, covering the count
bound, the size bound, an oversized record alone in its batch and
records of unknown size.

Co-Authored-By: Claude Fable 5.1 <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_01Wg6jZgyQp7XkvVvxuueJyV

Author: Pijukatel

src/crawlee/storage_clients/_redis/_key_value_store_client.py

@@ -302,19 +302,20 @@ async def iterate_entries(
             max_records=self._ITERATE_ENTRIES_BATCH_MAX_KEYS,
             max_bytes=self._ITERATE_ENTRIES_BATCH_MAX_BYTES,
         ):
-            async for record in self._fetch_records(batch):
+            for record in await self._fetch_records(batch):
                 yield record
 
-    async def _fetch_records(self, batch: list[KeyValueStoreRecordMetadata]) -> AsyncIterator[KeyValueStoreRecord]:
-        """Fetch the values of the given records with a single HMGET call and yield the deserialized records."""
+    @retry_on_error(RedisError)
+    async def _fetch_records(self, batch: list[KeyValueStoreRecordMetadata]) -> list[KeyValueStoreRecord]:
+        """Fetch the values of the given records with a single HMGET call and return the deserialized records."""
         keys = [item.key for item in batch]
         values = [expect_bytes(v) for v in await await_redis_response(self._redis.hmget(self._items_key, keys))]
 
-        for metadata_item, value_bytes in zip(batch, values, strict=True):
-            record = self._build_record(metadata_item, value_bytes)
-            if record is None:
-                continue
-            yield record
+        records = (
+            self._build_record(metadata_item, value_bytes)
+            for metadata_item, value_bytes in zip(batch, values, strict=True)
+        )
+        return [record for record in records if record is not None]
 
     @override
     async def get_public_url(self, *, key: str) -> str:

tests/unit/storage_clients/test_utils.py

@@ -0,0 +1,48 @@
+from __future__ import annotations
+
+from crawlee.storage_clients._utils import batch_records_by_size
+from crawlee.storage_clients.models import KeyValueStoreRecordMetadata
+
+
+def _record(key: str, size: int | None) -> KeyValueStoreRecordMetadata:
+    return KeyValueStoreRecordMetadata(key=key, content_type='application/octet-stream', size=size)
+
+
+def _keys(batches: list[list[KeyValueStoreRecordMetadata]]) -> list[list[str]]:
+    return [[record.key for record in batch] for batch in batches]
+
+
+def test_batches_are_bounded_by_record_count() -> None:
+    records = [_record(f'k{i}', 1) for i in range(5)]
+
+    batches = list(batch_records_by_size(records, max_records=2, max_bytes=1000))
+
+    assert _keys(batches) == [['k0', 'k1'], ['k2', 'k3'], ['k4']]
+
+
+def test_batches_are_bounded_by_total_size() -> None:
+    records = [_record('a', 4), _record('b', 4), _record('c', 4), _record('d', 4)]
+
+    batches = list(batch_records_by_size(records, max_records=100, max_bytes=10))
+
+    assert _keys(batches) == [['a', 'b'], ['c', 'd']]
+
+
+def test_oversized_record_is_alone_in_its_batch() -> None:
+    records = [_record('small1', 2), _record('large', 100), _record('small2', 2), _record('small3', 2)]
+
+    batches = list(batch_records_by_size(records, max_records=100, max_bytes=10))
+
+    assert _keys(batches) == [['small1'], ['large'], ['small2', 'small3']]
+
+
+def test_unknown_size_counts_as_empty() -> None:
+    records = [_record('a', None), _record('b', None), _record('c', 10)]
+
+    batches = list(batch_records_by_size(records, max_records=100, max_bytes=10))
+
+    assert _keys(batches) == [['a', 'b', 'c']]
+
+
+def test_no_records_yield_no_batches() -> None:
+    assert list(batch_records_by_size([], max_records=100, max_bytes=10)) == []