From 0b79c05a99a1011ec6599771a64d2c56424aa56c Mon Sep 17 00:00:00 2001 From: Tom Artiom Fiodorov Date: Thu, 24 Sep 2026 16:22:27 +0200 Subject: [PATCH 1/5] feat(redis): opt-in delete_on_ack to XDEL entries when they are acked Acked stream entries stay in Redis until MAXLEN trims them, and MAXLEN trims by count regardless of ack state. With delete_on_ack=True a subscription XACKs and XDELs each entry in one MULTI: after a successful handler run (filtered-out messages included), on the retry stream, and in the reclaimer when an entry moves to the retry stream or DLQ. Failing handlers leave the entry in the PEL as before. Opt-in because XDEL removes the entry for every consumer group; it assumes one group per stream. Co-Authored-By: Claude Opus 5.5 (1M context) --- sdk/CHANGELOG.md | 10 + sdk/eggai/transport/pending_reclaimer.py | 23 +- sdk/eggai/transport/redis.py | 82 ++++++++ sdk/tests/test_redis_delete_on_ack.py | 254 +++++++++++++++++++++++ 4 files changed, 364 insertions(+), 5 deletions(-) create mode 100644 sdk/tests/test_redis_delete_on_ack.py diff --git a/sdk/CHANGELOG.md b/sdk/CHANGELOG.md index aa6d1206..83690146 100644 --- a/sdk/CHANGELOG.md +++ b/sdk/CHANGELOG.md @@ -7,6 +7,16 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0 ## [Unreleased] +### Added + +- Redis `subscribe(..., delete_on_ack=True)`: every entry the subscription + acks is also `XDEL`'d (same `MULTI`), so consumed messages stop occupying + Redis memory instead of waiting for `MAXLEN` trimming. Covers handler + success (including filtered-out messages), the retry stream, and the + reclaimer's moves to the retry stream / DLQ; a failing handler's entry + stays in the PEL. Opt-in because it assumes a single consumer group per + stream: `XDEL` removes the entry for every group. + ### Fixed - MCP adapter reads tool schemas from fastmcp's own `Tool` instead of the diff --git a/sdk/eggai/transport/pending_reclaimer.py b/sdk/eggai/transport/pending_reclaimer.py index 2311c0ac..18026055 100644 --- a/sdk/eggai/transport/pending_reclaimer.py +++ b/sdk/eggai/transport/pending_reclaimer.py @@ -81,6 +81,10 @@ class ReclaimerConfig: # stream key no longer does. source_stream: str | None = None handler: str | None = None + # XDEL the original entry together with its XACK once it has been moved to the + # retry stream or DLQ (the subscription's delete_on_ack). Assumes this group is + # the only consumer of `stream`. + delete_on_ack: bool = False def _encode_envelope(headers: dict, body_bytes: bytes) -> bytes: @@ -263,7 +267,7 @@ class PendingReclaimerManager: 1. Pages through XPENDING to find entries idle longer than min_idle_ms. 2. XCLAIM them under a dedicated reclaimer consumer. 3. XADD the fields to retry_stream (a separate stream — avoids duplicates). - 4. XACK the original PEL entry. + 4. XACK the original PEL entry (plus XDEL with delete_on_ack). Delivery guarantee: at-least-once. XADD and XACK are not atomic; a crash between them will re-deliver the message on the next reclaim cycle. @@ -549,7 +553,7 @@ async def _reclaim_once(self, config: ReclaimerConfig) -> None: dlq_fields = dict(fields) dlq_fields[data_key] = _encode_body(headers, body) await self._xadd(config.dlq_stream, dlq_fields, config.dlq_max_len) - await self._client.xack(config.stream, config.group, msg_id) + await self._ack(config, msg_id) logger.warning( "Message %s has an unparseable envelope; moved to DLQ %s " "(retry count cannot be tracked)", @@ -558,7 +562,7 @@ async def _reclaim_once(self, config: ReclaimerConfig) -> None: ) await self._invoke_on_dlq(config, body, msg_id_str, 0) else: - await self._client.xack(config.stream, config.group, msg_id) + await self._ack(config, msg_id) logger.error( "Message %s has an unparseable envelope and no DLQ is " "configured; dropping it to avoid a retry-stream livelock", @@ -582,7 +586,7 @@ async def _reclaim_once(self, config: ReclaimerConfig) -> None: dlq_fields = dict(fields) dlq_fields[data_key] = _encode_body(headers, body) await self._xadd(config.dlq_stream, dlq_fields, config.dlq_max_len) - await self._client.xack(config.stream, config.group, msg_id) + await self._ack(config, msg_id) logger.warning( "Message %s exceeded max_retries=%d; moved to DLQ %s", msg_id_str, @@ -593,9 +597,18 @@ async def _reclaim_once(self, config: ReclaimerConfig) -> None: else: fields[data_key] = _encode_body(headers, body) await self._xadd(config.retry_stream, fields, config.max_len) - await self._client.xack(config.stream, config.group, msg_id) + await self._ack(config, msg_id) logger.debug("Reclaimed %s → %s", msg_id_str, config.retry_stream) + async def _ack(self, config: ReclaimerConfig, msg_id: Any) -> None: + if not config.delete_on_ack: + await self._client.xack(config.stream, config.group, msg_id) + return + async with self._client.pipeline(transaction=True) as pipe: + pipe.xack(config.stream, config.group, msg_id) + pipe.xdel(config.stream, msg_id) + await pipe.execute() + async def _invoke_on_dlq( self, config: ReclaimerConfig, diff --git a/sdk/eggai/transport/redis.py b/sdk/eggai/transport/redis.py index 6ee464bd..11498c2a 100644 --- a/sdk/eggai/transport/redis.py +++ b/sdk/eggai/transport/redis.py @@ -306,6 +306,16 @@ async def subscribe(self, channel: str, handler, **kwargs) -> Callable: Not allowed together with a consumer group: XREADGROUP with an explicit id only returns this consumer's own pending entries, never new ones. Use group_start instead. no_ack (bool, optional): Whether to skip acknowledgment of stream messages (default is False for durability). + delete_on_ack (bool, optional): Delete each stream entry (``XDEL``) at the moment it is acknowledged, + so a consumed message no longer occupies Redis memory (default False: acked entries stay in the + stream until ``MAXLEN`` trims them). Applies wherever this subscription acks: after the handler + returns (including messages skipped by a filter), on the SDK-managed retry stream, and when + the reclaimer moves a stuck entry to the retry stream or DLQ. A handler that raises is NOT + deleted, it stays in the PEL for retry. ``XACK`` and ``XDEL`` run in one ``MULTI``. + ASSUMES THIS IS THE ONLY CONSUMER GROUP ON THE STREAM: ``XDEL`` removes the entry for every + group, so another group (or a group-less ``XREAD`` subscriber) that has not read it yet never + will. Requires a consumer group; incompatible with ``no_ack=True`` and + ``ack_policy=AckPolicy.MANUAL``. On Redis >= 8.2 the same effect is ``XACKDEL``. ack_policy (AckPolicy, optional): Acknowledgment policy for message handling (default is AckPolicy.NACK_ON_ERROR). - NACK_ON_ERROR: Messages are NOT acknowledged on handler errors, allowing redelivery (recommended). - ACK: Messages are acknowledged regardless of handler success/failure. @@ -470,6 +480,25 @@ async def subscribe(self, channel: str, handler, **kwargs) -> Callable: group_start = kwargs.pop("group_start", "$") no_ack = kwargs.pop("no_ack", False) min_idle_time = kwargs.pop("min_idle_time", None) + delete_on_ack = kwargs.pop("delete_on_ack", False) + + if delete_on_ack: + if not group: + raise ValueError( + "delete_on_ack requires a consumer group: pass handler_id= or " + "group= (Agent.subscribe / Channel.subscribe set one automatically)." + ) + if no_ack: + raise ValueError( + "delete_on_ack is incompatible with no_ack=True: with no ack " + "there is no point at which the entry is known to be processed." + ) + if kwargs.get("ack_policy") == AckPolicy.MANUAL: + raise ValueError( + "delete_on_ack is incompatible with ack_policy=AckPolicy.MANUAL: " + "the SDK cannot tell when a manually-acked message is done. " + "Call XDEL yourself after msg.ack()." + ) if group and last_id != ">": raise ValueError( @@ -614,6 +643,12 @@ async def subscribe(self, channel: str, handler, **kwargs) -> Callable: # allowing them to be redelivered for retry ack_policy = kwargs.pop("ack_policy", AckPolicy.NACK_ON_ERROR) + # Outermost wrapper, so a message a filter skips is deleted too (it is + # acked either way). Applied per stream: the recursive retry subscribe + # below wraps its own handler with the retry group. + if delete_on_ack: + handler = self._wrap_delete_on_ack(handler, group) + # stream must be passed as keyword-only argument registered_handler = self.broker.subscriber( stream=stream_sub, ack_policy=ack_policy, **kwargs @@ -678,6 +713,7 @@ async def subscribe(self, channel: str, handler, **kwargs) -> Callable: source_stream=channel, handler=handler_suffix, dlq_max_len=dlq_max_len, + delete_on_ack=delete_on_ack, ) ) @@ -720,6 +756,7 @@ async def subscribe(self, channel: str, handler, **kwargs) -> Callable: last_id=">", no_ack=no_ack, ack_policy=ack_policy, + delete_on_ack=delete_on_ack, **kwargs, ) @@ -745,6 +782,7 @@ async def subscribe(self, channel: str, handler, **kwargs) -> Callable: source_stream=channel, handler=handler_suffix, dlq_max_len=dlq_max_len, + delete_on_ack=delete_on_ack, ) ) except Exception: @@ -840,6 +878,48 @@ async def _monitor_stream_groups(self) -> None: finally: await client.aclose() + def _wrap_delete_on_ack(self, handler: Callable, group: str) -> Callable: + """Wrap ``handler`` so a successful run XACKs and XDELs its entry. + + FastStream's own ack (XACK after the handler returns) stays in place and + becomes a harmless no-op on the already-deleted id. The raw stream message + (channel + ids, a list even for one entry, so batch mode works too) comes + from FastStream's per-message context. functools.wraps keeps the user's + signature visible to fast_depends, same as the tracing wrapper. + """ + import functools + import inspect + + @functools.wraps(handler) + async def delete_on_ack_handler(*args, **kwargs): + result = handler(*args, **kwargs) + if inspect.isawaitable(result): + result = await result + + message = self.broker.context.get_local("message") + if message is None: + return result + raw = message.raw_message + channel, ids = raw["channel"], raw["message_ids"] + try: + client = self.broker.config.connection.client + async with client.pipeline(transaction=True) as pipe: + pipe.xack(channel, group, *ids) + pipe.xdel(channel, *ids) + await pipe.execute() + except Exception: + # The handler already succeeded: raising here would NACK it and + # re-run it. Leave the entry for FastStream's XACK / MAXLEN. + logger.warning( + "delete_on_ack: XACK+XDEL failed for %s %s; entry left in stream", + channel, + ids, + exc_info=True, + ) + return result + + return delete_on_ack_handler + def _setup_reclaimer( self, *, @@ -858,6 +938,7 @@ def _setup_reclaimer( source_stream: str | None = None, handler: str | None = None, dlq_max_len: int | None = None, + delete_on_ack: bool = False, ) -> tuple[str, str, str]: if self._reclaimer_manager is None: self._reclaimer_manager = PendingReclaimerManager( @@ -881,5 +962,6 @@ def _setup_reclaimer( source_stream=source_stream, handler=handler, dlq_max_len=dlq_max_len, + delete_on_ack=delete_on_ack, ) ) diff --git a/sdk/tests/test_redis_delete_on_ack.py b/sdk/tests/test_redis_delete_on_ack.py new file mode 100644 index 00000000..407b5d1c --- /dev/null +++ b/sdk/tests/test_redis_delete_on_ack.py @@ -0,0 +1,254 @@ +""" +delete_on_ack: acked stream entries are XDEL'd, so consumed messages stop +occupying Redis memory. Integration tests against a real Redis at localhost:6379. +""" + +import asyncio +import uuid + +import pytest +import pytest_asyncio +import redis.asyncio as redis +from faststream import AckPolicy + +from eggai import Agent, Channel +from eggai.transport import RedisTransport + + +async def _wait_for(predicate, timeout=10.0, interval=0.1): + deadline = asyncio.get_running_loop().time() + timeout + while True: + if await predicate(): + return + if asyncio.get_running_loop().time() > deadline: + raise AssertionError("condition not met before timeout") + await asyncio.sleep(interval) + + +def _pending(info): + return info.get("pending", 0) if isinstance(info, dict) else info[0] + + +@pytest_asyncio.fixture +async def redis_client(): + client = redis.Redis(host="localhost", port=6379, decode_responses=True) + yield client + await client.aclose() + + +def _names(prefix): + test_id = uuid.uuid4().hex[:8] + agent_name = f"{prefix}-{test_id}" + channel_name = f"{prefix}-ch-{test_id}" + return agent_name, channel_name, f"eggai.{channel_name}" + + +@pytest.mark.asyncio +async def test_delete_on_ack_removes_handled_entries(redis_client): + agent_name, channel_name, stream = _names("doa-ok") + transport = RedisTransport() + agent = Agent(agent_name, transport=transport) + channel = Channel(channel_name, transport=transport) + group = f"{agent_name}-handler-1" + + seen = [] + + @agent.subscribe(channel=channel, delete_on_ack=True) + async def handler(message): + seen.append(message["n"]) + + await agent.start() + try: + for n in range(5): + await channel.publish({"type": "t", "n": n}) + await _wait_for(lambda: _async(len(seen) == 5)) + await _wait_for(lambda: _xlen_is(redis_client, stream, 0)) + assert _pending(await redis_client.xpending(stream, group)) == 0 + finally: + await agent.stop() + assert sorted(seen) == [0, 1, 2, 3, 4] + + +@pytest.mark.asyncio +async def test_without_delete_on_ack_entries_stay(redis_client): + """Contrast: default behaviour keeps acked entries in the stream.""" + agent_name, channel_name, stream = _names("doa-off") + transport = RedisTransport() + agent = Agent(agent_name, transport=transport) + channel = Channel(channel_name, transport=transport) + + seen = [] + + @agent.subscribe(channel=channel) + async def handler(message): + seen.append(message["n"]) + + await agent.start() + try: + for n in range(3): + await channel.publish({"type": "t", "n": n}) + await _wait_for(lambda: _async(len(seen) == 3)) + await asyncio.sleep(0.3) + assert await redis_client.xlen(stream) == 3 + finally: + await agent.stop() + + +@pytest.mark.asyncio +async def test_delete_on_ack_keeps_failed_entry_in_pel(redis_client): + agent_name, channel_name, stream = _names("doa-fail") + transport = RedisTransport() + agent = Agent(agent_name, transport=transport) + channel = Channel(channel_name, transport=transport) + group = f"{agent_name}-handler-1" + + called = asyncio.Event() + + @agent.subscribe(channel=channel, delete_on_ack=True) + async def handler(message): + called.set() + raise RuntimeError("boom") + + await agent.start() + try: + await channel.publish({"type": "t"}) + await asyncio.wait_for(called.wait(), timeout=10.0) + await asyncio.sleep(0.3) + assert await redis_client.xlen(stream) == 1 + assert _pending(await redis_client.xpending(stream, group)) == 1 + finally: + await agent.stop() + + +@pytest.mark.asyncio +async def test_delete_on_ack_deletes_filtered_out_entries(redis_client): + agent_name, channel_name, stream = _names("doa-filter") + transport = RedisTransport() + agent = Agent(agent_name, transport=transport) + channel = Channel(channel_name, transport=transport) + + seen = [] + + @agent.subscribe( + channel=channel, + delete_on_ack=True, + filter_by_message=lambda m: m.get("keep"), + ) + async def handler(message): + seen.append(message["n"]) + + await agent.start() + try: + await channel.publish({"type": "t", "n": 1, "keep": False}) + await channel.publish({"type": "t", "n": 2, "keep": True}) + await _wait_for(lambda: _async(seen == [2])) + await _wait_for(lambda: _xlen_is(redis_client, stream, 0)) + finally: + await agent.stop() + + +@pytest.mark.asyncio +async def test_delete_on_ack_through_retry_stream(redis_client): + """Fail once, succeed on retry: main and retry streams both end empty.""" + agent_name, channel_name, stream = _names("doa-retry") + group = f"{agent_name}-handler-1" + retry_stream = f"{stream}.{group}.retry" + transport = RedisTransport() + agent = Agent(agent_name, transport=transport) + channel = Channel(channel_name, transport=transport) + + calls = 0 + done = asyncio.Event() + + @agent.subscribe( + channel=channel, + delete_on_ack=True, + retry_on_idle_ms=300, + retry_reclaim_interval_s=0.5, + ) + async def handler(message): + nonlocal calls + calls += 1 + if calls == 1: + raise RuntimeError("transient") + done.set() + + await agent.start() + try: + await channel.publish({"type": "t"}) + await asyncio.wait_for(done.wait(), timeout=10.0) + await _wait_for(lambda: _xlen_is(redis_client, stream, 0)) + await _wait_for(lambda: _xlen_is(redis_client, retry_stream, 0)) + finally: + await agent.stop() + assert calls == 2 + + +@pytest.mark.asyncio +async def test_delete_on_ack_through_dlq(redis_client): + """Always failing: main and retry streams end empty, the DLQ keeps the entry.""" + agent_name, channel_name, stream = _names("doa-dlq") + group = f"{agent_name}-handler-1" + retry_stream = f"{stream}.{group}.retry" + dlq_stream = f"{stream}.{group}.dlq" + transport = RedisTransport() + agent = Agent(agent_name, transport=transport) + channel = Channel(channel_name, transport=transport) + + @agent.subscribe( + channel=channel, + delete_on_ack=True, + retry_on_idle_ms=300, + retry_reclaim_interval_s=0.5, + max_retries=1, + ) + async def handler(message): + raise RuntimeError("permanent") + + await agent.start() + try: + await channel.publish({"type": "t"}) + await _wait_for(lambda: _xlen_is(redis_client, dlq_stream, 1), timeout=15.0) + await _wait_for(lambda: _xlen_is(redis_client, stream, 0)) + await _wait_for(lambda: _xlen_is(redis_client, retry_stream, 0)) + finally: + await agent.stop() + + +@pytest.mark.asyncio +@pytest.mark.parametrize( + "extra, match", + [ + ({"no_ack": True}, "no_ack"), + ({"ack_policy": AckPolicy.MANUAL}, "MANUAL"), + ], +) +async def test_delete_on_ack_rejects_incompatible_options(extra, match): + transport = RedisTransport() + + async def handler(message): + pass + + with pytest.raises(ValueError, match=match): + await transport.subscribe( + "doa-invalid", handler, handler_id="h", delete_on_ack=True, **extra + ) + + +@pytest.mark.asyncio +async def test_delete_on_ack_requires_group(): + transport = RedisTransport() + + async def handler(message): + pass + + with pytest.raises(ValueError, match="consumer group"): + await transport.subscribe("doa-nogroup", handler, delete_on_ack=True) + + +async def _async(value): + return value + + +async def _xlen_is(client, stream, n): + return await client.xlen(stream) == n From c83192e0d102ae9dfc8f3a6f22eac2d1f8105211 Mon Sep 17 00:00:00 2001 From: Tom Artiom Fiodorov Date: Thu, 24 Sep 2026 16:50:09 +0200 Subject: [PATCH 2/5] fix(redis): address delete_on_ack review findings - ACK / ACK_FIRST ack a failed run, so delete_on_ack now deletes it too; before, it was acked but never trimmed. - Sync handlers run in a worker thread again. The async wrapper had made FastStream await them on the event loop. - One xack_del() helper shared by the handler wrapper and the reclaimer, so the Redis 8.2 XACKDEL swap is a single change. - A poison entry dropped with no DLQ is only XACKed, keeping the sole copy readable with XRANGE. - Document delete_on_ack in the Redis transport docs and the SDK README. Co-Authored-By: Claude Opus 5.5 (1M context) --- docs/docs/sdk/redis-transport.md | 14 ++++ sdk/CHANGELOG.md | 8 ++- sdk/README.md | 9 +++ sdk/eggai/transport/pending_reclaimer.py | 25 +++++-- sdk/eggai/transport/redis.py | 74 +++++++++++++------- sdk/tests/test_redis_delete_on_ack.py | 86 ++++++++++++++++++++++++ 6 files changed, 181 insertions(+), 35 deletions(-) diff --git a/docs/docs/sdk/redis-transport.md b/docs/docs/sdk/redis-transport.md index d0e6b860..c656ed5f 100644 --- a/docs/docs/sdk/redis-transport.md +++ b/docs/docs/sdk/redis-transport.md @@ -281,6 +281,20 @@ Two rules make a DLQ entry safe to *consume*, not just inspect. `_retry_count` i Validation: `dlq_channel` requires `retry_on_idle_ms` and a non-`None` `max_retries` (with `max_retries=None` there is no DLQ to redirect), must differ from the subscribed channel, and must not end in `.retry` (every `.retry` stream is auto-consumed by some handler, so dead-lettering into one would feed that handler's retry loop). All of it fails at `subscribe()` time, before anything is registered on the broker. +### Freeing Memory with `delete_on_ack` + +Redis keeps a stream entry after it is acked; only `MAXLEN` trims it, and `MAXLEN` trims by count whether or not an entry was processed. With `delete_on_ack=True` a subscription `XDEL`s every entry at the moment it acks it (`XACK` + `XDEL` in one `MULTI`), so a stream only holds in-flight messages: + +```python +@agent.subscribe(channel=emails, delete_on_ack=True) +async def handle_email(message): + ... +``` + +It applies wherever the subscription acks: after the handler returns (including messages a filter skips), on the SDK retry stream, and when the reclaimer moves a stuck entry to `.retry` or the DLQ. Under the default `NACK_ON_ERROR` a failing handler's entry stays in the PEL for retry; under `AckPolicy.ACK` / `ACK_FIRST`, which ack failures too, it is deleted too. DLQ entries are never deleted by this option, and a poison entry dropped with no DLQ configured is only acked so it stays readable with `XRANGE`. + +**Only one consumer group per stream.** `XDEL` removes the entry for every group, so a second group (or a group-less subscriber) that has not read an entry yet never will. Competing workers inside one group are fine. For fan-out to several groups, keep `delete_on_ack` off and bound memory with `max_len`. Requires a consumer group; rejected with `no_ack=True` and `AckPolicy.MANUAL`. On Redis >= 8.2 the equivalent command is `XACKDEL`. + ### Tuning the Reclaimer ```python diff --git a/sdk/CHANGELOG.md b/sdk/CHANGELOG.md index 83690146..9f2ed286 100644 --- a/sdk/CHANGELOG.md +++ b/sdk/CHANGELOG.md @@ -13,9 +13,11 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0 acks is also `XDEL`'d (same `MULTI`), so consumed messages stop occupying Redis memory instead of waiting for `MAXLEN` trimming. Covers handler success (including filtered-out messages), the retry stream, and the - reclaimer's moves to the retry stream / DLQ; a failing handler's entry - stays in the PEL. Opt-in because it assumes a single consumer group per - stream: `XDEL` removes the entry for every group. + reclaimer's moves to the retry stream / DLQ. A failing handler's entry + stays in the PEL under the default `NACK_ON_ERROR`, and is deleted under + `AckPolicy.ACK` / `ACK_FIRST` (which ack it anyway). Opt-in because it + assumes a single consumer group per stream: `XDEL` removes the entry for + every group. ### Fixed diff --git a/sdk/README.md b/sdk/README.md index 534a7292..7f4a3e8d 100644 --- a/sdk/README.md +++ b/sdk/README.md @@ -241,6 +241,15 @@ A poison entry (unparseable envelope) is no longer copied to the DLQ verbatim: i wrapped in a fresh envelope whose body holds the metadata above plus the original bytes as `_dlq_raw_b64`, so a DLQ subscriber always receives a decodable JSON object. +**Freeing memory with `delete_on_ack`:** acked stream entries stay in Redis until `MAXLEN` +trims them, and `MAXLEN` trims by count regardless of whether an entry was processed. +`subscribe(..., delete_on_ack=True)` `XDEL`s each entry when it is acked (`XACK` + `XDEL` +in one `MULTI`): after the handler returns, on the retry stream, and when the reclaimer +moves an entry to `.retry` or the DLQ. A failing handler's entry still stays in the PEL +for retry (under the default `NACK_ON_ERROR`). **It assumes one consumer group per +stream:** `XDEL` removes the entry for every group, so don't use it on a stream that +fans out to several groups. + **Retry backoff:** By default retries fire at a constant cadence equal to `retry_on_idle_ms`. Set `retry_backoff_multiplier > 1.0` to back off exponentially: a message is treated as due for reclaim once it has been idle for diff --git a/sdk/eggai/transport/pending_reclaimer.py b/sdk/eggai/transport/pending_reclaimer.py index 18026055..fb4973d3 100644 --- a/sdk/eggai/transport/pending_reclaimer.py +++ b/sdk/eggai/transport/pending_reclaimer.py @@ -87,6 +87,18 @@ class ReclaimerConfig: delete_on_ack: bool = False +async def xack_del(client: Any, stream: str, group: str, *ids: Any) -> None: + """XACK and XDEL ``ids`` in one MULTI (the ``delete_on_ack`` primitive). + + Both commands touch the same key, so this is safe on Redis Cluster / + Enterprise too. On Redis >= 8.2 this is ``XACKDEL``; swap it here. + """ + async with client.pipeline(transaction=True) as pipe: + pipe.xack(stream, group, *ids) + pipe.xdel(stream, *ids) + await pipe.execute() + + def _encode_envelope(headers: dict, body_bytes: bytes) -> bytes: """Build a FastStream BinaryMessageFormatV1 envelope from headers + JSON body. @@ -562,7 +574,9 @@ async def _reclaim_once(self, config: ReclaimerConfig) -> None: ) await self._invoke_on_dlq(config, body, msg_id_str, 0) else: - await self._ack(config, msg_id) + # Plain XACK even with delete_on_ack: with no DLQ this entry + # is the only copy, keep it for XRANGE forensics. + await self._client.xack(config.stream, config.group, msg_id) logger.error( "Message %s has an unparseable envelope and no DLQ is " "configured; dropping it to avoid a retry-stream livelock", @@ -601,13 +615,10 @@ async def _reclaim_once(self, config: ReclaimerConfig) -> None: logger.debug("Reclaimed %s → %s", msg_id_str, config.retry_stream) async def _ack(self, config: ReclaimerConfig, msg_id: Any) -> None: - if not config.delete_on_ack: + if config.delete_on_ack: + await xack_del(self._client, config.stream, config.group, msg_id) + else: await self._client.xack(config.stream, config.group, msg_id) - return - async with self._client.pipeline(transaction=True) as pipe: - pipe.xack(config.stream, config.group, msg_id) - pipe.xdel(config.stream, msg_id) - await pipe.execute() async def _invoke_on_dlq( self, diff --git a/sdk/eggai/transport/redis.py b/sdk/eggai/transport/redis.py index 11498c2a..30382184 100644 --- a/sdk/eggai/transport/redis.py +++ b/sdk/eggai/transport/redis.py @@ -310,8 +310,11 @@ async def subscribe(self, channel: str, handler, **kwargs) -> Callable: so a consumed message no longer occupies Redis memory (default False: acked entries stay in the stream until ``MAXLEN`` trims them). Applies wherever this subscription acks: after the handler returns (including messages skipped by a filter), on the SDK-managed retry stream, and when - the reclaimer moves a stuck entry to the retry stream or DLQ. A handler that raises is NOT - deleted, it stays in the PEL for retry. ``XACK`` and ``XDEL`` run in one ``MULTI``. + the reclaimer moves a stuck entry to the retry stream or DLQ. Under the default + ``NACK_ON_ERROR`` a handler that raises is NOT deleted, it stays in the PEL for retry; under + ``AckPolicy.ACK`` / ``ACK_FIRST`` (which ack failures too) it is deleted as well. A poison entry + dropped with no DLQ configured is only acked, so it stays readable for ``XRANGE``. ``XACK`` and + ``XDEL`` run in one ``MULTI``. ASSUMES THIS IS THE ONLY CONSUMER GROUP ON THE STREAM: ``XDEL`` removes the entry for every group, so another group (or a group-less ``XREAD`` subscriber) that has not read it yet never will. Requires a consumer group; incompatible with ``no_ack=True`` and @@ -644,10 +647,10 @@ async def subscribe(self, channel: str, handler, **kwargs) -> Callable: ack_policy = kwargs.pop("ack_policy", AckPolicy.NACK_ON_ERROR) # Outermost wrapper, so a message a filter skips is deleted too (it is - # acked either way). Applied per stream: the recursive retry subscribe + # acked either way). Deletes exactly where ack_policy acks. Applied per stream: the recursive retry subscribe # below wraps its own handler with the retry group. if delete_on_ack: - handler = self._wrap_delete_on_ack(handler, group) + handler = self._wrap_delete_on_ack(handler, group, ack_policy) # stream must be passed as keyword-only argument registered_handler = self.broker.subscriber( @@ -878,37 +881,43 @@ async def _monitor_stream_groups(self) -> None: finally: await client.aclose() - def _wrap_delete_on_ack(self, handler: Callable, group: str) -> Callable: - """Wrap ``handler`` so a successful run XACKs and XDELs its entry. - - FastStream's own ack (XACK after the handler returns) stays in place and - becomes a harmless no-op on the already-deleted id. The raw stream message - (channel + ids, a list even for one entry, so batch mode works too) comes - from FastStream's per-message context. functools.wraps keeps the user's - signature visible to fast_depends, same as the tracing wrapper. + def _wrap_delete_on_ack( + self, handler: Callable, group: str, ack_policy: AckPolicy + ) -> Callable: + """Wrap ``handler`` so every entry FastStream acks is also XDEL'd. + + Under the default NACK_ON_ERROR (and REJECT_ON_ERROR, a no-op on Redis) + only a successful run acks, so only a successful run deletes; a failure + stays in the PEL. ACK / ACK_FIRST ack a failed run too, so for them a + failure is deleted as well, otherwise it would be acked yet never + trimmed. FastStream's own XACK still runs afterwards, a harmless no-op on + the deleted id. The raw stream message (channel + ids, a list even for + one entry, so batch mode works) comes from FastStream's per-message + context. functools.wraps keeps the user's signature visible to + fast_depends, same as the tracing wrapper. A sync handler is sent to a + worker thread, as FastStream would have done without this wrapper. """ import functools - import inspect - @functools.wraps(handler) - async def delete_on_ack_handler(*args, **kwargs): - result = handler(*args, **kwargs) - if inspect.isawaitable(result): - result = await result + import anyio + + from eggai.transport.pending_reclaimer import xack_del + delete_on_error = ack_policy in (AckPolicy.ACK, AckPolicy.ACK_FIRST) + is_async = asyncio.iscoroutinefunction(handler) + + async def delete_current() -> None: message = self.broker.context.get_local("message") if message is None: - return result + return raw = message.raw_message channel, ids = raw["channel"], raw["message_ids"] try: - client = self.broker.config.connection.client - async with client.pipeline(transaction=True) as pipe: - pipe.xack(channel, group, *ids) - pipe.xdel(channel, *ids) - await pipe.execute() + await xack_del( + self.broker.config.connection.client, channel, group, *ids + ) except Exception: - # The handler already succeeded: raising here would NACK it and + # Raising here would NACK a handler that already succeeded and # re-run it. Leave the entry for FastStream's XACK / MAXLEN. logger.warning( "delete_on_ack: XACK+XDEL failed for %s %s; entry left in stream", @@ -916,6 +925,21 @@ async def delete_on_ack_handler(*args, **kwargs): ids, exc_info=True, ) + + @functools.wraps(handler) + async def delete_on_ack_handler(*args, **kwargs): + try: + if is_async: + result = await handler(*args, **kwargs) + else: + result = await anyio.to_thread.run_sync( + functools.partial(handler, *args, **kwargs) + ) + except Exception: + if delete_on_error: + await delete_current() + raise + await delete_current() return result return delete_on_ack_handler diff --git a/sdk/tests/test_redis_delete_on_ack.py b/sdk/tests/test_redis_delete_on_ack.py index 407b5d1c..3ed36f30 100644 --- a/sdk/tests/test_redis_delete_on_ack.py +++ b/sdk/tests/test_redis_delete_on_ack.py @@ -4,6 +4,7 @@ """ import asyncio +import threading import uuid import pytest @@ -215,6 +216,91 @@ async def handler(message): await agent.stop() +@pytest.mark.asyncio +async def test_delete_on_ack_with_ack_policy_ack_deletes_failures(redis_client): + """AckPolicy.ACK acks a failed run too, so delete_on_ack must delete it too.""" + agent_name, channel_name, stream = _names("doa-ackpol") + transport = RedisTransport() + agent = Agent(agent_name, transport=transport) + channel = Channel(channel_name, transport=transport) + group = f"{agent_name}-handler-1" + + called = asyncio.Event() + + @agent.subscribe(channel=channel, delete_on_ack=True, ack_policy=AckPolicy.ACK) + async def handler(message): + called.set() + raise RuntimeError("boom") + + await agent.start() + try: + await channel.publish({"type": "t"}) + await asyncio.wait_for(called.wait(), timeout=10.0) + await _wait_for(lambda: _xlen_is(redis_client, stream, 0)) + assert _pending(await redis_client.xpending(stream, group)) == 0 + finally: + await agent.stop() + + +@pytest.mark.asyncio +async def test_delete_on_ack_runs_sync_handler_off_the_event_loop(redis_client): + agent_name, channel_name, stream = _names("doa-sync") + transport = RedisTransport() + agent = Agent(agent_name, transport=transport) + channel = Channel(channel_name, transport=transport) + + loop_thread = threading.get_ident() + handler_threads = [] + + @agent.subscribe(channel=channel, delete_on_ack=True) + def handler(message): + handler_threads.append(threading.get_ident()) + + await agent.start() + try: + await channel.publish({"type": "t"}) + await _wait_for(lambda: _async(len(handler_threads) == 1)) + await _wait_for(lambda: _xlen_is(redis_client, stream, 0)) + finally: + await agent.stop() + assert handler_threads[0] != loop_thread + + +@pytest.mark.asyncio +async def test_delete_on_ack_keeps_poison_entry_when_no_dlq(redis_client): + """With no DLQ a dropped poison entry is the only copy: ack it, don't XDEL it.""" + agent_name, channel_name, stream = _names("doa-poison") + group = f"{agent_name}-handler-1" + transport = RedisTransport() + agent = Agent(agent_name, transport=transport) + channel = Channel(channel_name, transport=transport) + + @agent.subscribe( + channel=channel, + delete_on_ack=True, + retry_on_idle_ms=300, + retry_reclaim_interval_s=0.5, + max_retries=None, + ) + async def handler(message): + raise RuntimeError("never succeeds") + + await agent.start() + try: + # No __data__ field: the reclaimer cannot parse it (poison). + await redis_client.xadd(stream, {"garbage": "x"}) + + async def acked(): + return _pending(await redis_client.xpending(stream, group)) == 0 and ( + await redis_client.xlen(stream) == 1 + ) + + await asyncio.sleep(0.5) + await _wait_for(acked, timeout=15.0) + finally: + await agent.stop() + + @pytest.mark.asyncio @pytest.mark.parametrize( "extra, match", From 1701feb40c236c8c49d6b4be1edd9c09e3900a96 Mon Sep 17 00:00:00 2001 From: Tom Artiom Fiodorov Date: Thu, 24 Sep 2026 17:00:28 +0200 Subject: [PATCH 3/5] fix(redis): guard delete_on_ack against other consumer groups - connect() raises RuntimeError when a delete_on_ack stream already has another consumer group, before any entry is consumed. The group monitor logs an error (once per set) for a group that joins later. - An object with an async __call__ is awaited, not sent to a thread where its coroutine would be dropped. - The raw_message lookup sits inside the try, so no failure after a successful handler can NACK it into a re-run. Co-Authored-By: Claude Opus 5.5 (1M context) --- docs/docs/sdk/redis-transport.md | 2 +- sdk/CHANGELOG.md | 3 +- sdk/README.md | 3 +- sdk/eggai/transport/redis.py | 85 ++++++++++++++++++++++++--- sdk/tests/test_redis_delete_on_ack.py | 85 +++++++++++++++++++++++++++ 5 files changed, 166 insertions(+), 12 deletions(-) diff --git a/docs/docs/sdk/redis-transport.md b/docs/docs/sdk/redis-transport.md index c656ed5f..7e27cf0e 100644 --- a/docs/docs/sdk/redis-transport.md +++ b/docs/docs/sdk/redis-transport.md @@ -293,7 +293,7 @@ async def handle_email(message): It applies wherever the subscription acks: after the handler returns (including messages a filter skips), on the SDK retry stream, and when the reclaimer moves a stuck entry to `.retry` or the DLQ. Under the default `NACK_ON_ERROR` a failing handler's entry stays in the PEL for retry; under `AckPolicy.ACK` / `ACK_FIRST`, which ack failures too, it is deleted too. DLQ entries are never deleted by this option, and a poison entry dropped with no DLQ configured is only acked so it stays readable with `XRANGE`. -**Only one consumer group per stream.** `XDEL` removes the entry for every group, so a second group (or a group-less subscriber) that has not read an entry yet never will. Competing workers inside one group are fine. For fan-out to several groups, keep `delete_on_ack` off and bound memory with `max_len`. Requires a consumer group; rejected with `no_ack=True` and `AckPolicy.MANUAL`. On Redis >= 8.2 the equivalent command is `XACKDEL`. +**Only one consumer group per stream.** `XDEL` removes the entry for every group, so a second group (or a group-less subscriber) that has not read an entry yet never will. Competing workers inside one group are fine. `connect()` checks this: if another group already exists on the stream it raises `RuntimeError` before anything is consumed, and the group monitor logs an error (once) if one appears later. For fan-out to several groups, keep `delete_on_ack` off and bound memory with `max_len`. Requires a consumer group; rejected with `no_ack=True` and `AckPolicy.MANUAL`. On Redis >= 8.2 the equivalent command is `XACKDEL`. ### Tuning the Reclaimer diff --git a/sdk/CHANGELOG.md b/sdk/CHANGELOG.md index 9f2ed286..3eb81bd8 100644 --- a/sdk/CHANGELOG.md +++ b/sdk/CHANGELOG.md @@ -17,7 +17,8 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0 stays in the PEL under the default `NACK_ON_ERROR`, and is deleted under `AckPolicy.ACK` / `ACK_FIRST` (which ack it anyway). Opt-in because it assumes a single consumer group per stream: `XDEL` removes the entry for - every group. + every group. `connect()` raises `RuntimeError` if another group already + reads the stream, and the group monitor logs an error if one joins later. ### Fixed diff --git a/sdk/README.md b/sdk/README.md index 7f4a3e8d..8234d4cb 100644 --- a/sdk/README.md +++ b/sdk/README.md @@ -248,7 +248,8 @@ in one `MULTI`): after the handler returns, on the retry stream, and when the re moves an entry to `.retry` or the DLQ. A failing handler's entry still stays in the PEL for retry (under the default `NACK_ON_ERROR`). **It assumes one consumer group per stream:** `XDEL` removes the entry for every group, so don't use it on a stream that -fans out to several groups. +fans out to several groups. `connect()` raises if another group already reads the stream, +and the group monitor logs an error if one joins later. **Retry backoff:** By default retries fire at a constant cadence equal to `retry_on_idle_ms`. Set `retry_backoff_multiplier > 1.0` to back off exponentially: a diff --git a/sdk/eggai/transport/redis.py b/sdk/eggai/transport/redis.py index 30382184..2fc8f538 100644 --- a/sdk/eggai/transport/redis.py +++ b/sdk/eggai/transport/redis.py @@ -184,6 +184,11 @@ def __init__( self._stream_subscriptions: set[_StreamGroupInfo] = set() self._group_monitor_task: asyncio.Task | None = None self._group_monitor_interval_s: float = group_monitor_interval_s + # (stream_key, group) of every delete_on_ack subscription, checked for + # other consumer groups on the same stream (XDEL would starve them). + self._delete_on_ack_groups: set[tuple[str, str]] = set() + # Foreign-group sets already reported by the monitor, so it logs once. + self._reported_foreign_groups: set[tuple[str, frozenset[str]]] = set() async def connect(self): """ @@ -317,7 +322,9 @@ async def subscribe(self, channel: str, handler, **kwargs) -> Callable: ``XDEL`` run in one ``MULTI``. ASSUMES THIS IS THE ONLY CONSUMER GROUP ON THE STREAM: ``XDEL`` removes the entry for every group, so another group (or a group-less ``XREAD`` subscriber) that has not read it yet never - will. Requires a consumer group; incompatible with ``no_ack=True`` and + will. Checked at ``connect()``: another group already on the stream raises ``RuntimeError`` + before anything is consumed, and the group monitor logs an error for one that appears + later. Requires a consumer group; incompatible with ``no_ack=True`` and ``ack_policy=AckPolicy.MANUAL``. On Redis >= 8.2 the same effect is ``XACKDEL``. ack_policy (AckPolicy, optional): Acknowledgment policy for message handling (default is AckPolicy.NACK_ON_ERROR). - NACK_ON_ERROR: Messages are NOT acknowledged on handler errors, allowing redelivery (recommended). @@ -651,6 +658,7 @@ async def subscribe(self, channel: str, handler, **kwargs) -> Callable: # below wraps its own handler with the retry group. if delete_on_ack: handler = self._wrap_delete_on_ack(handler, group, ack_policy) + self._delete_on_ack_groups.add((channel, group)) # stream must be passed as keyword-only argument registered_handler = self.broker.subscriber( @@ -793,6 +801,8 @@ async def subscribe(self, channel: str, handler, **kwargs) -> Callable: if main_sub_info is not None: self._stream_subscriptions.discard(main_sub_info) self._stream_subscriptions.discard(retry_sub_info) + self._delete_on_ack_groups.discard((channel, group)) + self._delete_on_ack_groups.discard((retry_stream, retry_handler_id)) if self._reclaimer_manager is not None: for key in added_reclaimer_keys: self._reclaimer_manager.discard(key) @@ -823,9 +833,30 @@ async def _create_groups(self) -> None: except ResponseError as e: if "BUSYGROUP" not in str(e): raise + # Runs before broker.start(), so a violation stops startup before + # any entry is consumed (and deleted). + for stream_key, group in sorted(self._delete_on_ack_groups): + foreign = await self._foreign_groups(client, stream_key, group) + if foreign: + raise RuntimeError( + f"delete_on_ack on stream {stream_key!r} (group {group!r}) " + f"but other consumer groups also read it: {foreign}. XDEL " + "would delete entries they have not read yet. Drop " + "delete_on_ack or remove the other groups " + "(XGROUP DESTROY)." + ) finally: await client.aclose() + @staticmethod + async def _foreign_groups(client: Any, stream_key: str, group: str) -> list[str]: + """Names of consumer groups on ``stream_key`` other than ``group``.""" + try: + groups = await client.xinfo_groups(stream_key) + except ResponseError: + return [] # stream does not exist (yet): nobody to starve + return sorted(g["name"] for g in groups if g["name"] != group) + async def _monitor_stream_groups(self) -> None: """Periodically ensure all registered stream consumer groups exist. @@ -876,11 +907,36 @@ async def _monitor_stream_groups(self) -> None: info.stream_key, e, ) + await self._check_delete_on_ack_groups(client) except asyncio.CancelledError: pass finally: await client.aclose() + async def _check_delete_on_ack_groups(self, client: Any) -> None: + """Log (once per set) a consumer group that joined a delete_on_ack + stream after startup. The startup check in _create_groups raises; here + the consumers are already running, so the monitor can only report.""" + for stream_key, group in list(self._delete_on_ack_groups): + try: + foreign = await self._foreign_groups(client, stream_key, group) + except Exception as e: # connection blips: next cycle retries + logger.debug( + "delete_on_ack group check failed on %s: %s", stream_key, e + ) + continue + key = (stream_key, frozenset(foreign)) + if foreign and key not in self._reported_foreign_groups: + self._reported_foreign_groups.add(key) + logger.error( + "delete_on_ack on stream %s (group %s) but other consumer " + "groups now read it too: %s. They will miss entries this " + "group deletes first.", + stream_key, + group, + foreign, + ) + def _wrap_delete_on_ack( self, handler: Callable, group: str, ack_policy: AckPolicy ) -> Callable: @@ -898,27 +954,34 @@ def _wrap_delete_on_ack( worker thread, as FastStream would have done without this wrapper. """ import functools + import inspect import anyio from eggai.transport.pending_reclaimer import xack_del delete_on_error = ack_policy in (AckPolicy.ACK, AckPolicy.ACK_FIRST) - is_async = asyncio.iscoroutinefunction(handler) + # An object with an async __call__ is async too, not a sync function. + is_async = asyncio.iscoroutinefunction(handler) or asyncio.iscoroutinefunction( + type(handler).__call__ + ) async def delete_current() -> None: - message = self.broker.context.get_local("message") - if message is None: - return - raw = message.raw_message - channel, ids = raw["channel"], raw["message_ids"] + # Everything, lookups included, inside the try: raising here would + # NACK a handler that already succeeded and re-run it. On failure + # the entry is left for FastStream's XACK / MAXLEN. + channel: Any = None + ids: Any = None try: + message = self.broker.context.get_local("message") + if message is None: + return + raw = message.raw_message + channel, ids = raw["channel"], raw["message_ids"] await xack_del( self.broker.config.connection.client, channel, group, *ids ) except Exception: - # Raising here would NACK a handler that already succeeded and - # re-run it. Leave the entry for FastStream's XACK / MAXLEN. logger.warning( "delete_on_ack: XACK+XDEL failed for %s %s; entry left in stream", channel, @@ -935,6 +998,10 @@ async def delete_on_ack_handler(*args, **kwargs): result = await anyio.to_thread.run_sync( functools.partial(handler, *args, **kwargs) ) + # A callable that still returned a coroutine: await it here + # rather than drop it (it would never run). + if inspect.isawaitable(result): + result = await result except Exception: if delete_on_error: await delete_current() diff --git a/sdk/tests/test_redis_delete_on_ack.py b/sdk/tests/test_redis_delete_on_ack.py index 3ed36f30..6df67265 100644 --- a/sdk/tests/test_redis_delete_on_ack.py +++ b/sdk/tests/test_redis_delete_on_ack.py @@ -4,6 +4,7 @@ """ import asyncio +import logging import threading import uuid @@ -301,6 +302,90 @@ async def acked(): await agent.stop() +@pytest.mark.asyncio +async def test_delete_on_ack_refuses_to_start_next_to_another_group(redis_client): + """Another group on the stream (e.g. a monitoring bridge) would miss deleted + entries: connect() must fail before anything is consumed.""" + agent_name, channel_name, stream = _names("doa-foreign") + await redis_client.xgroup_create(stream, "someone-else", id="$", mkstream=True) + await redis_client.xadd(stream, {"k": "v"}) + + transport = RedisTransport() + agent = Agent(agent_name, transport=transport) + channel = Channel(channel_name, transport=transport) + calls = [] + + @agent.subscribe(channel=channel, delete_on_ack=True) + async def handler(message): + calls.append(message) + + with pytest.raises(RuntimeError, match="someone-else"): + await agent.start() + await transport.disconnect() + assert calls == [] + assert await redis_client.xlen(stream) == 1 + + +@pytest.mark.asyncio +async def test_delete_on_ack_refuses_two_local_groups_on_one_stream(): + agent_name, channel_name, _ = _names("doa-twolocal") + transport = RedisTransport() + agent = Agent(agent_name, transport=transport) + channel = Channel(channel_name, transport=transport) + + @agent.subscribe(channel=channel, delete_on_ack=True) + async def first(message): + pass + + @agent.subscribe(channel=channel) + async def second(message): + pass + + with pytest.raises(RuntimeError, match="other consumer groups"): + await agent.start() + await transport.disconnect() + + +@pytest.mark.asyncio +async def test_delete_on_ack_monitor_reports_group_added_later(redis_client, caplog): + agent_name, channel_name, stream = _names("doa-late") + transport = RedisTransport(group_monitor_interval_s=0.2) + agent = Agent(agent_name, transport=transport) + channel = Channel(channel_name, transport=transport) + + @agent.subscribe(channel=channel, delete_on_ack=True) + async def handler(message): + pass + + await agent.start() + try: + with caplog.at_level(logging.ERROR, logger="eggai.transport.redis"): + await redis_client.xgroup_create(stream, "late-joiner", id="$") + await _wait_for(lambda: _async("late-joiner" in caplog.text)) + await asyncio.sleep(0.6) # several more monitor cycles + assert caplog.text.count("late-joiner") == 1 # reported once, not per cycle + finally: + await agent.stop() + + +@pytest.mark.asyncio +async def test_delete_on_ack_awaits_callable_with_async_call(): + """An object with an async __call__ must be awaited, not sent to a thread + where its coroutine would be created and dropped.""" + calls = [] + + class Handler: + async def __call__(self, message): + calls.append(message) + return "done" + + transport = RedisTransport() + wrapped = transport._wrap_delete_on_ack(Handler(), "g", AckPolicy.NACK_ON_ERROR) + # Outside a FastStream consume there is no current message: nothing to delete. + assert await wrapped({"n": 1}) == "done" + assert calls == [{"n": 1}] + + @pytest.mark.asyncio @pytest.mark.parametrize( "extra, match", From 7cd2495a992b9fb51c9265b1dcf05d0f588fbae9 Mon Sep 17 00:00:00 2001 From: Tom Artiom Fiodorov Date: Fri, 25 Sep 2026 10:36:41 +0200 Subject: [PATCH 4/5] refactor(redis): delete_on_ack uses the transport's own Redis client XACK+XDEL no longer reaches into FastStream internals (broker.config.connection.client). The transport opens a client in connect(), built from the same URL and connection kwargs as the reclaimer and the group monitor, and closes it in disconnect() after the broker stops. It is only opened when a subscription uses delete_on_ack. A failed delete now logs at error level. Co-Authored-By: Claude Opus 5.5 (1M context) --- sdk/eggai/transport/redis.py | 26 +++++++++++++++++++++----- sdk/tests/test_redis_delete_on_ack.py | 3 +++ 2 files changed, 24 insertions(+), 5 deletions(-) diff --git a/sdk/eggai/transport/redis.py b/sdk/eggai/transport/redis.py index 2fc8f538..bbb1f371 100644 --- a/sdk/eggai/transport/redis.py +++ b/sdk/eggai/transport/redis.py @@ -189,6 +189,9 @@ def __init__( self._delete_on_ack_groups: set[tuple[str, str]] = set() # Foreign-group sets already reported by the monitor, so it logs once. self._reported_foreign_groups: set[tuple[str, frozenset[str]]] = set() + # delete_on_ack's own client, opened in connect() like the reclaimer's and + # the group monitor's, rather than reaching into FastStream's internals. + self._delete_client: aioredis.Redis | None = None async def connect(self): """ @@ -221,6 +224,12 @@ async def connect(self): if self._reclaimer_manager is not None: # start() is idempotent — it skips reclaimer tasks already running. await self._reclaimer_manager.start() + # Per connect(), not per subscribe: a second Agent on a shared transport + # may add the first delete_on_ack subscription after an earlier connect(). + if self._delete_on_ack_groups and self._delete_client is None: + self._delete_client = aioredis.from_url( + self._redis_url, **self._connection_kwargs + ) self._running = True # Don't overwrite (and orphan) a monitor task still running from an # earlier connect() on a shared transport — it already iterates the @@ -251,6 +260,10 @@ async def disconnect(self): if self._reclaimer_manager is not None: await self._reclaimer_manager.stop() await self.broker.stop() + # After broker.stop(): handlers still finishing may need it to delete. + if self._delete_client is not None: + await self._delete_client.aclose() + self._delete_client = None async def publish(self, channel: str, message: dict[str, Any] | BaseModel): """ @@ -949,7 +962,8 @@ def _wrap_delete_on_ack( trimmed. FastStream's own XACK still runs afterwards, a harmless no-op on the deleted id. The raw stream message (channel + ids, a list even for one entry, so batch mode works) comes from FastStream's per-message - context. functools.wraps keeps the user's signature visible to + context; the XACK+XDEL goes through the transport's own client + (self._delete_client, opened in connect()). functools.wraps keeps the user's signature visible to fast_depends, same as the tracing wrapper. A sync handler is sent to a worker thread, as FastStream would have done without this wrapper. """ @@ -978,11 +992,13 @@ async def delete_current() -> None: return raw = message.raw_message channel, ids = raw["channel"], raw["message_ids"] - await xack_del( - self.broker.config.connection.client, channel, group, *ids - ) + if self._delete_client is None: + raise RuntimeError( + "delete_on_ack client not open (connect() not run)" + ) + await xack_del(self._delete_client, channel, group, *ids) except Exception: - logger.warning( + logger.error( "delete_on_ack: XACK+XDEL failed for %s %s; entry left in stream", channel, ids, diff --git a/sdk/tests/test_redis_delete_on_ack.py b/sdk/tests/test_redis_delete_on_ack.py index 6df67265..edc31c9a 100644 --- a/sdk/tests/test_redis_delete_on_ack.py +++ b/sdk/tests/test_redis_delete_on_ack.py @@ -61,6 +61,7 @@ async def handler(message): await agent.start() try: + assert transport._delete_client is not None # opened by connect() for n in range(5): await channel.publish({"type": "t", "n": n}) await _wait_for(lambda: _async(len(seen) == 5)) @@ -69,6 +70,7 @@ async def handler(message): finally: await agent.stop() assert sorted(seen) == [0, 1, 2, 3, 4] + assert transport._delete_client is None # closed by disconnect() @pytest.mark.asyncio @@ -86,6 +88,7 @@ async def handler(message): seen.append(message["n"]) await agent.start() + assert transport._delete_client is None # no delete_on_ack, no extra client try: for n in range(3): await channel.publish({"type": "t", "n": n}) From a5a116da51c3914e6cd974c7a607ea1751590946 Mon Sep 17 00:00:00 2001 From: Tom Artiom Fiodorov Date: Fri, 25 Sep 2026 12:21:48 +0200 Subject: [PATCH 5/5] fix(redis): make delete_on_ack fail safe (review follow-ups) - Open the delete client before _create_groups() / broker.start(), so a backlog handled at startup is deleted too; close it if the foreign-group check refuses to start. - When the group monitor finds a group that joined a delete_on_ack stream after startup, turn deletion off for that stream (plain XACK in the handler wrapper and the reclaimer, via a shared set) instead of only logging. Stays off until restart. - The startup error names the XGROUP DESTROY fix and explains that renaming a handler renames its group. - Docs: keep max_len as a backstop, deleted entries can't be replayed, tested on Redis 7.4 OSS only, groups renamed with handlers. - Test that XINFO GROUPS still reports lag after deletes (KEDA lagCount). - Imports moved to module level; docstring note on @publisher return values. Co-Authored-By: Claude Opus 5.5 (1M context) --- docs/docs/sdk/redis-transport.md | 15 +++- sdk/CHANGELOG.md | 3 +- sdk/README.md | 6 +- sdk/eggai/transport/pending_reclaimer.py | 15 +++- sdk/eggai/transport/redis.py | 91 +++++++++++++++++------- sdk/tests/test_redis_delete_on_ack.py | 88 ++++++++++++++++++++++- 6 files changed, 183 insertions(+), 35 deletions(-) diff --git a/docs/docs/sdk/redis-transport.md b/docs/docs/sdk/redis-transport.md index 7e27cf0e..c6e385e4 100644 --- a/docs/docs/sdk/redis-transport.md +++ b/docs/docs/sdk/redis-transport.md @@ -293,7 +293,20 @@ async def handle_email(message): It applies wherever the subscription acks: after the handler returns (including messages a filter skips), on the SDK retry stream, and when the reclaimer moves a stuck entry to `.retry` or the DLQ. Under the default `NACK_ON_ERROR` a failing handler's entry stays in the PEL for retry; under `AckPolicy.ACK` / `ACK_FIRST`, which ack failures too, it is deleted too. DLQ entries are never deleted by this option, and a poison entry dropped with no DLQ configured is only acked so it stays readable with `XRANGE`. -**Only one consumer group per stream.** `XDEL` removes the entry for every group, so a second group (or a group-less subscriber) that has not read an entry yet never will. Competing workers inside one group are fine. `connect()` checks this: if another group already exists on the stream it raises `RuntimeError` before anything is consumed, and the group monitor logs an error (once) if one appears later. For fan-out to several groups, keep `delete_on_ack` off and bound memory with `max_len`. Requires a consumer group; rejected with `no_ack=True` and `AckPolicy.MANUAL`. On Redis >= 8.2 the equivalent command is `XACKDEL`. +**Only one consumer group per stream.** `XDEL` removes the entry for every group, so a second group (or a group-less subscriber) that has not read an entry yet never will. Competing workers inside one group are fine. For fan-out to several groups, keep `delete_on_ack` off and bound memory with `max_len`. Requires a consumer group; rejected with `no_ack=True` and `AckPolicy.MANUAL`. On Redis >= 8.2 the equivalent command is `XACKDEL`. + +How the one-group rule is enforced: + +- **At `connect()`**, before any consumer starts: another group on the stream raises `RuntimeError` naming it. +- **At runtime**, the group monitor (every `group_monitor_interval_s`) checks again. If a group has appeared, it turns deletion **off** for that stream (plain `XACK`, entries kept) and logs an error. Memory grows again, but the other group only misses what was deleted during one monitor interval. Deletion stays off until restart, and the next `connect()` refuses to start until the extra group is gone. +- **Renaming a handler renames its group.** `Agent` names groups `--`, so after renaming a handler function the old group is still on the stream and every pod refuses to start. Remove it with `XGROUP DESTROY `, and on the old `.retry` stream `XGROUP DESTROY ..retry -retry`. + +Two consequences to plan for: + +- **Keep `max_len` as a backstop.** A few paths still leave acked entries behind: a delete that fails (it is logged, not retried, so a successful handler isn't run again), deletion turned off by the monitor, and a crash mid-handler under `ACK_FIRST`. A generous `max_len` next to `delete_on_ack` bounds them. +- **No replay.** Deleted entries are gone. Rewinding a group (`XGROUP SETID 0`) or adding a group to backfill history finds nothing. Runbooks that rely on rewinds need another source. + +Tested against Redis 7.4 (open source). Redis Enterprise Active-Active / geo-replicated Azure Managed Redis has not been tested; replicated consumer-group state may behave differently there. ### Tuning the Reclaimer diff --git a/sdk/CHANGELOG.md b/sdk/CHANGELOG.md index 3eb81bd8..3e1ededd 100644 --- a/sdk/CHANGELOG.md +++ b/sdk/CHANGELOG.md @@ -18,7 +18,8 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0 `AckPolicy.ACK` / `ACK_FIRST` (which ack it anyway). Opt-in because it assumes a single consumer group per stream: `XDEL` removes the entry for every group. `connect()` raises `RuntimeError` if another group already - reads the stream, and the group monitor logs an error if one joins later. + reads the stream; if one joins later, the group monitor turns deletion off + for that stream (plain `XACK`) and logs an error. ### Fixed diff --git a/sdk/README.md b/sdk/README.md index 8234d4cb..e1edd70d 100644 --- a/sdk/README.md +++ b/sdk/README.md @@ -248,8 +248,10 @@ in one `MULTI`): after the handler returns, on the retry stream, and when the re moves an entry to `.retry` or the DLQ. A failing handler's entry still stays in the PEL for retry (under the default `NACK_ON_ERROR`). **It assumes one consumer group per stream:** `XDEL` removes the entry for every group, so don't use it on a stream that -fans out to several groups. `connect()` raises if another group already reads the stream, -and the group monitor logs an error if one joins later. +fans out to several groups. `connect()` raises if another group already reads the stream; +if one joins later, the group monitor turns deletion off for that stream and logs an error. +Deleted entries can't be replayed, and a generous `max_len` is still worth keeping as a +backstop for the few paths that leave acked entries behind. **Retry backoff:** By default retries fire at a constant cadence equal to `retry_on_idle_ms`. Set `retry_backoff_multiplier > 1.0` to back off exponentially: a diff --git a/sdk/eggai/transport/pending_reclaimer.py b/sdk/eggai/transport/pending_reclaimer.py index fb4973d3..db696752 100644 --- a/sdk/eggai/transport/pending_reclaimer.py +++ b/sdk/eggai/transport/pending_reclaimer.py @@ -287,8 +287,19 @@ class PendingReclaimerManager: message body) can be used for application-level deduplication. """ - def __init__(self, redis_url: str, connection_kwargs: dict[str, Any] | None = None): + def __init__( + self, + redis_url: str, + connection_kwargs: dict[str, Any] | None = None, + delete_on_ack_disabled: set[str] | None = None, + ): self._redis_url = redis_url + # Stream keys where delete_on_ack was turned off at runtime (another + # consumer group appeared). Shared with, and mutated by, the transport's + # group monitor; _ack() falls back to a plain XACK for these. + self._delete_on_ack_disabled: set[str] = ( + delete_on_ack_disabled if delete_on_ack_disabled is not None else set() + ) # Connection-resilience settings (socket_timeout, socket_keepalive, # health_check_interval, retry_on_timeout, …) forwarded from the # transport so this independent client recovers from a silently dropped @@ -615,7 +626,7 @@ async def _reclaim_once(self, config: ReclaimerConfig) -> None: logger.debug("Reclaimed %s → %s", msg_id_str, config.retry_stream) async def _ack(self, config: ReclaimerConfig, msg_id: Any) -> None: - if config.delete_on_ack: + if config.delete_on_ack and config.stream not in self._delete_on_ack_disabled: await xack_del(self._client, config.stream, config.group, msg_id) else: await self._client.xack(config.stream, config.group, msg_id) diff --git a/sdk/eggai/transport/redis.py b/sdk/eggai/transport/redis.py index bbb1f371..53d64216 100644 --- a/sdk/eggai/transport/redis.py +++ b/sdk/eggai/transport/redis.py @@ -1,4 +1,6 @@ import asyncio +import functools +import inspect import logging import os import socket @@ -6,6 +8,7 @@ from dataclasses import dataclass from typing import Any +import anyio import redis.asyncio as aioredis from faststream import AckPolicy from faststream.redis import RedisBroker, StreamSub @@ -14,7 +17,11 @@ from eggai.transport.base import Transport from eggai.transport.middleware_utils import wrap_handler_with_filters -from eggai.transport.pending_reclaimer import PendingReclaimerManager, ReclaimerConfig +from eggai.transport.pending_reclaimer import ( + PendingReclaimerManager, + ReclaimerConfig, + xack_del, +) logger = logging.getLogger(__name__) @@ -192,6 +199,9 @@ def __init__( # delete_on_ack's own client, opened in connect() like the reclaimer's and # the group monitor's, rather than reaching into FastStream's internals. self._delete_client: aioredis.Redis | None = None + # Stream keys where the monitor turned delete_on_ack off because another + # group appeared; shared with the reclaimer. Plain XACK from then on. + self._delete_on_ack_disabled: set[str] = set() async def connect(self): """ @@ -213,7 +223,24 @@ async def connect(self): # subscribers that aren't running yet (e.g. those a second Agent just # registered). Falls back to the original full start() if FastStream ever # stops exposing the `running` flags. - await self._create_groups() + # delete_on_ack's client opens before any consumer starts, so no handler + # can run without it. Per connect(), not per subscribe: a second Agent on + # a shared transport may add the first delete_on_ack subscription after + # an earlier connect(). + opened_delete_client = False + if self._delete_on_ack_groups and self._delete_client is None: + self._delete_client = aioredis.from_url( + self._redis_url, **self._connection_kwargs + ) + opened_delete_client = True + try: + await self._create_groups() + except Exception: + # e.g. the foreign-group check refused to start: don't leak it. + if opened_delete_client and self._delete_client is not None: + await self._delete_client.aclose() + self._delete_client = None + raise if not getattr(self.broker, "running", False): await self.broker.start() else: @@ -224,12 +251,6 @@ async def connect(self): if self._reclaimer_manager is not None: # start() is idempotent — it skips reclaimer tasks already running. await self._reclaimer_manager.start() - # Per connect(), not per subscribe: a second Agent on a shared transport - # may add the first delete_on_ack subscription after an earlier connect(). - if self._delete_on_ack_groups and self._delete_client is None: - self._delete_client = aioredis.from_url( - self._redis_url, **self._connection_kwargs - ) self._running = True # Don't overwrite (and orphan) a monitor task still running from an # earlier connect() on a shared transport — it already iterates the @@ -336,9 +357,12 @@ async def subscribe(self, channel: str, handler, **kwargs) -> Callable: ASSUMES THIS IS THE ONLY CONSUMER GROUP ON THE STREAM: ``XDEL`` removes the entry for every group, so another group (or a group-less ``XREAD`` subscriber) that has not read it yet never will. Checked at ``connect()``: another group already on the stream raises ``RuntimeError`` - before anything is consumed, and the group monitor logs an error for one that appears - later. Requires a consumer group; incompatible with ``no_ack=True`` and - ``ack_policy=AckPolicy.MANUAL``. On Redis >= 8.2 the same effect is ``XACKDEL``. + before anything is consumed; if one appears later, the group monitor turns deletion off + for that stream (plain ``XACK`` until restart) and logs an error. Deleted entries can't be + replayed (``XGROUP SETID ... 0``, a new group backfilling). Keep a generous ``max_len`` as a + backstop for entries a failed delete leaves behind. Requires a consumer group; incompatible + with ``no_ack=True`` and ``ack_policy=AckPolicy.MANUAL``. On Redis >= 8.2 the same effect is + ``XACKDEL``. ack_policy (AckPolicy, optional): Acknowledgment policy for message handling (default is AckPolicy.NACK_ON_ERROR). - NACK_ON_ERROR: Messages are NOT acknowledged on handler errors, allowing redelivery (recommended). - ACK: Messages are acknowledged regardless of handler success/failure. @@ -855,8 +879,12 @@ async def _create_groups(self) -> None: f"delete_on_ack on stream {stream_key!r} (group {group!r}) " f"but other consumer groups also read it: {foreign}. XDEL " "would delete entries they have not read yet. Drop " - "delete_on_ack or remove the other groups " - "(XGROUP DESTROY)." + "delete_on_ack, or remove groups nobody uses any more with " + f"XGROUP DESTROY {stream_key} . A group left over " + "from this service is the usual cause: Agent names groups " + "'--', so renaming a handler " + "renames its group (destroy the old one's '-retry' " + "group on its '.retry' stream too)." ) finally: await client.aclose() @@ -927,9 +955,10 @@ async def _monitor_stream_groups(self) -> None: await client.aclose() async def _check_delete_on_ack_groups(self, client: Any) -> None: - """Log (once per set) a consumer group that joined a delete_on_ack - stream after startup. The startup check in _create_groups raises; here - the consumers are already running, so the monitor can only report.""" + """Handle a consumer group that joined a delete_on_ack stream after + startup: turn deletion off for that stream and log an error (once per + set). The startup check in _create_groups raises instead; here the + consumers are already running.""" for stream_key, group in list(self._delete_on_ack_groups): try: foreign = await self._foreign_groups(client, stream_key, group) @@ -938,13 +967,20 @@ async def _check_delete_on_ack_groups(self, client: Any) -> None: "delete_on_ack group check failed on %s: %s", stream_key, e ) continue + if not foreign: + continue + # Fail safe: stop deleting on this stream (plain XACK from now on, + # handler and reclaimer both), trading memory for the other group's + # messages. Stays off until restart, where connect() refuses to start. + self._delete_on_ack_disabled.add(stream_key) key = (stream_key, frozenset(foreign)) - if foreign and key not in self._reported_foreign_groups: + if key not in self._reported_foreign_groups: self._reported_foreign_groups.add(key) logger.error( "delete_on_ack on stream %s (group %s) but other consumer " - "groups now read it too: %s. They will miss entries this " - "group deletes first.", + "groups now read it too: %s. Deletion is OFF for this stream " + "until restart (plain XACK, entries kept); the next connect() " + "will refuse to start until the extra groups are removed.", stream_key, group, foreign, @@ -966,14 +1002,11 @@ def _wrap_delete_on_ack( (self._delete_client, opened in connect()). functools.wraps keeps the user's signature visible to fast_depends, same as the tracing wrapper. A sync handler is sent to a worker thread, as FastStream would have done without this wrapper. - """ - import functools - import inspect - - import anyio - - from eggai.transport.pending_reclaimer import xack_del + The delete happens when the handler returns, before FastStream + publishes a return value (``@publisher``); a failed publish can't be + retried from a deleted entry. EggAI handlers don't publish return values. + """ delete_on_error = ack_policy in (AckPolicy.ACK, AckPolicy.ACK_FIRST) # An object with an async __call__ is async too, not a sync function. is_async = asyncio.iscoroutinefunction(handler) or asyncio.iscoroutinefunction( @@ -992,6 +1025,8 @@ async def delete_current() -> None: return raw = message.raw_message channel, ids = raw["channel"], raw["message_ids"] + if channel in self._delete_on_ack_disabled: + return # FastStream's plain XACK still runs if self._delete_client is None: raise RuntimeError( "delete_on_ack client not open (connect() not run)" @@ -1049,7 +1084,9 @@ def _setup_reclaimer( ) -> tuple[str, str, str]: if self._reclaimer_manager is None: self._reclaimer_manager = PendingReclaimerManager( - self._redis_url, connection_kwargs=self._connection_kwargs + self._redis_url, + connection_kwargs=self._connection_kwargs, + delete_on_ack_disabled=self._delete_on_ack_disabled, ) return self._reclaimer_manager.add( ReclaimerConfig( diff --git a/sdk/tests/test_redis_delete_on_ack.py b/sdk/tests/test_redis_delete_on_ack.py index edc31c9a..9f411806 100644 --- a/sdk/tests/test_redis_delete_on_ack.py +++ b/sdk/tests/test_redis_delete_on_ack.py @@ -67,6 +67,12 @@ async def handler(message): await _wait_for(lambda: _async(len(seen) == 5)) await _wait_for(lambda: _xlen_is(redis_client, stream, 0)) assert _pending(await redis_client.xpending(stream, group)) == 0 + # Only delivered entries are deleted, so Redis can still compute the + # group's lag (null means it can't); KEDA's lagCount scaler reads it. + (info,) = [ + g for g in await redis_client.xinfo_groups(stream) if g["name"] == group + ] + assert info["lag"] == 0 finally: await agent.stop() assert sorted(seen) == [0, 1, 2, 3, 4] @@ -324,6 +330,7 @@ async def handler(message): with pytest.raises(RuntimeError, match="someone-else"): await agent.start() + assert transport._delete_client is None # not leaked by the failed connect() await transport.disconnect() assert calls == [] assert await redis_client.xlen(stream) == 1 @@ -350,27 +357,104 @@ async def second(message): @pytest.mark.asyncio -async def test_delete_on_ack_monitor_reports_group_added_later(redis_client, caplog): +async def test_delete_on_ack_monitor_turns_deletion_off_for_late_group( + redis_client, caplog +): + """A group that joins after startup: the monitor stops deleting on that + stream (plain XACK) so the newcomer gets every later entry, and logs once.""" agent_name, channel_name, stream = _names("doa-late") + group = f"{agent_name}-handler-1" transport = RedisTransport(group_monitor_interval_s=0.2) agent = Agent(agent_name, transport=transport) channel = Channel(channel_name, transport=transport) + seen = [] @agent.subscribe(channel=channel, delete_on_ack=True) async def handler(message): - pass + seen.append(message["n"]) await agent.start() try: + await channel.publish({"type": "t", "n": 1}) + await _wait_for(lambda: _xlen_is(redis_client, stream, 0)) # deleting + with caplog.at_level(logging.ERROR, logger="eggai.transport.redis"): await redis_client.xgroup_create(stream, "late-joiner", id="$") await _wait_for(lambda: _async("late-joiner" in caplog.text)) await asyncio.sleep(0.6) # several more monitor cycles assert caplog.text.count("late-joiner") == 1 # reported once, not per cycle + assert stream in transport._delete_on_ack_disabled + + await channel.publish({"type": "t", "n": 2}) + await _wait_for(lambda: _async(seen == [1, 2])) + await asyncio.sleep(0.3) + assert await redis_client.xlen(stream) == 1 # acked, kept for late-joiner + assert _pending(await redis_client.xpending(stream, group)) == 0 + finally: + await agent.stop() + + +@pytest.mark.asyncio +async def test_delete_on_ack_client_is_open_before_consumers_start(): + """A backlog handled right at startup must find the delete client open.""" + agent_name, channel_name, _ = _names("doa-order") + transport = RedisTransport() + agent = Agent(agent_name, transport=transport) + channel = Channel(channel_name, transport=transport) + + @agent.subscribe(channel=channel, delete_on_ack=True) + async def handler(message): + pass + + client_at_start = [] + original_start = transport.broker.start + + async def recording_start(*args, **kwargs): + client_at_start.append(transport._delete_client) + return await original_start(*args, **kwargs) + + transport.broker.start = recording_start + await agent.start() + try: + assert client_at_start and client_at_start[0] is not None finally: await agent.stop() +@pytest.mark.asyncio +async def test_reclaimer_ack_respects_runtime_disabled_streams(): + """Once the monitor disables deletion on a stream, the reclaimer's moves to + .retry / the DLQ fall back to a plain XACK too.""" + from unittest.mock import AsyncMock, MagicMock + + from eggai.transport.pending_reclaimer import ( + PendingReclaimerManager, + ReclaimerConfig, + ) + + disabled: set[str] = set() + manager = PendingReclaimerManager( + "redis://localhost:6379", delete_on_ack_disabled=disabled + ) + client = MagicMock() + client.xack = AsyncMock() + manager._redis_client = client + config = ReclaimerConfig( + stream="s", + group="g", + consumer="c", + retry_stream="s.retry", + min_idle_ms=1, + interval_s=1.0, + delete_on_ack=True, + ) + + disabled.add("s") + await manager._ack(config, b"1-0") + client.xack.assert_awaited_once_with("s", "g", b"1-0") + client.pipeline.assert_not_called() + + @pytest.mark.asyncio async def test_delete_on_ack_awaits_callable_with_async_call(): """An object with an async __call__ must be awaited, not sent to a thread