feat(redis): dlq_channel — one shared DLQ stream, provenance on every entry

[afiodorov]

Summary

Adds an opt-in dlq_channel subscribe option to the Redis transport so many handlers (or many services) can dead-letter into one stream that a single consumer subscribes to, instead of one {channel}.{handler_suffix}.dlq key per handler. Every DLQ entry now also carries provenance in its body (_dlq_source, _dlq_handler, _dlq_at, _dlq_reason, _dlq_retries), is safe to consume (not just inspect), and poison entries are wrapped in a decodable envelope instead of being copied as raw bytes.

Default behaviour without dlq_channel is unchanged. Opened as a draft to discuss the API shape before it lands; see "Questions for maintainers" at the end.

Why

The DLQ is already an XADD into a plain stream with the normal FastStream envelope, so an eggai Channel can consume it today. What makes "subscribe to the DLQ" impractical is everything around that:

1. One key per handler, derived from <agent>-<function>-<n>. A listener has to know every key, and a renamed handler leaves it watching an empty stream. Handlers registered via Channel.subscribe even embed the namespace twice in the key.
2. No origin in the entry. Once several handlers write to one stream, the key no longer says which channel or handler gave up on a message, and the body only has _retry_count / _original_message_id.
3. Poison entries are opaque. An envelope the reclaimer can't parse is copied verbatim; FastStream's parser then falls back to raw bytes, which a typed subscription silently skips and an untyped one can't use.

Concrete driver: a pipeline with six handlers where exhausted retries need to move an email to a "Processing Error" folder via another service. Today that means six fragile subscriptions and no way to tell the entries apart.

What changes

dlq_channel (new subscribe option)

dlq = Channel("dlq", transport=transport)

@agent.subscribe(channel=orders,   retry_on_idle_ms=30_000, max_retries=3, dlq_channel=dlq)
async def handle_order(message): ...

@agent.subscribe(channel=payments, retry_on_idle_ms=30_000, max_retries=3, dlq_channel="dlq")
async def handle_payment(message): ...

@ops_agent.subscribe(channel=dlq, group_start="0")   # picks up the backlog
async def on_dead_letter(message):
    message["_dlq_source"], message["_dlq_handler"], message["_dlq_retries"]

• Agent.subscribe / Channel.subscribe accept a Channel or a topic name; a name is namespaced by going through Channel(name) itself ("dlq" → <EGGAI_NAMESPACE>.dlq). Resolution lives in channel.py (resolve_dlq_channel), so the transport only ever sees a full key and exactly one layer prefixes (the RedisTransport: with a non-default EGGAI_NAMESPACE the reclaimer, retry and DLQ streams use eggai.<ns>.<topic> while consumption uses <ns>.<topic>, so retries never fire #261/Agent.subscribe() with no channel hardcodes "eggai.channel", ignoring EGGAI_NAMESPACE #264 lesson).
• RedisTransport.subscribe takes the full key (consistent with its channel argument).
• Resolution order: explicit dlq_channel, else the existing per-handler default. Both reclaimers of a subscription (main stream and retry stream) target the same DLQ, as before.
• The retry stream stays per handler. Only the terminal sink is shared. Sharing the retry stream is exactly the cross-handler fan-out that Redis transport: shared *.retry stream broadcasts a single group's failures to every other handler on the same channel #225 fixed.
• Validation, at decoration time in agent.py and again in the transport before the broker subscriber is registered (that registration cannot be undone): requires retry_on_idle_ms; requires max_retries not None; rejects the subscribed channel (self-loop); rejects any key ending in .retry (every .retry stream is auto-consumed by some handler, so dead-lettering into one feeds that handler's retry loop).
• Agent.subscribe works on a copy of its kwargs, so one decorator object applied to several handlers no longer re-namespaces an already-resolved key.

A DLQ entry is safe to consume, not just inspect

A DLQ entry is two things at once: the record of a finished retry history, and, once something subscribes to the DLQ, a first delivery to a new consumer. Two rules follow (_stamp_dead_letter in pending_reclaimer.py):

• _retry_count is reset to "0" on the DLQ write; the exhausted budget moves to _dlq_retries (== max_retries, "0" for poison). Without this a DLQ consumer with its own retry_on_idle_ms inherited an already-exceeded count: zero retries, dead-lettered again on its first transient failure, and with backoff a first reclaim threshold of base × multiplier^exhausted.
• Provenance is set-if-absent. If that consumer itself gives up and dead-letters the entry into its own DLQ, the original _dlq_source / _dlq_handler / _dlq_at survive instead of being overwritten with the consumer's own.

Key	Value
_dlq_source	full key of the channel the handler subscribed to (not the retry stream the reclaimer was scanning)
_dlq_handler	handler suffix / consumer group
_dlq_at	ISO-8601 UTC
_dlq_reason	"max_retries" or "poison"
_dlq_retries	retries that actually ran before giving up
_retry_count	reset to "0"

Keys go in the JSON body rather than as extra XADD fields because FastStream's decoder reads only __data__. ReclaimerConfig gains source_stream and handler to carry this. The on_dlq callback receives the same body dict a subscriber would see.

No MAXLEN on a shared DLQ

XADD MAXLEN trims the whole stream regardless of which writer appended or whether an entry was consumed. With several services writing to one DLQ, each with its own retry_max_len, the smallest cap among them would silently delete the others' unconsumed dead letters. ReclaimerConfig now has a separate dlq_max_len; the transport passes RedisTransport(dlq_max_len=...) (default None, i.e. untrimmed, retention is the sink's job) for an explicit dlq_channel and retry_max_len for the per-handler default, as before.

Poison entries are wrapped, and more of them are caught

_poison_body builds a fresh envelope whose body holds the _dlq_* fields, _original_message_id, a zero budget, and the original bytes as _dlq_raw_b64. Envelope headers (correlation id, traceparent) are kept when only the body was unusable. Every DLQ entry now decodes to a JSON object.

Poison detection (_parse_entry) also now covers three cases that previously escaped it, all pre-existing on main: a parseable envelope with a non-object JSON body or a non-integer _retry_count raised outside the guarded parse and aborted the whole reclaim cycle after XCLAIM, head-of-line blocking every later pending entry; an entry with no __data__ field (non-eggai producer) ping-ponged between main and retry stream forever with a count that could never grow.

Refactor

The envelope is parsed once and encoded once per reclaimed entry. _inject_retry_metadata now returns (headers, body, new_count) | None; the DLQ path merges provenance into the same dict and encodes; on_dlq gets that dict. _encode_envelope is shared by all writers (the #263 byte-length fix is preserved and still covered by its test).

Backward compatibility

Change	Effect on existing users
dlq_channel option	none unless passed; default key unchanged
_dlq_* body keys	extra keys in DLQ entries only; typed models ignore them, same precedent as _retry_count
_retry_count reset to "0" on DLQ entries	anything reading the exhausted count off a DLQ entry should read _dlq_retries; on_dlq's retry_count argument is unchanged
poison entries wrapped	per-handler DLQ entries for unparseable envelopes change from raw bytes to a JSON body with the bytes base64'd; on_dlq gets that dict instead of {b"__data__": bytes}
.retry rejected as dlq_channel	only affects the new option
retry stream key and cap	unchanged

Kafka and in-memory transports are untouched; they have no retry/DLQ machinery and already don't accept the other retry_* kwargs.

Version: 0.5.0 (agreed in review), to flag the DLQ-entry content changes to existing users.

Files

• eggai/transport/pending_reclaimer.py — source_stream/handler/dlq_max_len on ReclaimerConfig; _encode_envelope, _parse_entry, _encode_body, _stamp_dead_letter, _poison_body, _fields_as_json; single parse/encode in _reclaim_once; _invoke_on_dlq takes the body dict
• eggai/transport/redis.py — pop + validate dlq_channel before broker registration; choose dlq_stream / dlq_max_len; pass provenance into both _setup_reclaimer calls; docstrings
• eggai/channel.py — resolve_dlq_channel; Channel.subscribe resolves the option
• eggai/agent.py — kwargs copied per application; decoration-time validation and resolution
• tests/test_redis_shared_dlq.py — new module, 20 tests (below); tests/test_redis.py — three tests adapted to the new return shape / _dlq_retries
• README.md, docs/docs/sdk/redis-transport.md, CHANGELOG.md

Tests

New tests/test_redis_shared_dlq.py (auto-skipped without Redis on 6379, like test_redis.py):

• pure helpers: resolve_dlq_channel namespacing; _stamp_dead_letter resets the budget and keeps the first origin on a second dead-lettering; _parse_entry rejects garbage, list bodies and non-integer counts; _poison_body round-trips the raw bytes, keeps headers, zero budget; _dlq_metadata records the original channel from a retry-stream config
• wiring: two handlers on two channels → four reclaimer configs, one dlq_stream, two distinct retry streams, correct (source_stream, handler) pairs; shared DLQ has dlq_max_len=None while retry streams and per-handler DLQs keep the cap; default unchanged when the option is absent; every validation error at transport level (incl. another handler's .retry, and nothing registered afterwards), at Agent.subscribe decoration time, and Channel.subscribe resolution; decorator reuse does not double-namespace
• end-to-end against Redis: two failing handlers dead-letter into one stream, the per-handler DLQ keys are never created, a sink that starts after the failures consumes both with group_start="0" and sees _dlq_source / _dlq_handler / _dlq_reason / _dlq_retries / _retry_count == "0"
• poison: an unparseable __data__ and an entry with no __data__ left in the PEL are both dead-lettered as decodable envelopes with _dlq_reason="poison" and _dlq_raw_b64, the originals are acked, nothing hits the retry stream

Locally: full tests/ suite green against Redis (105 passed, Kafka tests skipped), ruff check/format --check clean, mypy eggai clean. CI on the first commit was green across 3.10–3.13.

Review history on this PR

QualOps' second round (on the second commit) reported four highs. Two are false positives: _read_retry_count does read the stream the entry lives in (the retry reclaimer is configured with stream=retry_stream), and the envelope header count is pairs, exactly what FastStream's parser loops over (covered by the non-ASCII round-trip test). The other two are addressed in the fourth commit: retry_on_idle_ms without a consumer group is now a ValueError (pre-existing gap, direct transport callers only), and a dlq_channel string that already starts with the namespace is rejected instead of being namespaced twice.

Two independent reviews ran on the first commit (a high-effort local review and QualOps). Everything they confirmed is addressed in the follow-up commits: shared-DLQ MAXLEN trimming, _retry_count carry-over and provenance overwrite, retry-stream collision checked after broker registration and only against the handler's own stream, decorator kwargs mutation, poison entries without _retry_count, non-object bodies / missing __data__ escaping the poison path, on_dlq poison shape undocumented, duplicated namespacing rule, triple parse/encode.

Open questions, resolved in review

1. API shape: per-subscription dlq_channel only; a transport-level default can come later without breaking this form.
2. _retry_count reset applies to every DLQ entry, shared or per-handler, so both have one shape.
3. Poison wrapping likewise.
4. Version: 0.5.0, to flag the DLQ-entry content change to existing users. Release itself happens in the usual separate release PR; the changelog entry above opens with a migration note for DLQ consumers.
5. Error text in DLQ entries stays a follow-up (needs a handler-layer wrapper plus a Redis side key).

Still open, to be discussed offline: retention of a shared DLQ under a poison flood (scheduled XTRIM guidance vs. an opt-in RedisTransport(dlq_max_len=...)).

🤖 Generated with Claude Code

https://claude.ai/code/session_01K48zTJV8n9ecRV4x9ME5Sv

[github-actions]

QualOps Code Quality Analysis

Status: ⚠️ FAILED - Critical or high severity issues found

Summary

• Total Issues: 14
• Critical: 0 🔴
• High: 1 🟠
• Medium: 5 🟡
• Low: 8 🟢
• Files Analyzed: 9

🟠 High Issues (1)

• sdk/eggai/transport/pending_reclaimer.py:592 - bug
  The on_dlq callback for the max-retries DLQ path receives new_count (the would-be next retry) instead of the actual number of retries that ran (new_count - 1).

🟡 Medium Issues (5)

• docs/docs/sdk/redis-transport.md:1 - security
  _stamp_dead_letter uses setdefault for provenance fields, letting any message publisher inject fake _dlq_source, _dlq_handler, and _dlq_at into DLQ entries

• docs/docs/sdk/redis-transport.md:1 - security
  Publisher-controlled _retry_count in the message body is read without bounds-checking, allowing a message to be immediately DLQ'd without any retry by setting _retry_count to a value exceeding max_retries

• docs/docs/sdk/redis-transport.md:1 - bug
  _parse_entry catches all exceptions silently and returns None; the caller's warning log has no exc_info, making it impossible to diagnose why a valid-looking message was treated as poison

...and 2 more medium issues

📊 Full Report

View detailed report

---

Powered by QualOps

[tsiaeggai]

Reviewed all four commits and ran the branch locally. The code does what the description says, I found no correctness bug, and the default (no dlq_channel) path is unchanged apart from the documented DLQ-entry content changes. Inline comments are nits and one design note; none block merging once the draft is ready.

Verification

• Scratch venv with faststream 0.7.5 (the poetry.lock pin), redis-py 8.1, Python 3.12, Redis 7 in Docker.
• ruff check / ruff format --check: clean. mypy eggai: clean.
• pytest tests: 105 passed, 13 skipped (Kafka). Includes the 17 new tests in test_redis_shared_dlq.py and the three adapted assertions in test_redis.py.
• Diff is on top of current main (merge-base == main).

Things I specifically checked

• Double namespacing: Agent.subscribe resolves dlq_channel to a full key and Channel.subscribe resolves too, so I traced whether a key could be resolved twice. Agent.start() calls the transport directly, not Channel.subscribe, so no. The decorator-reuse test covers the other way that could happen.
• Validation runs before StreamSub / broker.subscriber() and before _stream_subscriptions.add, so a rejected subscribe leaves nothing behind. Confirmed by the test that asserts no h-4/h-4b groups are registered.
• The recursive _internal_retry subscribe does not receive dlq_channel (popped earlier), and both reclaimers of one handler get the same dlq_stream, source_stream=channel and handler=handler_suffix.
• ReclaimerConfig registry key is (stream, group, consumer), so two handlers sharing one DLQ cannot collide.
• _parse_entry now also guards _read_retry_count, so a non-integer _retry_count is treated as count 0 and then poisoned on claim rather than skipped forever.
• Downstream compatibility: I checked two production services I maintain that use on_dlq and read _original_message_id / _retry_count. Both keep working: the retry_count argument is unchanged, _original_message_id is still present, and _retry_count is only read off live deliveries, not DLQ entries. Their integration tests only assert that _retry_count is present in the DLQ payload, which stays true.

Answers to the maintainer questions, for what they're worth

1. Per-subscription only is the right first cut. A transport-level default can be added later without breaking this shape.
2. Resetting _retry_count everywhere is the consistent choice. Scoping it to shared DLQs would make the per-handler and shared entry shapes differ, which is worse than the one-time content change.
3. Same for poison wrapping.
4. I'd go 0.5.0. The _retry_count reset and the poison shape are observable by existing users (any redrive tooling or assertion reading the exhausted count), and a minor bump is the cheapest way to flag that. Functionally both are improvements.
5. Agree that error text belongs in a follow-up.

One design note, not a blocker

A shared DLQ is written with no MAXLEN, for good reasons. Combined with the new poison wrapping, a producer that floods a stream with unparseable payloads now grows the shared DLQ without bound, each entry carrying the raw bytes base64-encoded. The docs say retention is the sink's job, which is fine, but it may be worth an explicit sentence recommending a scheduled XTRIM and, later, an opt-in RedisTransport(dlq_max_len=...) for operators who prefer a hard cap over the risk.

[tsiaeggai]

Re-checked 371def8 and 9a777cc. All four threads addressed as discussed and resolved:

• on_dlq comment updated.
• _stamp_dead_letter: origin keys first-wins, hop keys (_dlq_reason / _dlq_retries) overwritten; docstring, test, README, docs and changelog match.
• RedisTransport(dlq_max_len=...) opt-in cap for shared DLQs, default None, per-handler DLQs keep retry_max_len. Happy to settle the default and XTRIM guidance on the call.
• Namespace-prefixed string rejection kept, rationale makes sense.

Re-ran locally with faststream 0.7.5 (lock) against Redis 7: ruff, mypy and 105 tests green; CI green on 3.10–3.13. The changelog migration note covers everything a DLQ reader needs. Downstream services I checked remain compatible.

On the QualOps high (on_dlq gets new_count rather than new_count - 1): that is pre-existing, documented behaviour and changing it would alter what existing callbacks log, so I'd leave it.

Approving.