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

[afiodorov]

Description

Problem (#303). With retry_on_idle_ms set and max_records=None (the default), FastStream's XREADGROUP sends no COUNT. When a backlog is waiting, one read puts every entry in the consumer's PEL, and FastStream handles them one by one. An entry's idle time starts at delivery, so an entry queued behind slow handlers goes idle past retry_on_idle_ms before it starts. The reclaimer then moves it to .retry, and the handler runs it twice: once from the local list, once from the retry stream. With 4 queued entries, a 1 s handler and retry_on_idle_ms=1000, main runs the handler 8 times: [1, 2, 2, 3, 3, 4, 4, 4].

Fix. When retry_on_idle_ms is set, batch=False and max_records is not set, max_records defaults to 1. The recursive .retry subscribe already forwards max_records, so it inherits it. An explicit max_records is kept, and batch=True is left alone.

Type of Change

• Bug fix (non-breaking change which fixes an issue)
• Documentation update

Related Issue

Fixes #303

Changes Made

• RedisTransport.subscribe: the max_records default described above, next to the other retry_on_idle_ms checks. The max_records docstring says why, and that an explicit value should stay at or below what the handler gets through within retry_on_idle_ms.
• docs/docs/sdk/redis-transport.md: one line under the retry guidelines.
• sdk/CHANGELOG.md: under Fixed, including the behaviour change.

Behaviour change

A subscription with retry_on_idle_ms now does one XREADGROUP per entry instead of one per backlog. For slow handlers, which is where the reclaimer matters, this costs nothing. Setups with small, fast messages that relied on the unbounded read for throughput can set max_records explicitly. Subscriptions without retry_on_idle_ms are unchanged.

Relation to #299

#299 already defaults max_records to 1 under renew_lease=True and renews prefetched entries. This PR makes the default apply to every retry_on_idle_ms subscription, so the lease isn't needed just to avoid this case. Whichever PR merges second needs a small rebase in subscribe(): #299's if not batch and max_records is None: max_records = 1 under renew_lease becomes redundant. (#299's scan_prefetched still matters for an explicit max_records > 1 and for max_workers > 1.)

Out of scope here, and listed as open questions in #303: making renew_lease the default with retry_on_idle_ms, a warning for an explicit max_records > 1 without the lease, and skipping live consumers via XINFO CONSUMERS.

Testing

Against a throwaway redis:7-alpine, FastStream 0.7.5, Python 3.12:

• test_retry_on_idle_ms_does_not_reclaim_entries_queued_behind_a_slow_handler (integration): 4 entries published before start, group_start="0", 1 s handler, retry_on_idle_ms=1000. It checks that 0.5 s in, only 1 entry is pending and 3 are still in the stream (lag), and that each entry is handled exactly once. On main it fails (4 pending; with the PEL checks skipped, the runs are [1, 2, 2, 3, 3, 4, 4, 4]).
• test_retry_on_idle_ms_defaults_max_records_to_one (unit): 1 on the main and the retry subscriber, an explicit max_records=10 kept on both, batch=True and no-reclaimer subscriptions unchanged.
• Full suite: 125 passed, 14 skipped (Kafka not available).
• ruff check, ruff format --check, mypy (locked versions) clean.

Changelog

• I have updated sdk/CHANGELOG.md under [Unreleased] section

Checklist

• My code follows the project's code style (make lint passes)
• My code is properly formatted (make format applied)
• I have added/updated tests that prove my fix/feature works
• I have added/updated documentation as needed
• All tests pass locally
• I have reviewed my own code
• My changes generate no new warnings
• My commit messages follow the Conventional Commits standard

🤖 Generated with Claude Code

[github-actions]

QualOps Code Quality Analysis

Status: ⚠️ FAILED - Critical or high severity issues found

Summary

• Total Issues: 3
• Critical: 0 🔴
• High: 1 🟠
• Medium: 1 🟡
• Low: 1 🟢
• Files Analyzed: 4

🟠 High Issues (1)

• sdk/eggai/transport/redis.py:636 - bug
  The new max_records=1 default (line 637) is applied after max_records is extracted from kwargs (line 529), but the retry subscribe call at line 816 passes max_records=max_records explicitly — meaning the retry stream subscriber correctly inherits the forced value. However, if a caller explicitly passes max_records=0 or another falsy-but-not-None value, the guard max_records is None on line 636 correctly skips the override. The real issue is that max_records=1 is set on the outer scope variable after extraction, so it is also used in the StreamSub construction at line 683 — meaning the main stream subscriber also gets max_records=1 forced, which is the intended behavior per the comment. This is consistent and correct.

🟡 Medium Issues (1)

• sdk/eggai/transport/redis.py:636 - performance
  Setting max_records=1 when retry_on_idle_ms is set forces one Redis round-trip per message, eliminating batching throughput even when the handler is fast.

📊 Full Report

View detailed report

---

Powered by QualOps

[afiodorov]

Re QualOps high issue (sdk/eggai/transport/redis.py:636): false positive. The finding itself concludes the behavior is "consistent and correct" — max_records=1 is meant to apply to both the main StreamSub and the retry subscribe, and an explicit caller value is left untouched.

[valdis]

Good catch 👍