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