fix(redis): default max_records to 1 when retry_on_idle_ms is set

docs/docs/sdk/redis-transport.md

@@ -326,6 +326,7 @@ Guidelines:
 - `retry_on_idle_ms` should be comfortably longer than your handler's expected worst-case execution time to avoid false positives.
 - `retry_reclaim_interval_s` controls how often the background reclaimer wakes up. Lower values increase Redis load; 15 s is a sensible default for most workloads.
 - `max_retries` prevents poison messages from looping forever. Set to `None` for unlimited retries (no DLQ).
+- With `retry_on_idle_ms`, `max_records` defaults to 1 (unless `batch=True`). An entry's idle time starts when it is read, not when its handler starts, so a read of the whole backlog would let entries queued behind slow handlers be reclaimed and run twice. If you set `max_records` yourself, keep it at or below what the handler gets through within `retry_on_idle_ms`.
 
 ### Automatic Recovery from Redis Stream Loss (NOGROUP)
 

sdk/CHANGELOG.md

@@ -7,6 +7,20 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0
 
 ## [Unreleased]
 
+### Fixed
+
+- Redis `subscribe(..., retry_on_idle_ms=...)` without `max_records` read
+  every waiting entry in one `XREADGROUP` (no `COUNT`). All of them entered
+  the consumer's PEL at once and were handled one by one, so an entry queued
+  behind slow handlers went idle past `retry_on_idle_ms` before it started,
+  and the reclaimer re-delivered it: the handler ran twice, `_retry_count`
+  climbed (up to the DLQ) without a failure, a large backlog came back as one
+  reply, and other replicas got nothing to do. `max_records` now defaults to
+  1 when `retry_on_idle_ms` is set (not with `batch=True`), on the main and
+  the retry stream. An explicit `max_records` is kept. This means one
+  `XREADGROUP` per entry: setups that relied on the unbounded read for
+  throughput with small, fast messages can set `max_records` explicitly.
+
 ## [0.6.0] - 2026-09-25
 
 ### Added

sdk/eggai/transport/redis.py

@@ -337,7 +337,11 @@
                 per-process-unique name (``{handler_id}-{hostname}-{pid}``) so each worker in the group owns
                 a distinct slice of the PEL. Pass an explicit value only if you need a stable consumer name.
             batch (bool, optional): Whether to consume messages in batches (default is False).
-            max_records (Optional[int], optional): Maximum number of records to consume in one batch (default is None).
+            max_records (Optional[int], optional): Maximum number of records to consume in one batch (default is None,
+                no limit). With ``retry_on_idle_ms`` and ``batch=False`` it defaults to 1: every entry read enters this
+                consumer's PEL at once, and an entry queued behind slow handlers would go idle past
+                ``retry_on_idle_ms`` and be re-delivered before it starts. An explicit value is kept; keep it at or
+                below what the handler gets through within ``retry_on_idle_ms``.
             group_start (str, optional): Stream id a NEW consumer group is created at: "$" (default, only entries
                 published after the group exists), "0" (the whole existing backlog) or an explicit stream id.
                 Ignored when the group already exists, a group remembers its own position. Reads always use ">".
@@ -623,6 +627,15 @@
                 "retries/DLQ never trigger. Use the default NACK_ON_ERROR."
             )
 
+        # FastStream reads max_records entries per XREADGROUP (every waiting one
+        # when None) and handles them one by one. All of them enter this
+        # consumer's PEL at read time, so the ones queued behind a slow handler
+        # go idle past retry_on_idle_ms before they start, and the reclaimer
+        # re-delivers them while they are still queued here. Read one at a time.
+        # The retry subscribe below inherits it.
+        if retry_on_idle_ms is not None and not batch and max_records is None:
+            max_records = 1
+
         # Default max_retries only applies when retry_on_idle_ms is set.
         if retry_on_idle_ms is None:
             max_retries = None

sdk/tests/test_redis.py

@@ -703,6 +703,126 @@ async def handler(message):
     assert subs["orders.orders-handler-1.retry"].group_create_id == "$"
 
 
+@pytest.mark.asyncio
+async def test_retry_on_idle_ms_defaults_max_records_to_one():
+    """With the reclaimer on, an unset max_records becomes 1 on the main and
+    the retry stream; an explicit value and batch mode are left alone."""
+
+    async def handler(message):
+        return message
+
+    def max_records(transport):
+        return {
+            s.stream_sub.name: s.stream_sub.max_records
+            for s in transport.broker.subscribers
+        }
+
+    transport = RedisTransport()
+    await transport.subscribe(
+        "orders", handler, handler_id="orders-handler-1", retry_on_idle_ms=500
+    )
+    assert max_records(transport) == {
+        "orders": 1,
+        "orders.orders-handler-1.retry": 1,
+    }
+
+    transport = RedisTransport()
+    await transport.subscribe(
+        "orders",
+        handler,
+        handler_id="orders-handler-1",
+        retry_on_idle_ms=500,
+        max_records=10,
+    )
+    assert max_records(transport) == {
+        "orders": 10,
+        "orders.orders-handler-1.retry": 10,
+    }
+
+    transport = RedisTransport()
+    await transport.subscribe(
+        "orders",
+        handler,
+        handler_id="orders-handler-1",
+        retry_on_idle_ms=500,
+        batch=True,
+    )
+    assert set(max_records(transport).values()) == {None}
+
+    # No reclaimer: unchanged.
+    transport = RedisTransport()
+    await transport.subscribe("orders", handler, handler_id="orders-handler-1")
+    assert max_records(transport) == {"orders": None}
+
+
+@pytest.mark.asyncio
+async def test_retry_on_idle_ms_does_not_reclaim_entries_queued_behind_a_slow_handler():
+    """A backlog waiting at startup must not be read into the PEL all at once.
+
+    Without a COUNT, one XREADGROUP hands the consumer every waiting entry; they
+    all enter its PEL and are handled one by one. An entry queued behind slow
+    handlers goes idle past retry_on_idle_ms without ever having started, the
+    reclaimer moves it to .retry, and the handler runs it twice (once from the
+    local list, once from the retry stream).
+    """
+    redis_client = redis.Redis(host="localhost", port=6379, decode_responses=True)
+
+    test_id = uuid.uuid4().hex[:8]
+    channel_name = f"test-prefetch-{test_id}"
+    stream_name = f"eggai.{channel_name}"
+    group = f"prefetch-agent-{test_id}-handler-1"
+
+    producer = RedisTransport()
+    await producer.connect()
+    for n in range(1, 5):
+        await Channel(channel_name, transport=producer).publish(
+            {"type": "t", "data": {"n": n}}
+        )
+
+    transport = RedisTransport()
+    agent = Agent(f"prefetch-agent-{test_id}", transport=transport)
+    calls = []
+    all_seen = asyncio.Event()
+
+    @agent.subscribe(
+        channel=Channel(channel_name, transport=transport),
+        group=group,
+        group_start="0",
+        retry_on_idle_ms=1000,
+        retry_reclaim_interval_s=0.25,
+    )
+    async def handler(message):
+        calls.append(message["data"]["n"])
+        if set(calls) == {1, 2, 3, 4}:
+            all_seen.set()
+        await asyncio.sleep(1.0)
+
+    try:
+        await agent.start()
+        await asyncio.sleep(0.5)
+        # Only the entry being handled is claimed; the rest stays in the stream
+        # where another consumer could take it.
+        pending = await redis_client.xpending(stream_name, group)
+        (info,) = [
+            g
+            for g in await redis_client.xinfo_groups(stream_name)
+            if g["name"] == group
+        ]
+        assert pending["pending"] == 1
+        assert info["lag"] == 3
+
+        await asyncio.wait_for(all_seen.wait(), timeout=10.0)
+        # Leave time for a redelivery through .retry to show up.
+        await asyncio.sleep(2.0)
+    finally:
+        await agent.stop()
+        await producer.disconnect()
+        await redis_client.delete(stream_name, f"{stream_name}.{group}.retry")
+        await redis_client.aclose()
+
+    assert sorted(calls) == [1, 2, 3, 4], f"handler runs: {calls}"
+
+
 @pytest.mark.asyncio
 async def test_retry_stream_gets_its_own_tracing_destination(monkeypatch):
     """The retry-stream subscriber must be traced with the retry stream as its