Repository navigation
feat(redis): opt-in delete_on_ack to XDEL entries when they are acked - #297
Conversation
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) <[email protected]>
QualOps Code Quality AnalysisStatus: Summary
🟠 High Issues (1)
🟡 Medium Issues (5)
...and 2 more medium issues 📊 Full ReportPowered by QualOps |
- 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) <[email protected]>
- 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) <[email protected]>
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) <[email protected]>
tsiaeggai
left a comment
tsiaeggai
left a comment
There was a problem hiding this comment.
Good addition, and a careful implementation. It's opt-in, it deletes only where the ack policy already acks, XACK+XDEL run in one MULTI so no pending entries are left pointing at deleted messages, the DLQ and the last copy of a bad message are kept, and it refuses unsafe setups at subscribe time and at connect(). The integration tests against real Redis cover the tricky paths well.
Being off by default matters here. Fan-out, where several agents each subscribe to the same channel and so each get their own consumer group, is a common EggAI pattern. Deployments like that can't use this option, but they can upgrade without any change in behaviour. Approving as is. The comments below are optional suggestions for making it fail safe for teams that do enable it. None of them block merging, and they could be a follow-up PR just as well.
Questions (no change requested)
- Return-value publishing: if a handler's return value is published by FastStream (
@publisher), that publish happens after thedelete_on_ackwrapper has already deleted the entry, so a failed publish can't be retried. Probably doesn't matter for how EggAI is used, but worth a line in the docstring if it's supported. - Redis Enterprise Active-Active / geo-replicated Azure Managed Redis: have you checked
MULTI XACK XDELand the consumer-group check there? Consumer-group state behaves differently on replicated streams (for example,XINFO GROUPSmay leave out thelag/entries-readfields), so it may be worth a note in the docs about which setups have been tested.
| 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: |
There was a problem hiding this comment.
The delete client is opened after broker.start() has started the consumers. Any message handled in between gets delete_on_ack client not open (connect() not run) and is logged but not deleted. Nothing is lost, but that message is never deleted, and on startup with a backlog that could be quite a few. Small suggestion: could the client be opened before _create_groups() / broker.start()? That also means any handler that runs always finds it open.
(Also, the QualOps finding about the client leaking when the foreign-group check raises doesn't apply: _create_groups() runs before the client is opened.)
There was a problem hiding this comment.
Good catch, fixed in a5a116d. The delete client now opens before _create_groups() / broker.start(), so no handler can run without it, and a backlog at startup is deleted too. If the foreign-group check refuses to start, the client opened by that connect() is closed again. Tests: test_delete_on_ack_client_is_open_before_consumers_start checks the client exists when broker.start() is called, and the foreign-group test asserts the client isn't leaked after the raise.
| finally: | ||
| await client.aclose() | ||
|
|
||
| async def _check_delete_on_ack_groups(self, client: Any) -> None: |
There was a problem hiding this comment.
Optional, but probably the most valuable of these suggestions. Every Agent.subscribe gets its own group, so a second service on another pod that subscribes after this one started creates a group the startup check never sees. The monitor logs it once, but the transport keeps XDELing, so that service quietly misses entries from then on.
Could the monitor turn deletion off for that stream when it finds a foreign group (e.g. add the stream key to a _delete_on_ack_disabled set that delete_current and the reclaimer's _ack check, falling back to plain XACK)? Memory grows again, but no messages are lost, and the error log still tells someone to look at it. It would also shrink the window to one monitor interval rather than "until someone reads the logs".
There was a problem hiding this comment.
Agreed, done in a5a116d. When the monitor finds a foreign group on a delete_on_ack stream, it adds the stream key to _delete_on_ack_disabled. That set is shared with the reclaimer (PendingReclaimerManager(delete_on_ack_disabled=...)), so both the handler wrapper and the reclaimer's _ack fall back to a plain XACK. It stays off until restart, and the next connect() then refuses to start until the extra group is gone. The error log says deletion is OFF. The window is now one monitor interval (5s default). Test: test_delete_on_ack_monitor_turns_deletion_off_for_late_group checks that deletion works, a group joins, it's logged once, and later entries are acked but kept.
| for stream_key, group in sorted(self._delete_on_ack_groups): | ||
| foreign = await self._foreign_groups(client, stream_key, group) | ||
| if foreign: | ||
| raise RuntimeError( |
There was a problem hiding this comment.
Refusing to start is the right call, but it interacts with default group naming. Groups are {agent}-handler-{n}, numbered by subscription order, so adding or reordering a subscription renames the group. The old group stays on the stream, and on the next deploy every pod fails with this error until someone runs XGROUP DESTROY (for the .retry stream's group too). Suggest mentioning this in the error message and the docs, and recommending an explicit handler_id= on delete_on_ack subscriptions so the group name doesn't change.
There was a problem hiding this comment.
Agreed on the risk, with one correction to the details. Groups are <agent>-<handler function>-<n>, where n counts handlers with the same function name (agent.py:207). So reordering subscriptions doesn't rename anything unless two handlers share a name. Renaming a handler function does. Also Agent.start() unconditionally overwrites handler_id, so recommending an explicit handler_id= wouldn't help today. A way to pin the group name through Agent would be a separate change (happy to open an issue).
What I did in a5a116d: the error now gives the fix (XGROUP DESTROY <stream> <group>), explains that renaming a handler renames its group, and mentions the old <group>-retry group on the old .retry stream. The docs have the same with exact commands.
|
|
||
| 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`. |
There was a problem hiding this comment.
Two things worth stating here:
- Keep
max_lenas a backstop. Some paths still leave acked messages that are never deleted: anXDELfailure that is only logged, the startup race inconnect(), and withACK_FIRSTa crash while the handler is running. A generousmax_lenalongsidedelete_on_ackbounds that. As written, the docs read as ifdelete_on_ackreplacesmax_len. - No replay. Once entries are deleted, rewinding a group (
XGROUP SETID <stream> <group> 0) or adding a new group to backfill history has nothing to replay. Expected, but ops runbooks that rely on rewinds should know.
There was a problem hiding this comment.
Both added in a5a116d (docs, subscribe docstring and README):
- Keep
max_lenas a backstop. It lists the paths that leave acked entries: a delete that fails (logged, not retried, so a successful handler isn't re-run), deletion turned off by the monitor, and a crash mid-handler underACK_FIRST. - No replay.
XGROUP SETID … 0and backfilling a new group find nothing. Runbooks need another source.
Also added a line on which setups were tested (Redis 7.4 OSS, not Active-Active; see the reply on the review).
| 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 |
There was a problem hiding this comment.
Could we add an assertion that XINFO GROUPS still reports a non-null lag after deletes? Redis sets lag to null when it can't compute it after deletions. Since only already-delivered messages are deleted it should stay valid, but KEDA's Redis Streams scaler (lagCount) depends on it, and a test would catch a regression.
There was a problem hiding this comment.
Added in a5a116d: after 5 messages are handled and deleted, the test asserts XINFO GROUPS reports lag == 0 for the group (not null). On Redis 7.4 it holds, as expected, since only entries already handed to the group are deleted.
| 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 |
There was a problem hiding this comment.
Nit: The functools / inspect / anyio / xack_del imports can move to module level, unless there's a circular import I'm missing with pending_reclaimer.
There was a problem hiding this comment.
Moved to module level in a5a116d. No cycle: redis.py already imported PendingReclaimerManager from pending_reclaimer at the top.
- 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) <[email protected]>
|
Thanks for the thorough review @tsiaeggai. All six inline suggestions are addressed in a5a116d (replies in each thread). On the two questions:
Full suite: 123 passed (Kafka skipped locally), ruff/mypy clean. |
3 participants
Problem
Redis stream entries stay in memory after they are acked. The only cleanup today is
MAXLEN ~onXADD(RedisTransport(max_len=...)/retry_max_len). That trims by count, whether or not an entry has been processed, so:In a pipeline where each stage republishes a payload (e.g. whole emails with attachments), that is N copies of every message kept indefinitely. Redis persistence doesn't help: it is all-or-nothing.
Keeping acked entries is only useful when several consumer groups read the same stream (fan-out): each group needs the entry until it has read it. Many deployments have exactly one group per stream (competing consumers inside the group), so for them the retained entries are pure waste.
Change
New Redis subscribe option, off by default:
With
delete_on_ack=True, every point where this subscription acks an entry becomesMULTI; XACK; XDEL; EXECon the same stream key:delete_on_ack=TrueXACKXACK+XDELfilter_by_message/data_type/filter_by_dataXACKXACK+XDEL(the wrapper is outermost)XACKXACK+XDEL(retry group).retryXADD retry+XACKXADD retry+XACK+XDELXADD dlq+XACKXADD dlq+XACK+XDELXACKXACKonly. It's the sole copy, so it stays readable withXRANGENACK_ON_ERRORAckPolicy.ACK/ACK_FIRSTXACK(acked anyway)XACK+XDEL, otherwise it would be acked but never trimmedDLQ entries are never deleted by this option. The DLQ is the terminal sink, and its retention stays governed by
dlq_max_len/ its consumer.End state for a healthy pipeline: each stream only holds in-flight entries (PEL plus not yet delivered).
How
RedisTransport._wrap_delete_on_ack): wraps the handler outermost, around tracing and filters. It deletes exactly whereack_policyacks: on success, plus on failure forACK/ACK_FIRST. It reads the raw stream message (channel,message_ids) from FastStream's per-message context (broker.context.get_local("message")) and runsXACK+XDELin a transaction on the transport's own client (_delete_client). That client is opened inconnect(), before any consumer starts, only when some subscription usesdelete_on_ack, from the same URL and connection kwargs as the reclaimer and group monitor, and closed indisconnect()after the broker stops. So no FastStream internals are involved (earlier revisions borrowedbroker.config.connection.client).anyio.to_thread.run_sync), as FastStream does for a bare sync handler, so a blocking handler doesn't stall the event loop. An object with anasync __call__counts as async, and a coroutine that still comes back from the thread is awaited rather than dropped.pending_reclaimer.xack_del(client, stream, group, *ids)is shared by the wrapper and the reclaimer. The Redis 8.2XACKDELswap is a single change there. FastStream's ownXACKstill runs afterwards and is a harmless no-op on the deleted id, so ack policies and FastStream internals are untouched.message_idsis a list, so batch subscriptions work too.functools.wrapskeeps the user's signature visible to fast_depends, the same technique the tracing wrapper uses.XACK+XDELfails (e.g. a connection blip), or the message lookup itself fails, we log an error and don't raise. All of it sits inside onetry. The handler already succeeded, and raising would NACK it and run it again. The entry then falls back to FastStream's normalXACK(it just isn't deleted).ReclaimerConfig.delete_on_ack, plus aPendingReclaimerManager._ack()helper replacing 3 of the 4 directxackcalls. The fourth, the poison drop with no DLQ, stays a plainXACK(see table).subscribe()passesdelete_on_ackthrough, so the retry handler and the retry reclaimer delete too.Validation (ValueError at subscribe time)
delete_on_ackwithout a consumer group. Group-lessXREADhas no ack, so there's nothing to hook.delete_on_ack+no_ack=True.delete_on_ack+ack_policy=AckPolicy.MANUAL. The SDK can't know when a manual ack happens.XDELremoves the entry for every group. If a second group, or a group-less subscriber, subscribes to the same stream and hasn't read an entry yet, it never will. That's why this is opt-in. Handlers competing inside one group (the normal worker-fleet setup) are fine.This is easy to get wrong in EggAI, because every
Agent.subscribeon a channel gets its own group. A second subscriber (e.g. a monitoring bridge reading alongside the services) silently becomes a second group. So it is checked:connect():_create_groups()runsXINFO GROUPSon everydelete_on_ackstream after creating the groups and beforebroker.start(). Any other group raisesRuntimeErrornaming it, so nothing is consumed or deleted. This also catches two local handlers on one channel where one usesdelete_on_ack.group_monitor_interval_s) runs the same check. If a group has appeared, it turns deletion off for that stream (a shared_delete_on_ack_disabledset checked by both the handler wrapper and the reclaimer, which fall back to plainXACK) and logs an error once. Memory grows again, but the newcomer misses at most one monitor interval of entries. Deletion stays off until restart, and the nextconnect()refuses to start.XGROUP DESTROY) and the usual cause:Agentnames groups<agent>-<handler function>-<n>, so renaming a handler renames its group.Group-less
XREADreaders leave no trace in Redis, so they can't be detected.Alternatives considered
XACKDEL … ACKED/XADD … ACKEDtrimming. These are built-in and safe with multiple groups (delete or trim only once every group has acked). They are not available on managed Azure Redis today: Azure Managed Redis offers 7.4 only on all tiers (docs). When 8.2 lands,xack_delcan switch toXACKDELwithout changing the public option.XTRIM MINIDtrimmer computed from every group's oldest pending / last-delivered id. Safe with multiple groups and works on ≥ 6.2, but it's more machinery (per-streamXINFO GROUPS/XPENDINGpolling) and frees memory with a lag. Not needed while one group per stream is the norm, and still possible later as a separate opt-in.MAXLENalone. Trims unprocessed entries (see Problem).Notes / caveats
Keep a generous
max_lenas a backstop, and note that deleted entries can't be replayed (both documented).Tested on Redis 7.4 OSS only, not Redis Enterprise Active-Active or geo-replicated Azure.
Memory is reclaimed per radix-tree node:
XDELfrees a node once all its entries are deleted. With in-order consumption, nodes empty out steadily.XLENdrops immediately.Each successful message costs our
MULTIplus FastStream's now-redundantXACK. Avoiding the second round-trip means marking the message committed inside FastStream internals; left as a follow-up.Deletion hooks "handler returned" rather than FastStream's ack primitive, which is why
MANUALis rejected. Hooking the ack itself would cover every path uniformly but digs into FastStream internals; possible follow-up.Redis-only. Kafka / in-memory transports don't know the kwarg, same as
retry_on_idle_ms.Tests
New
sdk/tests/test_redis_delete_on_ack.py(real Redis 7.4):XLEN == 3).retrystreams both end emptymax_retries=1: main and.retryempty, DLQ holds the entryAckPolicy.ACK+ failing handler: entry deleted, PEL emptyagent.start()raises, nothing consumed, entry intactdelete_on_ack: raisesasync __call__: awaited, not droppedbroker.start(), and not leaked when startup is refusedXACKon a runtime-disabled streamXINFO GROUPSstill reportslag(0) after deletes, which KEDA'slagCountrelies onno_ack=True,AckPolicy.MANUAL, no groupLocal runs:
pytest tests/→ 123 passed, 14 skipped (Kafka, no broker locally)ruff check,ruff format --check,mypy→ cleanCHANGELOG updated under
[Unreleased] / Added. Documented indocs/docs/sdk/redis-transport.md(new "Freeing Memory withdelete_on_ack" section) andsdk/README.md.🤖 Generated with Claude Code