diff --git a/docs/docs/sdk/redis-transport.md b/docs/docs/sdk/redis-transport.md index 5ac71fa4..d0e6b860 100644 --- a/docs/docs/sdk/redis-transport.md +++ b/docs/docs/sdk/redis-transport.md @@ -226,6 +226,61 @@ The callback can be sync or async. Errors in the callback are logged but never p !!! note "At-least-once applies to DLQ writes too" The `XADD` (to DLQ) and `XACK` are not atomic. A crash between them can produce duplicate DLQ entries — and fire `on_dlq` more than once — for the same logical message. Re-drive tooling and `on_dlq` callbacks should deduplicate using `_original_message_id`. +#### Shared DLQ channel (`dlq_channel`) + +A DLQ stream is a plain Redis stream with the same envelope as any channel, so an eggai subscriber can consume it. What makes that awkward by default is the naming: one key per handler, derived from the agent name, the function name and a counter. Rename a handler and a listener silently watches an empty stream. + +`dlq_channel` replaces the per-handler key with one you choose, so every handler — across services, if they share the value — dead-letters into a single stream that one consumer can subscribe to: + +```python +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): ... +``` + +`dlq_channel` accepts a `Channel` or a topic name; a name is namespaced exactly like `Channel(name)`, so `"dlq"` becomes `.dlq`. Only the terminal sink is shared — each handler keeps its own `.retry` stream (sharing it would re-introduce the cross-handler fan-out fixed in #225). + +The consumer is an ordinary subscription. Two things to get right: + +```python +@ops_agent.subscribe( + channel=dlq, + group_start="0", # new group starts at the beginning: failures that happened +) # before this service booted are picked up, not skipped +async def on_dead_letter(message): + # No data_type here: the shared stream carries every payload type. + # Dispatch on message["type"] or on the provenance below. + source = message["_dlq_source"] # e.g. "eggai.orders" + handler = message["_dlq_handler"] # e.g. "order-service-handle_order-1" + ... +``` + +Do **not** give the sink `retry_on_idle_ms` together with the same `dlq_channel`: a handler dead-lettering into its own input is a loop, and `subscribe()` rejects it. If the sink needs retries, let it default to its own per-handler DLQ. + +**Provenance on every DLQ entry.** Because a shared stream's key no longer says where an entry came from, the reclaimer stamps it into the JSON body (body, not extra `XADD` fields: FastStream's decoder reads only `__data__`, so body keys are all a subscriber sees). This applies to per-handler DLQs too. + +| Field | Value | +|-------|-------| +| `_dlq_source` | full key of the channel the handler subscribed to | +| `_dlq_handler` | handler suffix / consumer group | +| `_dlq_at` | ISO-8601 UTC timestamp of the DLQ write | +| `_dlq_reason` | `"max_retries"`, or `"poison"` for an unparseable envelope | +| `_dlq_retries` | how many retries actually ran before giving up (`max_retries`; `"0"` for poison) | +| `_retry_count` | **reset to `"0"`** on the DLQ write | +| `_original_message_id` | as on retry delivery | + +Two rules make a DLQ entry safe to *consume*, not just inspect. `_retry_count` is reset so a DLQ consumer that has its own `retry_on_idle_ms` starts with a fresh budget — otherwise it would inherit an already-exceeded count, be dead-lettered again on its first transient failure and, with backoff, wait `base × multiplier^exhausted` for its first reclaim. And when that consumer does give up and dead-letters the entry into its own DLQ, `_dlq_source` / `_dlq_handler` / `_dlq_at` keep the original failure (set-if-absent) while `_dlq_reason` / `_dlq_retries` describe the latest hop, so neither piece of information is lost. + +**No `MAXLEN` on a shared DLQ.** `retry_max_len` caps the retry streams and per-handler DLQs, not a shared `dlq_channel`. `XADD MAXLEN` trims the whole stream regardless of which writer appended, so with several services writing to one DLQ the smallest cap among them would silently delete the others' unconsumed dead letters. A shared DLQ is written untrimmed by default; retention is the sink's responsibility (`XTRIM`, or a scheduled trim once entries are processed). `RedisTransport(dlq_max_len=...)` is the opt-in hard ceiling for operators who prefer bounded memory over the risk; use the same value on every writer to that DLQ. + +**Poison entries are wrapped.** An envelope the reclaimer cannot parse used to be copied to the DLQ verbatim; FastStream's parser then falls back to raw bytes, which a typed subscription silently skips and an untyped one cannot use. It is now written as a fresh envelope whose body holds the fields above plus the original bytes as `_dlq_raw_b64`, so every DLQ entry decodes to a JSON object. Re-drive tooling that handled raw poison entries should read `_dlq_raw_b64` instead. + +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. + ### Tuning the Reclaimer ```python diff --git a/sdk/CHANGELOG.md b/sdk/CHANGELOG.md index 37e45520..4235295c 100644 --- a/sdk/CHANGELOG.md +++ b/sdk/CHANGELOG.md @@ -7,6 +7,89 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0 ## [Unreleased] +**Migration note for DLQ consumers.** The *shape of DLQ entries* changes in +this release; the SDK API, retry-stream entries, handler-side +`_retry_count` / `_original_message_id`, stream key names and per-handler +trimming do not. If you read DLQ entries yourself (`XRANGE`, re-drive +scripts, monitoring) or read the dict passed to `on_dlq`: + +- read the exhausted retry budget from `_dlq_retries`, not `_retry_count` + (which is now `"0"` on every DLQ entry); +- expect the additional `_dlq_source`, `_dlq_handler`, `_dlq_at`, + `_dlq_reason` keys; +- poison entries are no longer raw bytes: recover the original payload by + base64-decoding `_dlq_raw_b64`, and `on_dlq` receives that decoded dict; +- stream entries with no `__data__` field (only possible from a non-eggai + producer) are no longer retried at all; they are dead-lettered as poison on + the first reclaim. + +The `on_dlq(body, msg_id, retry_count)` signature and its `retry_count` +argument are unchanged. + +### Added +- `RedisTransport`: new `dlq_channel` subscribe option. Dead-letters into ONE + stream you name instead of the per-handler `{channel}.{handler_suffix}.dlq`, + so a single consumer can subscribe to every handler's (or every service's) + failures with a plain `@agent.subscribe(channel=dlq, group_start="0")`. + Accepts a `Channel` or a topic name (namespaced like `Channel(name)`) on + `Agent.subscribe` / `Channel.subscribe`; a full key on + `RedisTransport.subscribe`. Requires `retry_on_idle_ms` and a non-`None` + `max_retries`; rejects the subscribed channel and the handler's retry stream + as targets. The `.retry` stream stays per-handler (#225) — only the terminal + sink is shared. A shared DLQ is written without `MAXLEN` by default + (`retry_max_len` keeps applying to retry streams and per-handler DLQs): + stream-wide trimming would let one writer's cap delete other writers' + unconsumed dead letters. `RedisTransport(dlq_max_len=...)` is the opt-in + hard ceiling for a shared DLQ. Default behaviour without `dlq_channel` is + unchanged. +- Every DLQ entry now carries provenance in its JSON body, alongside the + existing `_retry_count` / `_original_message_id`: `_dlq_source` (the channel + key the handler subscribed to), `_dlq_handler` (handler suffix / consumer + group), `_dlq_at` (ISO-8601 UTC) and `_dlq_reason` (`"max_retries"` or + `"poison"`) and `_dlq_retries` (how many retries actually ran). Additive: + typed models ignore the extra keys, as they already do for `_retry_count`. + On an entry dead-lettered a second time (by a DLQ consumer that gave up) + the origin keys `_dlq_source` / `_dlq_handler` / `_dlq_at` keep the first + failure, while `_dlq_reason` / `_dlq_retries` describe the latest hop. + +### Changed +- **`_retry_count` is reset to `"0"` on the DLQ write** (the exhausted budget + moves to `_dlq_retries`). A DLQ entry is also a *first* delivery to whatever + consumes the DLQ; carrying the exceeded count over meant a DLQ consumer with + its own `retry_on_idle_ms` got zero retries and, with backoff, an escalated + first reclaim. Anything that read `_retry_count` off a DLQ entry should read + `_dlq_retries`; the `on_dlq` callback's `retry_count` argument is unchanged. +- Poison messages (envelopes the reclaimer cannot parse) are no longer copied to + the DLQ verbatim. They are wrapped in a well-formed envelope whose body holds + the `_dlq_*` fields, `_original_message_id`, and the original bytes as + `_dlq_raw_b64`, so a DLQ subscriber always receives a decodable JSON object + instead of raw bytes. Envelope headers are preserved when only the body was + unusable. Re-drive scripts that handled raw poison entries should read + `_dlq_raw_b64`. Consequently `on_dlq` now receives that decoded dict for + poison entries too, instead of the raw `{b"__data__": bytes}` fields. +- `Agent.subscribe` / `Channel.subscribe` reject a `dlq_channel` *string* that + already starts with `EGGAI_NAMESPACE` (e.g. a pasted `channel.get_name()`), + which would otherwise be namespaced twice and route dead letters to an + unwatched `..dlq`. Pass the bare topic name or a `Channel`. +- `RedisTransport.subscribe` rejects `retry_on_idle_ms` without a consumer + group (`handler_id=` / `group=`): the reclaimer works on a group's PEL, so + without one every reclaim cycle just errored. `Agent`/`Channel` always set a + group; this only affects direct transport callers. +- `dlq_channel` rejects any key ending in `.retry`, not just the handler's own + retry stream: every `.retry` stream is auto-consumed by some handler, so + dead-lettering into one would feed that handler's retry loop. The check runs + before the broker subscriber is registered. + +### Fixed +- `PendingReclaimer`: a pending entry with a parseable envelope but a non-object + JSON body (e.g. a published list), a non-integer `_retry_count`, or no + `__data__` field at all (non-eggai producer) escaped the poison handling. The + first two raised outside the guarded parse and aborted the whole reclaim + cycle after `XCLAIM`, head-of-line blocking every later pending entry; the + third ping-ponged between main and retry stream forever with a retry count + that could never grow. All three are now dead-lettered as poison; in + particular, entries without `__data__` are no longer retried at all. + ## [0.4.1] - 2026-09-07 ### Fixed diff --git a/sdk/README.md b/sdk/README.md index c1aebd17..534a7292 100644 --- a/sdk/README.md +++ b/sdk/README.md @@ -188,6 +188,59 @@ the per-handler `{channel}.{handler_suffix}.dlq`. The DLQ is terminal — no aut reclaimer. Set `max_retries=None` for unlimited retries. An optional `on_dlq` callback fires when a message lands in the DLQ. +**Shared DLQ channel:** A DLQ stream is an ordinary stream, so an eggai subscriber can +consume it — but with one key per handler there is nothing stable to subscribe to. Pass +`dlq_channel` to dead-letter every handler (or every service) into **one** stream instead: + +```python +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): ... # a topic name is namespaced like Channel("dlq") + +# One consumer for all of them — start the group at "0" to pick up failures that +# happened before this service booted. +@ops_agent.subscribe(channel=dlq, group_start="0") +async def on_dead_letter(message): + print(message["_dlq_source"], message["_dlq_handler"], message["_dlq_retries"]) +``` + +Only the terminal sink is shared; each handler keeps its own `.retry` stream (a shared +retry stream would re-introduce the cross-handler fan-out of #225). `dlq_channel` requires +`retry_on_idle_ms` and a non-`None` `max_retries`, and must not be the channel the handler +subscribes to. Don't give the sink itself `retry_on_idle_ms` with the same `dlq_channel`. + +Every DLQ entry — shared or per-handler — carries provenance in its body so a shared +consumer can tell entries apart: + +| Field | Value | +|-------|-------| +| `_dlq_source` | full key of the channel the handler subscribed to (e.g. `eggai.orders`) | +| `_dlq_handler` | handler suffix / consumer group (e.g. `order-service-handle_order-1`) | +| `_dlq_at` | ISO-8601 UTC timestamp of the DLQ write | +| `_dlq_reason` | `"max_retries"`, or `"poison"` for an envelope the reclaimer could not parse | +| `_dlq_retries` | how many retries actually ran before giving up (`max_retries`; `"0"` for poison) | +| `_retry_count` | **reset to `"0"`** on the DLQ write, so a DLQ consumer with its own `retry_on_idle_ms` starts with a fresh budget | +| `_original_message_id` | as on retry delivery | + +If a DLQ consumer itself gives up on an entry and dead-letters it again, `_dlq_source`, +`_dlq_handler` and `_dlq_at` keep the *original* failure while `_dlq_reason` and +`_dlq_retries` describe the latest hop. + +A shared DLQ is written **without** `MAXLEN` by default — `retry_max_len` applies to the +retry streams and to per-handler DLQs only. `XADD MAXLEN` trims the whole stream regardless +of which writer appended, so one service's cap would silently delete other services' +unconsumed dead letters. Retention of a shared DLQ is the sink's job (`XTRIM`, or ack and +trim on a schedule). Operators who prefer a hard memory ceiling can opt in with +`RedisTransport(dlq_max_len=...)`; set the same value on every writer to that DLQ. + +A poison entry (unparseable envelope) is no longer copied to the DLQ verbatim: it is +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. + **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 @@ -238,8 +291,9 @@ Retry stream (eggai.orders.order-service-handle_order-1.retry) ← per-handler _retry_count ≤ max_retries → XADD → XACK _retry_count > max_retries → XADD → XACK -DLQ stream (eggai.orders.order-service-handle_order-1.dlq) ← per-handler - terminal — no reclaimer, manual re-drive only +DLQ stream (eggai.orders.order-service-handle_order-1.dlq) ← per-handler by default, + terminal — no reclaimer or one shared key via dlq_channel= + entries carry _dlq_source / _dlq_handler / _dlq_at / _dlq_reason ``` ## Observability (OpenTelemetry) diff --git a/sdk/eggai/agent.py b/sdk/eggai/agent.py index dbb25579..3cc8c807 100644 --- a/sdk/eggai/agent.py +++ b/sdk/eggai/agent.py @@ -4,7 +4,7 @@ from collections.abc import Callable from typing import Any, TypeVar -from .channel import Channel +from .channel import Channel, resolve_dlq_channel from .hooks import eggai_register_stop from .transport import get_default_transport from .transport.base import Transport @@ -68,7 +68,7 @@ def _get_transport(self): self._transport = get_default_transport() return self._transport - def subscribe(self, channel: Channel | None = None, **kwargs): + def subscribe(self, channel: Channel | None = None, **subscribe_kwargs): """ Decorator for adding a subscription. @@ -81,6 +81,11 @@ def subscribe(self, channel: Channel | None = None, **kwargs): """ def decorator(handler: Callable[[dict[str, Any]], "asyncio.Future"]): + # Work on a copy: the decorator object may be applied to several + # handlers, and the resolution/popping below must not leak from one + # application into the next (e.g. re-namespacing an already-resolved + # dlq_channel into "..dlq"). + kwargs = dict(subscribe_kwargs) channel_name = (channel or Channel()).get_name() if "min_idle_time" in kwargs and "retry_on_idle_ms" in kwargs: raise ValueError( @@ -122,8 +127,41 @@ def decorator(handler: Callable[[dict[str, Any]], "asyncio.Future"]): "idle threshold); a cap below the base would disable backoff " "entirely." ) + # Plugins see the kwargs as the caller wrote them (before the DLQ + # key below is resolved to its namespaced form). original_kwargs = kwargs.copy() + # Shared DLQ: resolve Channel / topic name to a full key here (the + # transport only knows keys) and fail at decoration time, like the + # other retry knobs, so a typo can't silently no-op. + if kwargs.get("dlq_channel") is not None: + if "retry_on_idle_ms" not in kwargs: + raise ValueError( + "dlq_channel requires retry_on_idle_ms to be set. " + "Set retry_on_idle_ms to enable SDK-managed retries with a DLQ." + ) + if "max_retries" in kwargs and kwargs["max_retries"] is None: + raise ValueError( + "dlq_channel requires max_retries: max_retries=None disables " + "the DLQ entirely, so there is nothing to route to the shared " + "channel." + ) + dlq_key = resolve_dlq_channel(kwargs["dlq_channel"]) + if dlq_key is None: # unreachable: input checked non-None above + raise ValueError("dlq_channel could not be resolved to a key") + if dlq_key == channel_name: + raise ValueError( + f"dlq_channel {dlq_key!r} is the channel this handler " + "subscribes to; dead-lettering into its own input would loop." + ) + if dlq_key.endswith(".retry"): + raise ValueError( + f"dlq_channel {dlq_key!r} names an SDK-managed retry stream " + "('.retry' suffix); dead-lettering into a retry stream would " + "feed a handler's retry loop. Pick a different name." + ) + kwargs["dlq_channel"] = dlq_key + # Extract plugin-specific kwargs dynamically and clean them from kwargs plugin_found_keys = set() for plugin_name in PLUGIN_LIST: diff --git a/sdk/eggai/channel.py b/sdk/eggai/channel.py index c518e335..53be3185 100644 --- a/sdk/eggai/channel.py +++ b/sdk/eggai/channel.py @@ -23,6 +23,41 @@ DEFAULT_CHANNEL_NAME = "channel" +def resolve_dlq_channel(value: "Channel | str | None") -> str | None: + """Turn the ``dlq_channel`` subscribe option into a full stream key. + + ``Agent.subscribe`` / ``Channel.subscribe`` accept either a :class:`Channel` + (its namespaced name is used as-is) or a bare *topic name* — not a full + key — which is namespaced by going through ``Channel(name)`` itself, so + ``"dlq"`` becomes ``".dlq"``. The transport layer only ever + sees the full key. Reusing ``Channel`` rather than re-spelling the prefix + keeps the namespacing rule in one place (see #261/#264 for what happens + when two places both try to prefix). Pass a ``Channel`` if you already hold + a full key. + """ + if value is None: + return None + if isinstance(value, Channel): + return value.get_name() + if isinstance(value, str) and value: + # A string that already carries the namespace is almost certainly a full + # key pasted by mistake (e.g. `dlq.get_name()`); namespacing it again + # would route dead letters to "..dlq", a stream nobody watches. + # Fail loudly instead: the Channel form exists for full keys. + if value.startswith(f"{NAMESPACE}."): + raise ValueError( + f"dlq_channel {value!r} already starts with the namespace " + f"{NAMESPACE!r}; pass the bare topic name " + f"({value[len(NAMESPACE) + 1 :]!r}) or a Channel instance." + ) + # Channel.__init__ is inert (transport is lazy, no stop hook until it + # connects), so this is a pure name computation. + return Channel(value).get_name() + raise ValueError( + f"dlq_channel must be a Channel or a non-empty topic name string, got {value!r}" + ) + + class Channel: """ A channel that publishes messages to a given 'name' on its own Transport. @@ -101,6 +136,8 @@ async def subscribe( ) HANDLERS_IDS[handler_name] += 1 kwargs["handler_id"] = f"{handler_name}-{HANDLERS_IDS[handler_name]}" + if kwargs.get("dlq_channel") is not None: + kwargs["dlq_channel"] = resolve_dlq_channel(kwargs["dlq_channel"]) await self._get_transport().subscribe(self._name, callback, **kwargs) await self._ensure_connected() diff --git a/sdk/eggai/transport/pending_reclaimer.py b/sdk/eggai/transport/pending_reclaimer.py index db622ca9..2311c0ac 100644 --- a/sdk/eggai/transport/pending_reclaimer.py +++ b/sdk/eggai/transport/pending_reclaimer.py @@ -1,4 +1,6 @@ import asyncio +import base64 +import datetime import json import logging import random @@ -54,10 +56,16 @@ class ReclaimerConfig: dlq_stream: str | None = ( None # full key, e.g. ".orders.order-service-handle_order-1.dlq" ) - on_dlq: Callable | None = None # async or sync callback(fields, msg_id, count) + on_dlq: Callable | None = None # async or sync callback(body, msg_id, retry_count) max_len: int | None = ( - None # cap retry/DLQ stream length (XADD MAXLEN ~); None = unbounded + None # cap retry stream length (XADD MAXLEN ~); None = unbounded ) + # Cap for the DLQ stream, kept separate from `max_len` on purpose: MAXLEN trims + # the whole stream regardless of who wrote an entry or whether it was consumed, + # so on a *shared* DLQ (`dlq_channel`) every writer's cap would apply to every + # other writer's dead letters. The transport therefore passes None for a shared + # DLQ (retention is the sink's job) and `retry_max_len` for the per-handler one. + dlq_max_len: int | None = None # Exponential backoff between retry attempts. The reclaimer treats a PEL entry # as "stale" once it has been idle for `min_idle_ms * (backoff_multiplier ** # retry_count)` (capped at backoff_max_ms). multiplier=1.0 reproduces the @@ -65,48 +73,30 @@ class ReclaimerConfig: backoff_multiplier: float = 1.0 # 1.0 = constant cadence (no escalation) backoff_max_ms: int | None = None # cap on the escalated threshold; None = uncapped backoff_jitter: float = 0.0 # fraction in [0,1]; adds up to this share on top of the threshold at random + # Provenance recorded on every DLQ entry (see _dlq_metadata). `source_stream` + # is the channel the handler subscribed to — NOT `stream`, which for the retry + # reclaimer is the retry stream. `handler` is the handler suffix / consumer + # group. Both matter once several handlers share one DLQ stream + # (`dlq_channel`): the entry itself must say where it came from, because the + # stream key no longer does. + source_stream: str | None = None + handler: str | None = None -def _inject_retry_metadata(data: bytes, msg_id_str: str) -> tuple[bytes, int, bool]: - """ - Inject _retry_count and _original_message_id into a FastStream binary stream - entry's JSON body so the handler can read them for idempotency checks. +def _encode_envelope(headers: dict, body_bytes: bytes) -> bytes: + """Build a FastStream BinaryMessageFormatV1 envelope from headers + JSON body. - FastStream BinaryMessageFormatV1 layout: + Layout (mirrors ``BinaryMessageFormatV1.encode()``): [8B magic][2B version=1][4B headers_start][4B data_start] [2B num_headers]([2B key_len][key][2B val_len][val])* - [JSON body bytes] - - Returns a tuple of (modified __data__ bytes, new retry count, parsed_ok). - On parse failure returns (original data, 0, False): the caller treats an - unparseable envelope as a poison message (its retry count can never be - incremented, so re-queueing it would livelock the retry stream) and routes - it to the DLQ or drops it instead. + [body bytes] """ - try: - body_bytes, headers = BinaryMessageFormatV1.parse(data) - body_dict = json.loads(body_bytes) - except Exception: - logger.warning( - "Failed to inject retry metadata for message %s; treating as poison", - msg_id_str, - exc_info=True, - ) - return data, 0, False - - new_count = int(body_dict.get("_retry_count", "0")) + 1 - body_dict["_retry_count"] = str(new_count) - body_dict.setdefault("_original_message_id", msg_id_str) - new_body_bytes = json.dumps(body_dict, separators=(",", ":")).encode() - - # Re-encode headers section. headers_writer = _BinaryWriter() for key, value in headers.items(): headers_writer.write_string(key) headers_writer.write_string(value) headers_bytes = headers_writer.get_bytes() - # Rebuild the binary envelope — mirrors BinaryMessageFormatV1.encode(). writer = _BinaryWriter() writer.write(BinaryMessageFormatV1.IDENTITY_HEADER) # 8 bytes → len=8 writer.write_short(1) # version=1, 2B → len=10 @@ -116,8 +106,150 @@ def _inject_retry_metadata(data: bytes, msg_id_str: str) -> tuple[bytes, int, bo writer.write_int(data_start) # 4B → len=18 writer.write_short(len(headers)) # 2B → len=20 writer.write(headers_bytes) - writer.write(new_body_bytes) - return writer.get_bytes(), new_count, True + writer.write(body_bytes) + return writer.get_bytes() + + +def _dlq_metadata(config: ReclaimerConfig, reason: str) -> dict[str, str]: + """Provenance keys added to a DLQ entry's JSON body. + + Kept as string values, like ``_retry_count``, so they survive any consumer + that stringifies fields. ``reason`` is ``"max_retries"`` or ``"poison"``. + """ + meta = { + "_dlq_reason": reason, + "_dlq_at": datetime.datetime.now(datetime.timezone.utc).isoformat( + timespec="milliseconds" + ), + } + if config.source_stream is not None: + meta["_dlq_source"] = config.source_stream + if config.handler is not None: + meta["_dlq_handler"] = config.handler + return meta + + +def _parse_entry(data: bytes) -> tuple[dict, dict, int] | None: + """Parse a FastStream envelope into ``(headers, body, retry_count)``. + + Returns ``None`` for anything the reclaimer cannot safely re-queue: bytes + that are not an envelope, a body that is not a JSON object, or a + ``_retry_count`` that is not an integer. Such an entry is *poison* — its + count can never be incremented, so re-queueing it would livelock the retry + stream — and the caller dead-letters it instead of retrying. + """ + try: + body_bytes, headers = BinaryMessageFormatV1.parse(data) + body = json.loads(body_bytes) + if not isinstance(body, dict): + raise TypeError(f"body is {type(body).__name__}, not a JSON object") + retry_count = int(body.get("_retry_count", "0")) + except Exception: + return None + return headers, body, retry_count + + +def _encode_body(headers: dict, body: dict) -> bytes: + """Serialise ``body`` compactly and wrap it in an envelope with ``headers``.""" + return _encode_envelope(headers, json.dumps(body, separators=(",", ":")).encode()) + + +def _inject_retry_metadata( + data: bytes, msg_id_str: str +) -> tuple[dict, dict, int] | None: + """Parse an entry and bump its retry metadata in the body dict. + + Returns ``(headers, body, new_count)`` with ``body["_retry_count"]`` + incremented and ``_original_message_id`` set if absent, so the handler can + use them for idempotency checks. Returns ``None`` (after a warning) for a + poison entry — see :func:`_parse_entry`. The caller encodes the body exactly + once, after deciding whether it goes to the retry stream or the DLQ. + """ + parsed = _parse_entry(data) + if parsed is None: + logger.warning( + "Failed to inject retry metadata for message %s; treating as poison", + msg_id_str, + ) + return None + headers, body, retry_count = parsed + new_count = retry_count + 1 + body["_retry_count"] = str(new_count) + body.setdefault("_original_message_id", msg_id_str) + return headers, body, new_count + + +def _stamp_dead_letter(body: dict, meta: dict[str, str], retries: int) -> None: + """Turn a message body into a DLQ entry body, in place. + + 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. Hence: + + - origin keys (``_dlq_source``, ``_dlq_handler``, ``_dlq_at``) are written + set-if-absent: if the message is dead-lettered a second time (the DLQ + consumer itself gave up), they keep saying where it *originally* failed; + - hop keys (``_dlq_reason``, ``_dlq_retries``) are overwritten on every + dead-lettering: they say why the entry is in *this* DLQ, which is what its + reader needs (the origin is still in the keys above); + - ``_retry_count`` is reset to ``"0"``, so a DLQ consumer with its own + ``retry_on_idle_ms`` starts with a fresh budget instead of inheriting an + already-exceeded one (which would dead-letter it again on its first + failure and, with backoff, make its first reclaim wait + ``base * multiplier ** exhausted``). + + Metadata lives in the *body*, not in extra XADD fields: FastStream's decoder + reads only ``__data__``, so body keys are all a DLQ subscriber can see. + """ + for key in ("_dlq_source", "_dlq_handler", "_dlq_at"): + if key in meta: + body.setdefault(key, meta[key]) # first dead-lettering wins + body["_dlq_reason"] = meta["_dlq_reason"] # latest hop wins + body["_dlq_retries"] = str(retries) + body["_retry_count"] = "0" + + +def _poison_body( + data: bytes, meta: dict[str, str], msg_id_str: str +) -> tuple[dict, dict]: + """Build ``(headers, body)`` for a poison entry from its raw ``__data__``. + + A poison entry copied verbatim would reach a DLQ subscriber as raw bytes + (FastStream's parser falls back to the bare payload), which a typed + subscription silently skips and an untyped one cannot use. Instead the DLQ + entry is a fresh, well-formed envelope whose body carries the provenance, a + zero retry budget, and the original bytes base64-encoded as ``_dlq_raw_b64`` + for forensics and re-drive. Envelope headers (correlation id, traceparent…) + are kept when the envelope itself parsed and only its body was unusable. + """ + try: + _, headers = BinaryMessageFormatV1.parse(data) + except Exception: + headers = {} + body: dict = { + "_original_message_id": msg_id_str, + "_dlq_raw_b64": base64.b64encode(data).decode("ascii"), + } + _stamp_dead_letter(body, meta, 0) + return headers, body + + +def _fields_as_json(fields: dict) -> bytes: + """Render a stream entry that has no ``__data__`` field as JSON bytes. + + Such entries can only come from a non-eggai producer; they are wrapped as + poison so they reach the DLQ instead of ping-ponging through the retry + stream forever (their retry count can never be incremented). + """ + + def _text(value: Any) -> Any: + return ( + value.decode("utf-8", errors="replace") + if isinstance(value, bytes) + else value + ) + + return json.dumps({_text(k): _text(v) for k, v in fields.items()}).encode() class PendingReclaimerManager: @@ -322,8 +454,8 @@ async def _read_retry_count(self, stream: str, msg_id: Any) -> int: data = fields.get(b"__data__") if data is None: return 0 - body_bytes, _headers = BinaryMessageFormatV1.parse(data) - return int(json.loads(body_bytes).get("_retry_count", "0")) + parsed = _parse_entry(data) + return parsed[2] if parsed is not None else 0 except Exception: return 0 @@ -385,28 +517,38 @@ async def _reclaim_once(self, config: ReclaimerConfig) -> None: message_ids=stale_ids, ) + data_key = b"__data__" for msg_id, fields in claimed: msg_id_str = msg_id.decode() if isinstance(msg_id, bytes) else msg_id - # Inject retry metadata into the FastStream binary payload body so the - # handler sees _retry_count and _original_message_id in the message dict. + # Parse once and bump the retry metadata in the body dict; the body is + # encoded exactly once below, after routing. # NOTE: XADD then XACK is not atomic. A crash here re-delivers on the # next cycle (at-least-once). Use _original_message_id to deduplicate. - data_key = b"__data__" - new_count = 0 - parsed_ok = True - if data_key in fields: - fields[data_key], new_count, parsed_ok = _inject_retry_metadata( - fields[data_key], msg_id_str - ) + parsed = ( + _inject_retry_metadata(fields[data_key], msg_id_str) + if data_key in fields + else None + ) - # Poison message: a permanently-unparseable envelope can never have its - # retry count incremented, so re-queueing it to the retry stream would - # livelock forever. Route it to the DLQ if configured, otherwise drop it - # (XACK) with a loud warning rather than spin on it indefinitely. - if not parsed_ok: + # Poison: no __data__ field, not an envelope, a non-object body, or a + # non-integer _retry_count. Its count can never be incremented, so + # re-queueing it would livelock the retry stream. Route it to the DLQ + # (wrapped so it decodes) if configured, otherwise drop it (XACK) with + # a loud error rather than spin on it indefinitely. + if parsed is None: if config.dlq_stream is not None: - await self._xadd(config.dlq_stream, fields, config.max_len) + raw = ( + fields[data_key] + if data_key in fields + else _fields_as_json(fields) + ) + headers, body = _poison_body( + raw, _dlq_metadata(config, "poison"), msg_id_str + ) + 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) logger.warning( "Message %s has an unparseable envelope; moved to DLQ %s " @@ -414,9 +556,7 @@ async def _reclaim_once(self, config: ReclaimerConfig) -> None: msg_id_str, config.dlq_stream, ) - await self._invoke_on_dlq( - config, fields, data_key, msg_id_str, new_count - ) + await self._invoke_on_dlq(config, body, msg_id_str, 0) else: await self._client.xack(config.stream, config.group, msg_id) logger.error( @@ -426,13 +566,22 @@ async def _reclaim_once(self, config: ReclaimerConfig) -> None: ) continue + headers, body, new_count = parsed + # Route to DLQ if max retries exceeded, otherwise to retry stream. if ( config.max_retries is not None and config.dlq_stream is not None and new_count > config.max_retries ): - await self._xadd(config.dlq_stream, fields, config.max_len) + # new_count is the retry that would have run next; the retries + # that actually ran are new_count - 1 (== max_retries). + _stamp_dead_letter( + body, _dlq_metadata(config, "max_retries"), new_count - 1 + ) + 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) logger.warning( "Message %s exceeded max_retries=%d; moved to DLQ %s", @@ -440,10 +589,9 @@ async def _reclaim_once(self, config: ReclaimerConfig) -> None: config.max_retries, config.dlq_stream, ) - await self._invoke_on_dlq( - config, fields, data_key, msg_id_str, new_count - ) + await self._invoke_on_dlq(config, body, msg_id_str, new_count) 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) logger.debug("Reclaimed %s → %s", msg_id_str, config.retry_stream) @@ -451,28 +599,23 @@ async def _reclaim_once(self, config: ReclaimerConfig) -> None: async def _invoke_on_dlq( self, config: ReclaimerConfig, - fields: dict, - data_key: bytes, + body: dict, msg_id_str: str, - new_count: int, + retry_count: int, ) -> None: - """Invoke the optional on_dlq callback with a parsed message dict. - - Parses the binary envelope so the callback receives a plain dict instead - of raw bytes; falls back to the raw fields if parsing fails. Callback - errors are logged but never block the DLQ write that already happened. + """Invoke the optional on_dlq callback with the DLQ entry's body dict. + + The dict is exactly what a subscriber on the DLQ stream receives: the + payload plus ``_retry_count`` (reset to "0"), ``_dlq_retries``, + ``_original_message_id`` and the ``_dlq_*`` provenance; for a poison + entry the same metadata plus ``_dlq_raw_b64`` instead of a payload. + Callback errors are logged but never block the DLQ write that already + happened. """ if config.on_dlq is None: return try: - parsed_msg: dict | bytes = fields - if data_key in fields: - try: - body_bytes, _ = BinaryMessageFormatV1.parse(fields[data_key]) - parsed_msg = json.loads(body_bytes) - except Exception: - parsed_msg = fields - result = config.on_dlq(parsed_msg, msg_id_str, new_count) + result = config.on_dlq(body, msg_id_str, retry_count) if asyncio.iscoroutine(result): await result except Exception: diff --git a/sdk/eggai/transport/redis.py b/sdk/eggai/transport/redis.py index 009eb6b6..6ee464bd 100644 --- a/sdk/eggai/transport/redis.py +++ b/sdk/eggai/transport/redis.py @@ -77,6 +77,7 @@ def __init__( group_monitor_interval_s: float = 5.0, max_len: int | None = None, retry_max_len: int | None = 10_000, + dlq_max_len: int | None = None, **kwargs, ): """ @@ -97,7 +98,17 @@ def __init__( retry_max_len (Optional[int]): Approximate cap on the SDK-managed retry and DLQ streams (default 10_000). These hold only reclaimed failures, so their volume is bounded by your error rate and a default cap prevents a runaway retry loop from growing without bound. - Set to ``None`` to disable trimming on retry/DLQ streams. + Set to ``None`` to disable trimming on retry/DLQ streams. Does NOT apply to a shared DLQ + (``dlq_channel``); see ``dlq_max_len``. + dlq_max_len (Optional[int]): Approximate cap on a *shared* DLQ stream (one named via the + ``dlq_channel`` subscribe option), applied as ``XADD ... MAXLEN ~ dlq_max_len`` by this + transport's reclaimers. Default ``None`` (untrimmed), deliberately: ``MAXLEN`` trims the + whole stream regardless of which writer appended or whether the entry was consumed, so on a + stream several services write to, one service's cap silently deletes the others' unread + dead letters. Leave it ``None`` and let the consuming sink own retention (``XTRIM`` after + processing), or set it only if every writer to that DLQ uses the same value and you prefer + a hard memory ceiling over the risk of losing dead letters. Per-handler DLQs are unaffected + (they keep ``retry_max_len``). **kwargs: Additional keyword arguments to pass to the RedisBroker if a new instance is created. Attributes: @@ -164,6 +175,7 @@ def __init__( } self._max_len = max_len self._retry_max_len = retry_max_len + self._dlq_max_len = dlq_max_len self._running = False self._reclaimer_manager: PendingReclaimerManager | None = None # A set so repeated subscribe() calls with an identical (stream_key, group, @@ -330,8 +342,26 @@ async def subscribe(self, channel: str, handler, **kwargs) -> Callable: called up to 6 times total (1 original + 5 retries). Set to None for unlimited retries (no DLQ). Requires retry_on_idle_ms. on_dlq (Callable, optional): Callback invoked when a message is routed to the DLQ. Can be sync or async. - Signature: ``on_dlq(fields: dict, msg_id: str, retry_count: int)``. Errors in the callback are - logged but never prevent the DLQ write. Only used when retry_on_idle_ms and max_retries are set. + Signature: ``on_dlq(body: dict, msg_id: str, retry_count: int)``. ``body`` is the DLQ entry's + decoded body — exactly what a subscriber on the DLQ stream receives, including the ``_dlq_*`` + provenance keys (for a poison entry: the provenance plus ``_dlq_raw_b64``, never raw bytes). + ``retry_count`` is the count that exceeded ``max_retries`` (0 for poison). Errors in the + callback are logged but never prevent the DLQ write. Only used when retry_on_idle_ms and + max_retries are set. + dlq_channel (str, optional): Full stream key to dead-letter into instead of the per-handler + ``{channel}.{handler_suffix}.dlq``. Lets many handlers (or many services) share ONE DLQ stream + that a single consumer can subscribe to. At this level it is a raw key; ``Agent.subscribe`` / + ``Channel.subscribe`` accept a ``Channel`` or a topic name and namespace it for you. Requires + retry_on_idle_ms and a non-None max_retries; must differ from the subscribed channel (a + handler dead-lettering into its own input would loop) and must not end in ``.retry`` (SDK-managed + retry streams). The retry stream stays per-handler — only the terminal sink is shared. Every + DLQ entry carries ``_dlq_source`` (origin channel key), ``_dlq_handler``, ``_dlq_at``, + ``_dlq_reason`` and ``_dlq_retries`` in its body so a shared consumer can tell entries apart, + and ``_retry_count`` is reset to ``"0"`` so a DLQ consumer with its own retries starts with a + fresh budget; subscribe to the DLQ with ``group_start="0"`` to pick up an existing backlog. + A shared DLQ is written WITHOUT ``MAXLEN`` unless ``RedisTransport(dlq_max_len=...)`` is set + (``retry_max_len`` does not apply): trimming is stream-wide, so any writer's cap would delete + other writers' unconsumed dead letters. By default retention of a shared DLQ is the sink's job. retry_on_error (bool, optional): Whether to retry handler on error (default is True). # Durability parameters @@ -356,6 +386,7 @@ async def subscribe(self, channel: str, handler, **kwargs) -> Callable: _explicit_max_retries = "max_retries" in kwargs max_retries = kwargs.pop("max_retries", 5) on_dlq = kwargs.pop("on_dlq", None) + dlq_channel = kwargs.pop("dlq_channel", None) _explicit_backoff = ( "retry_backoff_multiplier" in kwargs or "retry_backoff_max_ms" in kwargs @@ -399,6 +430,21 @@ async def subscribe(self, channel: str, handler, **kwargs) -> Callable: handler_id = kwargs.pop("handler_id", None) + # Per-handler retry/DLQ keys (issue #225): derived once, up front, so the + # dlq_channel validation below can run BEFORE the broker subscriber is + # registered (that registration cannot be undone). `Agent`/`Channel` + # always inject a unique handler_id; the fallback only applies to callers + # that use transport.subscribe() directly, who should pass a distinct + # handler_id per handler on one channel. getattr guards objects without a + # __name__ (e.g. functools.partial). Derived from the original handler, + # not the tracing wrapper. + handler_name = ( + getattr(original_handler, "__name__", None) + or type(original_handler).__name__ + ) + handler_suffix = handler_id if handler_id else f"{channel}-{handler_name}" + retry_stream = f"{channel}.{handler_suffix}.retry" + # Ignore Kafka-specific parameter (Redis uses 'group' for streams, not 'group_id') kwargs.pop("group_id", None) @@ -438,6 +484,15 @@ async def subscribe(self, channel: str, handler, **kwargs) -> Callable: "Use retry_on_idle_ms for SDK-managed retry streams, or " "min_idle_time for FastStream's built-in XAUTOCLAIM." ) + # The reclaimer works on a consumer group's PEL (XPENDING / XCLAIM); with + # no group there is no PEL and every reclaim cycle would just error. + # Agent/Channel always supply a handler_id (hence a group); this only + # bites direct transport.subscribe() callers. + if retry_on_idle_ms is not None and not group: + raise ValueError( + "retry_on_idle_ms requires a consumer group: pass handler_id= or " + "group= (Agent.subscribe / Channel.subscribe set one automatically)." + ) if ( _explicit_max_retries @@ -496,6 +551,40 @@ async def subscribe(self, channel: str, handler, **kwargs) -> Callable: if retry_on_idle_ms is None: max_retries = None + # A shared DLQ only makes sense when there is a DLQ at all, and it must + # never be the stream this handler consumes (dead-lettering into your own + # input is a feedback loop, not a retry). + if dlq_channel is not None: + if not isinstance(dlq_channel, str) or not dlq_channel: + raise ValueError( + "dlq_channel must be a non-empty stream key (Agent.subscribe / " + "Channel.subscribe also accept a Channel instance)." + ) + if retry_on_idle_ms is None: + raise ValueError( + "dlq_channel requires retry_on_idle_ms to be set. " + "Set retry_on_idle_ms to enable SDK-managed retries with a DLQ." + ) + if max_retries is None: + raise ValueError( + "dlq_channel requires max_retries: max_retries=None disables the " + "DLQ entirely, so there is nothing to route to the shared channel." + ) + if dlq_channel == channel: + raise ValueError( + f"dlq_channel {dlq_channel!r} is the channel this handler " + "subscribes to; dead-lettering into its own input would loop." + ) + # Not just this handler's retry stream: another handler's `.retry` + # is auto-consumed by *that* handler, so dead-lettering into it + # would re-run the wrong handler on the message (the #225 class). + if dlq_channel.endswith(".retry"): + raise ValueError( + f"dlq_channel {dlq_channel!r} names an SDK-managed retry " + "stream ('.retry' suffix); dead-lettering into a retry stream " + "would feed a handler's retry loop. Pick a different key." + ) + stream_sub = StreamSub( channel, group=group, @@ -531,29 +620,28 @@ async def subscribe(self, channel: str, handler, **kwargs) -> Callable: )(handler) if retry_on_idle_ms is not None and not _internal_retry: - # Per-handler retry/dlq stream keys: when multiple handlers - # subscribe to the same channel with different consumer groups, - # each gets its own retry/dlq streams. A single shared retry - # stream would broadcast one handler's failures to every other - # handler on the channel via the auto-created `-retry` consumer - # groups. See issue #225. - # `Agent`/`Channel` always inject a unique handler_id (via the - # HANDLERS_IDS counter), so the fallback only applies to callers that - # use transport.subscribe() directly. getattr guards objects without a - # __name__ (e.g. functools.partial), which would otherwise raise after - # the main-stream subscriber is already registered. Direct callers that - # run multiple distinct handlers on one channel with retry_on_idle_ms - # should pass a distinct handler_id per handler — otherwise same-named - # handlers (or lambdas) collapse to the same per-handler key. Derive - # the name from the original handler, not the tracing wrapper. - handler_name = ( - getattr(original_handler, "__name__", None) - or type(original_handler).__name__ - ) - handler_suffix = handler_id if handler_id else f"{channel}-{handler_name}" - retry_stream = f"{channel}.{handler_suffix}.retry" - dlq_stream = ( - f"{channel}.{handler_suffix}.dlq" if max_retries is not None else None + # Per-handler retry/dlq stream keys (handler_suffix / retry_stream + # were derived above, next to handler_id): when multiple handlers + # subscribe to the same channel with different consumer groups, each + # gets its own retry/dlq streams. A single shared retry stream would + # broadcast one handler's failures to every other handler on the + # channel via the auto-created `-retry` consumer groups (#225). + # The DLQ is the one stream that may be shared across handlers: an + # explicit dlq_channel replaces the per-handler default. The retry + # stream is never shared — that is the #225 fan-out bug. + dlq_stream: str | None + if max_retries is None: + dlq_stream = None + elif dlq_channel is not None: + dlq_stream = dlq_channel + else: + dlq_stream = f"{channel}.{handler_suffix}.dlq" + # A shared DLQ is not trimmed by its writers unless the operator opts + # in with dlq_max_len: XADD MAXLEN applies to the whole stream, so one + # service's retry_max_len would silently discard other services' + # unconsumed dead letters. Per-handler DLQs keep retry_max_len. + dlq_max_len = ( + self._dlq_max_len if dlq_channel is not None else self._retry_max_len ) retry_handler_id = f"{handler_suffix}-retry" @@ -587,6 +675,9 @@ async def subscribe(self, channel: str, handler, **kwargs) -> Callable: backoff_multiplier=retry_backoff_multiplier, backoff_max_ms=retry_backoff_max_ms, backoff_jitter=retry_backoff_jitter, + source_stream=channel, + handler=handler_suffix, + dlq_max_len=dlq_max_len, ) ) @@ -649,6 +740,11 @@ async def subscribe(self, channel: str, handler, **kwargs) -> Callable: backoff_multiplier=retry_backoff_multiplier, backoff_max_ms=retry_backoff_max_ms, backoff_jitter=retry_backoff_jitter, + # Provenance is the *original* channel, not the retry + # stream this reclaimer scans. + source_stream=channel, + handler=handler_suffix, + dlq_max_len=dlq_max_len, ) ) except Exception: @@ -759,6 +855,9 @@ def _setup_reclaimer( backoff_multiplier: float = 1.0, backoff_max_ms: int | None = None, backoff_jitter: float = 0.0, + source_stream: str | None = None, + handler: str | None = None, + dlq_max_len: int | None = None, ) -> tuple[str, str, str]: if self._reclaimer_manager is None: self._reclaimer_manager = PendingReclaimerManager( @@ -779,5 +878,8 @@ def _setup_reclaimer( backoff_multiplier=backoff_multiplier, backoff_max_ms=backoff_max_ms, backoff_jitter=backoff_jitter, + source_stream=source_stream, + handler=handler, + dlq_max_len=dlq_max_len, ) ) diff --git a/sdk/tests/test_redis.py b/sdk/tests/test_redis.py index 90f0deb2..ad939ed0 100644 --- a/sdk/tests/test_redis.py +++ b/sdk/tests/test_redis.py @@ -845,12 +845,10 @@ async def test_inject_retry_metadata_malformed_payload(caplog): garbage = b"\x00\x01\x02not-valid-binary-format" with caplog.at_level(logging.WARNING, logger="eggai.transport.pending_reclaimer"): - result_data, result_count, parsed_ok = _inject_retry_metadata(garbage, "0-1") - assert result_data == garbage - assert result_count == 0 - # parsed_ok=False signals the caller to treat this as a poison message - # (route to DLQ / drop) instead of re-queueing it forever. - assert parsed_ok is False + result = _inject_retry_metadata(garbage, "0-1") + # None signals the caller to treat this as a poison message (route to DLQ / + # drop) instead of re-queueing it forever. + assert result is None assert any("Failed to inject retry metadata" in r.message for r in caplog.records) @@ -860,7 +858,10 @@ def test_inject_retry_metadata_non_ascii_header(): from faststream.redis.parser.binary import BinaryMessageFormatV1 - from eggai.transport.pending_reclaimer import _inject_retry_metadata + from eggai.transport.pending_reclaimer import ( + _encode_body, + _inject_retry_metadata, + ) headers = {"correlation_id": "c1", "x-tenant": "münchen", "x-after": "abc"} headers_bytes = b"" @@ -878,12 +879,14 @@ def test_inject_retry_metadata_non_ascii_header(): + b'{"type":"t","data":{}}' ) - new_data, count, parsed_ok = _inject_retry_metadata(envelope, "1-0") + parsed = _inject_retry_metadata(envelope, "1-0") - assert parsed_ok is True + assert parsed is not None + parsed_headers, body_dict, count = parsed assert count == 1 - body, parsed_headers = BinaryMessageFormatV1.parse(new_data) - assert parsed_headers == headers + new_data = _encode_body(parsed_headers, body_dict) + body, reparsed_headers = BinaryMessageFormatV1.parse(new_data) + assert reparsed_headers == headers assert json.loads(body)["_retry_count"] == "1" @@ -992,9 +995,11 @@ async def always_fails(message): body_bytes, _headers = BinaryMessageFormatV1.parse(fields[b"__data__"]) body = json.loads(body_bytes) - assert int(body["_retry_count"]) > 2, ( - f"Expected _retry_count > 2 (exceeded max_retries=2), got {body['_retry_count']}" - ) + # The exhausted budget is recorded as _dlq_retries (== max_retries) and + # _retry_count is reset so a DLQ consumer with its own retries starts fresh. + assert body["_dlq_retries"] == "2", body + assert body["_retry_count"] == "0", body + assert body["_dlq_reason"] == "max_retries" assert "_original_message_id" in body await redis_client.aclose() diff --git a/sdk/tests/test_redis_shared_dlq.py b/sdk/tests/test_redis_shared_dlq.py new file mode 100644 index 00000000..d8724f0f --- /dev/null +++ b/sdk/tests/test_redis_shared_dlq.py @@ -0,0 +1,609 @@ +""" +Shared DLQ channel (``dlq_channel``) tests. + +Covers three layers: + +- pure helpers in ``pending_reclaimer`` (envelope re-encoding, provenance + injection, poison wrapping) — no Redis needed; +- subscribe-time wiring and validation in ``RedisTransport`` / ``Agent`` / + ``Channel`` — builds configs without connecting; +- end-to-end against a live Redis: two failing handlers on two channels + dead-letter into ONE stream, a sink that starts *after* the failures consumes + it with ``group_start="0"`` and reads the provenance keys, and a poison entry + is wrapped so the sink still receives a JSON object. + +The file name contains "redis" so ``conftest.py`` skips the whole module when +no Redis is listening on localhost:6379. +""" + +import asyncio +import base64 +import json +import uuid + +import pytest +import redis.asyncio as redis +from faststream.redis.parser.binary import BinaryMessageFormatV1 + +from eggai import Agent, Channel +from eggai.channel import NAMESPACE, resolve_dlq_channel +from eggai.transport import RedisTransport +from eggai.transport.base import Transport +from eggai.transport.pending_reclaimer import ( + PendingReclaimerManager, + ReclaimerConfig, + _dlq_metadata, + _encode_envelope, + _parse_entry, + _poison_body, + _stamp_dead_letter, +) + +# --------------------------------------------------------------------------- # +# Pure helpers (no Redis) +# --------------------------------------------------------------------------- # + + +def test_resolve_dlq_channel_namespaces_topic_names(): + assert resolve_dlq_channel(None) is None + assert resolve_dlq_channel("dlq") == f"{NAMESPACE}.dlq" + assert resolve_dlq_channel(Channel("dlq")) == f"{NAMESPACE}.dlq" + with pytest.raises(ValueError, match="non-empty"): + resolve_dlq_channel("") + with pytest.raises(ValueError, match="non-empty"): + resolve_dlq_channel(42) # type: ignore[arg-type] + # A full key pasted as a string would be namespaced twice → reject loudly. + with pytest.raises(ValueError, match="already starts with the namespace"): + resolve_dlq_channel(f"{NAMESPACE}.dlq") + with pytest.raises(ValueError, match="already starts with the namespace"): + resolve_dlq_channel(Channel("dlq").get_name()) + + +def test_stamp_dead_letter_resets_budget_and_keeps_first_origin(): + body = { + "type": "order", + "data": {"k": 1}, + "_retry_count": "4", + "_original_message_id": "1-0", + } + origin = { + "_dlq_reason": "max_retries", + "_dlq_source": "ns.orders", + "_dlq_handler": "svc-h-1", + "_dlq_at": "t1", + } + _stamp_dead_letter(body, origin, 3) + assert body["data"] == {"k": 1} # payload untouched + assert body["_dlq_source"] == "ns.orders" + assert body["_dlq_retries"] == "3" + assert body["_retry_count"] == "0" # fresh budget for the DLQ consumer + assert body["_original_message_id"] == "1-0" + + # Dead-lettered a second time (the DLQ consumer itself gave up after 2 + # retries, here as poison): origin keys keep the first failure, hop keys + # describe the latest one. + sink = { + "_dlq_reason": "poison", + "_dlq_source": "ns.dlq", + "_dlq_handler": "sink-1", + "_dlq_at": "t2", + } + body["_retry_count"] = "3" + _stamp_dead_letter(body, sink, 2) + assert body["_dlq_source"] == "ns.orders" + assert body["_dlq_handler"] == "svc-h-1" + assert body["_dlq_at"] == "t1" + assert body["_dlq_reason"] == "poison" + assert body["_dlq_retries"] == "2" + assert body["_retry_count"] == "0" + + +def test_parse_entry_rejects_what_cannot_be_retried(): + assert _parse_entry(b"garbage") is None + assert _parse_entry(_encode_envelope({}, b"[1,2]")) is None # JSON array body + assert _parse_entry(_encode_envelope({}, b'{"_retry_count":"n/a"}')) is None + parsed = _parse_entry( + _encode_envelope({"h": "v"}, b'{"type":"t","_retry_count":"2"}') + ) + assert parsed is not None + headers, body, count = parsed + assert headers == {"h": "v"} + assert body["type"] == "t" + assert count == 2 + + +def test_poison_body_keeps_raw_bytes_headers_and_zero_budget(): + raw = b"\x00\x01definitely-not-an-envelope" + headers, body = _poison_body( + raw, {"_dlq_reason": "poison", "_dlq_source": "ns.o"}, "7-0" + ) + assert headers == {} + assert body["_dlq_reason"] == "poison" + assert body["_dlq_source"] == "ns.o" + assert body["_original_message_id"] == "7-0" + assert body["_retry_count"] == "0" + assert body["_dlq_retries"] == "0" + assert base64.b64decode(body["_dlq_raw_b64"]) == raw + + # An envelope whose body is unusable keeps its headers (correlation id…). + env = _encode_envelope({"correlation_id": "c1"}, b"[1,2]") + headers, body = _poison_body(env, {"_dlq_reason": "poison"}, "8-0") + assert headers == {"correlation_id": "c1"} + assert base64.b64decode(body["_dlq_raw_b64"]) == env + + +def test_dlq_metadata_records_original_channel_not_scanned_stream(): + # The retry reclaimer scans the retry stream; provenance must still name the + # channel the handler subscribed to. + cfg = ReclaimerConfig( + stream="ns.orders.svc-h-1.retry", + group="svc-h-1-retry", + consumer="svc-h-1-retry-reclaimer", + retry_stream="ns.orders.svc-h-1.retry", + min_idle_ms=1, + interval_s=1.0, + source_stream="ns.orders", + handler="svc-h-1", + ) + meta = _dlq_metadata(cfg, "max_retries") + assert meta["_dlq_source"] == "ns.orders" + assert meta["_dlq_handler"] == "svc-h-1" + assert meta["_dlq_reason"] == "max_retries" + assert meta["_dlq_at"].endswith("+00:00") + + +# --------------------------------------------------------------------------- # +# Wiring + validation (no connection) +# --------------------------------------------------------------------------- # + + +@pytest.mark.asyncio +async def test_dlq_channel_shared_by_both_reclaimers_and_across_handlers(): + transport = RedisTransport() + + async def handler_a(message): + return message + + async def handler_b(message): + return message + + await transport.subscribe( + "orders", + handler_a, + handler_id="orders-a-1", + retry_on_idle_ms=500, + max_retries=3, + dlq_channel="ops.dlq", + ) + await transport.subscribe( + "payments", + handler_b, + handler_id="payments-b-1", + retry_on_idle_ms=500, + max_retries=3, + dlq_channel="ops.dlq", + ) + + assert transport._reclaimer_manager is not None + configs = list(transport._reclaimer_manager._configs.values()) + assert len(configs) == 4 # main + retry reclaimer per handler + assert {c.dlq_stream for c in configs} == {"ops.dlq"} + # Retry streams stay per-handler (#225) — only the terminal sink is shared. + assert {c.retry_stream for c in configs} == { + "orders.orders-a-1.retry", + "payments.payments-b-1.retry", + } + assert {(c.source_stream, c.handler) for c in configs} == { + ("orders", "orders-a-1"), + ("payments", "payments-b-1"), + } + + +@pytest.mark.asyncio +async def test_shared_dlq_is_never_trimmed_by_its_writers(): + """XADD MAXLEN trims the whole stream regardless of writer, so a shared DLQ + must not inherit any single writer's retry_max_len; the per-handler DLQ and + the retry streams keep the cap.""" + transport = RedisTransport(retry_max_len=250) + + async def handler(message): + return message + + await transport.subscribe( + "orders", + handler, + handler_id="orders-h-1", + retry_on_idle_ms=500, + dlq_channel="ops.dlq", + ) + await transport.subscribe( + "payments", handler, handler_id="payments-h-1", retry_on_idle_ms=500 + ) + assert transport._reclaimer_manager is not None + by_source = {} + for c in transport._reclaimer_manager._configs.values(): + by_source.setdefault(c.source_stream, set()).add((c.max_len, c.dlq_max_len)) + # Shared DLQ: retry stream still capped, DLQ untrimmed. + assert by_source["orders"] == {(250, None)} + # Per-handler default: both capped, as before. + assert by_source["payments"] == {(250, 250)} + + # Opt-in hard ceiling for the shared DLQ only; per-handler keeps retry_max_len. + capped = RedisTransport(retry_max_len=250, dlq_max_len=500) + await capped.subscribe( + "orders", + handler, + handler_id="orders-h-2", + retry_on_idle_ms=500, + dlq_channel="ops.dlq", + ) + await capped.subscribe( + "payments", handler, handler_id="payments-h-2", retry_on_idle_ms=500 + ) + assert capped._reclaimer_manager is not None + caps = {} + for c in capped._reclaimer_manager._configs.values(): + caps.setdefault(c.source_stream, set()).add((c.max_len, c.dlq_max_len)) + assert caps["orders"] == {(250, 500)} + assert caps["payments"] == {(250, 250)} + + +@pytest.mark.asyncio +async def test_dlq_default_unchanged_when_dlq_channel_not_passed(): + transport = RedisTransport() + + async def handler(message): + return message + + await transport.subscribe( + "orders", + handler, + handler_id="orders-handler-1", + retry_on_idle_ms=500, + max_retries=3, + ) + assert transport._reclaimer_manager is not None + configs = list(transport._reclaimer_manager._configs.values()) + assert {c.dlq_stream for c in configs} == {"orders.orders-handler-1.dlq"} + # Provenance is recorded on per-handler DLQs too. + assert all( + c.source_stream == "orders" and c.handler == "orders-handler-1" for c in configs + ) + + +@pytest.mark.asyncio +async def test_dlq_channel_validation_at_transport(): + transport = RedisTransport() + + async def handler(message): + return message + + with pytest.raises(ValueError, match="requires retry_on_idle_ms"): + await transport.subscribe( + "orders", handler, handler_id="h-1", dlq_channel="ops.dlq" + ) + with pytest.raises(ValueError, match="requires max_retries"): + await transport.subscribe( + "orders", + handler, + handler_id="h-2", + retry_on_idle_ms=500, + max_retries=None, + dlq_channel="ops.dlq", + ) + with pytest.raises(ValueError, match="own input"): + await transport.subscribe( + "orders", + handler, + handler_id="h-3", + retry_on_idle_ms=500, + dlq_channel="orders", + ) + with pytest.raises(ValueError, match="retry stream"): + await transport.subscribe( + "orders", + handler, + handler_id="h-4", + retry_on_idle_ms=500, + dlq_channel="orders.h-4.retry", + ) + # …and ANOTHER handler's retry stream, which that handler auto-consumes. + with pytest.raises(ValueError, match="retry stream"): + await transport.subscribe( + "orders", + handler, + handler_id="h-4b", + retry_on_idle_ms=500, + dlq_channel="payments.other-1.retry", + ) + # Rejected before anything was registered on the broker for that handler. + assert not any( + info.group in ("h-4", "h-4b") for info in transport._stream_subscriptions + ) + with pytest.raises(ValueError, match="non-empty"): + await transport.subscribe( + "orders", handler, handler_id="h-5", retry_on_idle_ms=500, dlq_channel="" + ) + # SDK-managed retries need a consumer group (the reclaimer works on its PEL). + with pytest.raises(ValueError, match="requires a consumer group"): + await transport.subscribe("orders", handler, retry_on_idle_ms=500) + + +class _RecordingTransport(Transport): + """Captures subscribe kwargs so Agent/Channel-level resolution can be asserted.""" + + def __init__(self) -> None: + self.calls: list[tuple[str, dict]] = [] + + async def connect(self): + pass + + async def disconnect(self): + pass + + async def publish(self, channel, message): + pass + + async def subscribe(self, channel, handler, **kwargs): + self.calls.append((channel, kwargs)) + return handler + + +@pytest.mark.asyncio +async def test_agent_subscribe_resolves_dlq_channel_to_namespaced_key(): + transport = _RecordingTransport() + agent = Agent("svc", transport=transport) + + @agent.subscribe( + channel=Channel("orders", transport=transport), + retry_on_idle_ms=500, + dlq_channel="dlq", # topic name → namespaced + ) + async def h_str(message): + return message + + @agent.subscribe( + channel=Channel("payments", transport=transport), + retry_on_idle_ms=500, + dlq_channel=Channel("dlq"), # Channel → its own name + ) + async def h_channel(message): + return message + + await agent.start() + try: + assert [k["dlq_channel"] for _, k in transport.calls] == [ + f"{NAMESPACE}.dlq", + f"{NAMESPACE}.dlq", + ] + finally: + await agent.stop() + + +@pytest.mark.asyncio +async def test_agent_subscribe_decorator_reuse_does_not_double_namespace(): + """One decorator object applied to two handlers must not re-namespace the + already-resolved key (the closure's kwargs are shared between applications).""" + transport = _RecordingTransport() + agent = Agent("svc", transport=transport) + decorate = agent.subscribe( + channel=Channel("orders", transport=transport), + retry_on_idle_ms=500, + dlq_channel="dlq", + ) + + async def h1(message): + return message + + async def h2(message): + return message + + decorate(h1) + decorate(h2) + await agent.start() + try: + assert [k["dlq_channel"] for _, k in transport.calls] == [ + f"{NAMESPACE}.dlq", + f"{NAMESPACE}.dlq", + ] + finally: + await agent.stop() + + +def test_agent_subscribe_rejects_bad_dlq_channel_at_decoration_time(): + agent = Agent("svc", transport=_RecordingTransport()) + orders = Channel("orders") + + with pytest.raises(ValueError, match="requires retry_on_idle_ms"): + + @agent.subscribe(channel=orders, dlq_channel="dlq") + async def h1(message): + return message + + with pytest.raises(ValueError, match="requires max_retries"): + + @agent.subscribe( + channel=orders, retry_on_idle_ms=500, max_retries=None, dlq_channel="dlq" + ) + async def h2(message): + return message + + with pytest.raises(ValueError, match="own input"): + + @agent.subscribe(channel=orders, retry_on_idle_ms=500, dlq_channel=orders) + async def h3(message): + return message + + with pytest.raises(ValueError, match="retry stream"): + + @agent.subscribe( + channel=orders, retry_on_idle_ms=500, dlq_channel="orders.svc-h3-1.retry" + ) + async def h4(message): + return message + + +@pytest.mark.asyncio +async def test_channel_subscribe_resolves_dlq_channel(): + transport = _RecordingTransport() + orders = Channel("orders", transport=transport) + + async def handler(message): + return message + + await orders.subscribe(handler, retry_on_idle_ms=500, dlq_channel="dlq") + try: + assert transport.calls[0][1]["dlq_channel"] == f"{NAMESPACE}.dlq" + finally: + await orders.stop() + + +# --------------------------------------------------------------------------- # +# End-to-end against Redis +# --------------------------------------------------------------------------- # + + +@pytest.mark.asyncio +async def test_shared_dlq_channel_end_to_end(): + """Two failing handlers on two channels → one DLQ stream → one sink.""" + redis_client = redis.Redis(host="localhost", port=6379, decode_responses=False) + test_id = uuid.uuid4().hex[:8] + + transport = RedisTransport() + producers = Agent(f"prod-{test_id}", transport=transport) + orders = Channel(f"orders-{test_id}", transport=transport) + payments = Channel(f"payments-{test_id}", transport=transport) + dlq_topic = f"dlq-{test_id}" + dlq = Channel(dlq_topic, transport=transport) + + retry_opts = { + "retry_on_idle_ms": 300, + "retry_reclaim_interval_s": 0.5, + "max_retries": 1, + } + + @producers.subscribe(channel=orders, dlq_channel=dlq, **retry_opts) + async def fail_orders(message): + raise RuntimeError("orders always fail") + + @producers.subscribe(channel=payments, dlq_channel=dlq_topic, **retry_opts) + async def fail_payments(message): + raise RuntimeError("payments always fail") + + await producers.start() + await orders.publish({"type": "order", "data": {"id": 1}}) + await payments.publish({"type": "payment", "data": {"id": 2}}) + + for _ in range(60): + await asyncio.sleep(0.5) + if await redis_client.xlen(dlq.get_name()) >= 2: + break + assert await redis_client.xlen(dlq.get_name()) == 2, ( + "both failures land in ONE stream" + ) + # The per-handler DLQ keys are never created when dlq_channel is set. + assert not await redis_client.exists( + f"{orders.get_name()}.prod-{test_id}-fail_orders-1.dlq" + ) + assert not await redis_client.exists( + f"{payments.get_name()}.prod-{test_id}-fail_payments-1.dlq" + ) + + # A sink that boots AFTER the failures happened: group_start="0" picks up + # the backlog, and the body says where each entry came from. + received: dict[str, dict] = {} + got_both = asyncio.Event() + sink_transport = RedisTransport() + sink = Agent(f"sink-{test_id}", transport=sink_transport) + + @sink.subscribe( + channel=Channel(dlq_topic, transport=sink_transport), group_start="0" + ) + async def on_dead_letter(message): + received[message["_dlq_source"]] = message + if len(received) == 2: + got_both.set() + + await sink.start() + try: + await asyncio.wait_for(got_both.wait(), timeout=15) + finally: + await sink.stop() + await producers.stop() + await redis_client.aclose() + + o = received[orders.get_name()] + assert o["type"] == "order" and o["data"] == {"id": 1} + assert o["_dlq_handler"] == f"prod-{test_id}-fail_orders-1" + assert o["_dlq_reason"] == "max_retries" + assert o["_dlq_retries"] == "1" # max_retries=1 → one retry actually ran + assert o["_retry_count"] == "0" # fresh budget for the sink + assert "_dlq_at" in o and "_original_message_id" in o + + p = received[payments.get_name()] + assert p["type"] == "payment" and p["data"] == {"id": 2} + assert p["_dlq_handler"] == f"prod-{test_id}-fail_payments-1" + + +@pytest.mark.asyncio +async def test_reclaimer_wraps_poison_entry_into_dlq(): + """An unparseable envelope is dead-lettered as a decodable JSON object.""" + redis_client = redis.Redis(host="localhost", port=6379, decode_responses=False) + test_id = uuid.uuid4().hex[:8] + stream = f"eggai.poison-{test_id}" + group = "svc-h-1" + dlq_stream = f"eggai.dlq-{test_id}" + + await redis_client.xgroup_create(stream, group, id="0", mkstream=True) + raw = b"\x00\x01definitely-not-an-envelope" + msg_id = await redis_client.xadd(stream, {b"__data__": raw}) + # A non-eggai producer's entry with no __data__ field at all. + no_data_id = await redis_client.xadd(stream, {b"foo": b"bar"}) + # Deliver to a consumer and never ack → sits in the PEL like a crashed handler. + await redis_client.xreadgroup(group, "worker", {stream: ">"}, count=2) + await asyncio.sleep(0.05) + + manager = PendingReclaimerManager("redis://localhost:6379") + config = ReclaimerConfig( + stream=stream, + group=group, + consumer=f"{group}-reclaimer", + retry_stream=f"{stream}.{group}.retry", + min_idle_ms=10, + interval_s=60.0, # never fires on its own during the test + max_retries=1, + dlq_stream=dlq_stream, + source_stream=stream, + handler=group, + ) + manager.add(config) + await manager.start() + try: + await manager._reclaim_once(config) + finally: + await manager.stop() + + entries = await redis_client.xrange(dlq_stream) + assert len(entries) == 2 + bodies = {} + for _id, fields in entries: + body_bytes, headers = BinaryMessageFormatV1.parse(fields[b"__data__"]) + assert headers == {} + body = json.loads(body_bytes) + bodies[body["_original_message_id"]] = body + body = bodies[msg_id.decode()] + assert body["_dlq_reason"] == "poison" + assert body["_dlq_source"] == stream + assert body["_dlq_handler"] == group + assert body["_retry_count"] == "0" + assert body["_dlq_retries"] == "0" + assert base64.b64decode(body["_dlq_raw_b64"]) == raw + # The __data__-less entry is wrapped too, instead of looping through retry. + other = bodies[no_data_id.decode()] + assert other["_dlq_reason"] == "poison" + assert json.loads(base64.b64decode(other["_dlq_raw_b64"])) == {"foo": "bar"} + + pending = await redis_client.xpending(stream, group) + assert pending["pending"] == 0, "original entry was acked" + assert not await redis_client.exists(config.retry_stream), "poison never retried" + await redis_client.aclose()