Redis: PEL reclaimer re-delivers entries a consumer has read but not started (max_records=None reads the whole backlog) #303

Summary

With retry_on_idle_ms set and max_records left at its default (None), the PEL reclaimer re-delivers entries that the consumer has already read but not started yet. The handler then runs them twice.

• RedisTransport.subscribe defaults max_records to None (sdk/eggai/transport/redis.py:525), so FastStream's XREADGROUP sends no COUNT (faststream/redis/subscriber/usecases/stream_subscriber.py, 0.7.5).
• When a backlog is waiting, one read hands the consumer every entry. All of them go into its PEL, and FastStream handles them one at a time (StreamSubscriber._get_msgs: for ...: await self.consume_one(msg)). The next read only happens when the whole list is done. With max_workers=N it's the same: the read is still unbounded, and the entries wait in a buffer of size N.
• An entry's PEL idle time starts when it is delivered, not when its handler starts. PendingReclaimerManager._reclaim_once treats "idle past retry_on_idle_ms" as "consumer died" and XCLAIMs the entry to .retry, even though it's only waiting for its turn in the local list.

Effects with slow handlers (LLM calls):

• duplicate handler runs, partly in parallel (local copy + .retry copy);
• _retry_count climbs while the entry is queued, so it can reach the DLQ without ever failing;
• one huge Redis reply for a large backlog (large payloads → memory);
• a second replica gets nothing, because the first one holds the whole backlog in its PEL.

It only happens when a backlog is waiting at read time: first start, consumer restart, a burst while the handler is busy. A steady trickle doesn't trigger it, because the blocking read wakes on the first entry and returns just that one.

Found in a downstream service (handlers making 20 s+ LLM calls), which now works around it with max_records=1.

Repro (main @ 3fb9a27, faststream 0.7.5, redis:7-alpine)

Publish 4 entries before the agent starts, then:

@agent.subscribe(
    channel=channel,
    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"])
    await asyncio.sleep(1.0)

• 0.5 s after start: XPENDING shows 4 pending, group lag is 0: everything is claimed by one consumer, though only one entry is being handled.
• Handler runs: [1, 2, 2, 3, 3, 4, 4, 4] (4 messages, 8 runs). Expected [1, 2, 3, 4].

With max_records=1: 1 pending, lag 3, and each entry runs once.

Relation to #299

#299 (lease renewal, renew_lease=True) covers this case when the lease is on: it defaults max_records to 1 and also renews prefetched entries. It is opt-in, so without renew_lease, the plain retry_on_idle_ms setup still has the bug.

Proposed fix

1. Bound the read whenever the reclaimer is on. If retry_on_idle_ms is set, batch=False and max_records is not set, default max_records to 1. The internal .retry subscription already forwards max_records, so it inherits it. An explicit max_records=N is kept. This fixes the duplicates, the oversized reply and the replica imbalance.

   It's a behaviour change for anyone who relied on the unbounded read for throughput with tiny, fast messages: one XREADGROUP per entry instead of one per backlog. They can set max_records explicitly. To be noted in the changelog.

2. Open questions (not in the first PR):

   • Should renew_lease (feat(redis): lease renewal for in-flight stream entries #299) become the default with retry_on_idle_ms? Then retry_on_idle_ms only has to cover crash detection, not the slowest handler. Today the docs say to set it above the worst-case handler time, so a dead consumer with a 300 s handler is noticed after ~6 min.
   • Should an explicit max_records > 1 with the reclaimer on (and no lease) log a warning?
   • Optional safety net: before claiming, check XINFO CONSUMERS (idle, Redis 7.2 inactive) and only reclaim from consumers that haven't read recently.