From 1eaadc719a2e786407a918fb79da4fa729eae766 Mon Sep 17 00:00:00 2001 From: Tom Artiom Fiodorov Date: Thu, 1 Oct 2026 19:27:02 +0200 Subject: [PATCH] fix(redis): default max_records to 1 when retry_on_idle_ms is set Without a COUNT, one XREADGROUP hands the consumer every waiting entry. They all enter its PEL at once and are handled one by one, so an entry queued behind slow handlers goes idle past retry_on_idle_ms before it starts, and the reclaimer re-delivers it through .retry: the handler runs it twice and _retry_count climbs without a failure. Read one entry at a time when the reclaimer is on (not with batch=True). The retry subscription inherits it; an explicit max_records is kept. Fixes #303 Co-Authored-By: Claude Opus 5.5 --- docs/docs/sdk/redis-transport.md | 1 + sdk/CHANGELOG.md | 14 ++++ sdk/eggai/transport/redis.py | 15 +++- sdk/tests/test_redis.py | 120 +++++++++++++++++++++++++++++++ 4 files changed, 149 insertions(+), 1 deletion(-) diff --git a/docs/docs/sdk/redis-transport.md b/docs/docs/sdk/redis-transport.md index c6e385e4..2a97e068 100644 --- a/docs/docs/sdk/redis-transport.md +++ b/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) diff --git a/sdk/CHANGELOG.md b/sdk/CHANGELOG.md index afe3aa72..ca894f1e 100644 --- a/sdk/CHANGELOG.md +++ b/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 diff --git a/sdk/eggai/transport/redis.py b/sdk/eggai/transport/redis.py index 53d64216..a2c6550e 100644 --- a/sdk/eggai/transport/redis.py +++ b/sdk/eggai/transport/redis.py @@ -337,7 +337,11 @@ async def subscribe(self, channel: str, handler, **kwargs) -> Callable: 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 @@ async def subscribe(self, channel: str, handler, **kwargs) -> Callable: "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 diff --git a/sdk/tests/test_redis.py b/sdk/tests/test_redis.py index ad939ed0..d7e3d61d 100644 --- a/sdk/tests/test_redis.py +++ b/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