diff --git a/docs/docs/sdk/redis-transport.md b/docs/docs/sdk/redis-transport.md index d0e6b860..c6e385e4 100644 --- a/docs/docs/sdk/redis-transport.md +++ b/docs/docs/sdk/redis-transport.md @@ -281,6 +281,33 @@ 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`. + +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 ```python diff --git a/sdk/CHANGELOG.md b/sdk/CHANGELOG.md index aa6d1206..3e1ededd 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] +### 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 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. `connect()` raises `RuntimeError` if another group already + reads the stream; if one joins later, the group monitor turns deletion off + for that stream (plain `XACK`) and logs an error. + ### Fixed - MCP adapter reads tool schemas from fastmcp's own `Tool` instead of the diff --git a/sdk/README.md b/sdk/README.md index 534a7292..e1edd70d 100644 --- a/sdk/README.md +++ b/sdk/README.md @@ -241,6 +241,18 @@ 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. `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 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 2311c0ac..db696752 100644 --- a/sdk/eggai/transport/pending_reclaimer.py +++ b/sdk/eggai/transport/pending_reclaimer.py @@ -81,6 +81,22 @@ 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 + + +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: @@ -263,7 +279,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. @@ -271,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 @@ -549,7 +576,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,6 +585,8 @@ async def _reclaim_once(self, config: ReclaimerConfig) -> None: ) await self._invoke_on_dlq(config, body, msg_id_str, 0) else: + # 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 " @@ -582,7 +611,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 +622,15 @@ 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 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) + async def _invoke_on_dlq( self, config: ReclaimerConfig, diff --git a/sdk/eggai/transport/redis.py b/sdk/eggai/transport/redis.py index 6ee464bd..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__) @@ -184,6 +191,17 @@ 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() + # 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): """ @@ -205,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: @@ -246,6 +281,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): """ @@ -306,6 +345,24 @@ 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. 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. Checked at ``connect()``: another group already on the stream raises ``RuntimeError`` + 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. @@ -470,6 +527,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 +690,13 @@ 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). 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, ack_policy) + self._delete_on_ack_groups.add((channel, group)) + # stream must be passed as keyword-only argument registered_handler = self.broker.subscriber( stream=stream_sub, ack_policy=ack_policy, **kwargs @@ -678,6 +761,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 +804,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 +830,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: @@ -752,6 +838,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) @@ -782,9 +870,34 @@ 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 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() + @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. @@ -835,11 +948,120 @@ 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: + """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) + except Exception as e: # connection blips: next cycle retries + logger.debug( + "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 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. 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, + ) + + 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; 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. + + 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( + type(handler).__call__ + ) + + async def delete_current() -> None: + # 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"] + 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)" + ) + await xack_del(self._delete_client, channel, group, *ids) + except Exception: + logger.error( + "delete_on_ack: XACK+XDEL failed for %s %s; entry left in stream", + channel, + 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) + ) + # 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() + raise + await delete_current() + return result + + return delete_on_ack_handler + def _setup_reclaimer( self, *, @@ -858,10 +1080,13 @@ 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( - 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( @@ -881,5 +1106,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..9f411806 --- /dev/null +++ b/sdk/tests/test_redis_delete_on_ack.py @@ -0,0 +1,512 @@ +""" +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 logging +import threading +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: + 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)) + 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] + assert transport._delete_client is None # closed by disconnect() + + +@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() + 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}) + 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 +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 +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() + assert transport._delete_client is None # not leaked by the failed connect() + 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_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): + 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 + 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", + [ + ({"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