From 9e2ae2573903a8d66acfbac0ee1c56f9d64e382c Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Muhammet=20=C5=9Eafak?= Date: Sun, 25 Oct 2026 07:56:04 +0300 Subject: [PATCH 1/2] Route identity-less and undecodable bodies as poison on every transport (ADR-0036) Identity-less JSON objects and invalid UTF-8 inside JSON strings are now poison instead of unknown-URN on AMQP, Pulsar, Artemis and SQS; Redis dead-letters identity-less bodies unconditionally; Kafka without retry tiers degrades to commit + drop per section 6.5. Vendors conformance v1.7.0 and runs its identity_verdict and pulsar_exhaustion sections. --- .github/workflows/ci.yml | 2 +- .github/workflows/release.yml | 4 +- CHANGELOG.md | 68 ++ README.md | 28 +- src/babelqueue/_envelope_shape.py | 2 +- src/babelqueue/app.py | 135 ++- src/babelqueue/artemis_transport.py | 3 +- src/babelqueue/codec.py | 47 +- src/babelqueue/kafka_transport.py | 10 +- src/babelqueue/pika_transport.py | 4 +- src/babelqueue/pulsar_transport.py | 37 +- src/babelqueue/redis_transport.py | 36 +- src/babelqueue/sqs_transport.py | 11 + src/babelqueue/transport.py | 14 + tests/conformance/CONFORMANCE_VERSION | 2 +- tests/conformance/manifest.json | 819 +++++++++++++++++- tests/conformance/schema/results.example.json | 2 +- tests/test_app.py | 27 +- tests/test_behaviour_conformance.py | 179 +++- tests/test_conformance.py | 2 + tests/test_kafka_retry.py | 56 ++ tests/test_poison_routing.py | 511 +++++++++++ tests/test_pulsar_transport.py | 4 + tests/test_redis_transport.py | 97 ++- 24 files changed, 2040 insertions(+), 60 deletions(-) create mode 100644 tests/test_poison_routing.py diff --git a/.github/workflows/ci.yml b/.github/workflows/ci.yml index b31b635..f140d54 100644 --- a/.github/workflows/ci.yml +++ b/.github/workflows/ci.yml @@ -177,7 +177,7 @@ jobs: --health-retries 10 env: # The pinned conformance release (ADR-0032). Bump together with tests/conformance/. - BQ_CONFORMANCE_VERSION: 1.6.0 + BQ_CONFORMANCE_VERSION: 1.7.0 # sha256 of that release's fetch.sh: the script runs on the runner, so it is pinned rather # than trusted from a mutable tag. Bump together with BQ_CONFORMANCE_VERSION. BQ_FETCH_SH_SHA256: 8bb6ba567b288fb0ac71a70fd17041b2503df119e01998d76fdad372570bf849 diff --git a/.github/workflows/release.yml b/.github/workflows/release.yml index 395c32b..0e46e36 100644 --- a/.github/workflows/release.yml +++ b/.github/workflows/release.yml @@ -14,7 +14,7 @@ jobs: # Recommended for PyPI Trusted Publishing — configure a "pypi" environment + # a trusted publisher on PyPI (repo BabelQueue/babelqueue-python, this workflow). environment: pypi - # redis_layout (conformance 1.6.0) drives the real Lua scripts: a real redis:7 is mandatory, + # redis_layout (conformance 1.7.0) drives the real Lua scripts: a real redis:7 is mandatory, # exactly as in ci.yml's `conformance` job — without it the cases are reported as failures. services: redis: @@ -28,7 +28,7 @@ jobs: --health-retries 10 env: # The pinned conformance release (ADR-0032). Keep identical to ci.yml. - BQ_CONFORMANCE_VERSION: 1.6.0 + BQ_CONFORMANCE_VERSION: 1.7.0 BQ_FETCH_SH_SHA256: 8bb6ba567b288fb0ac71a70fd17041b2503df119e01998d76fdad372570bf849 permissions: id-token: write # PyPI Trusted Publishing (OIDC) — no API token needed diff --git a/CHANGELOG.md b/CHANGELOG.md index 327e065..f02d5a3 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -9,6 +9,74 @@ The envelope wire format is versioned separately by `meta.schema_version` ## [Unreleased] +Implements ADR-0036 (post-R1 parity) decisions 2, 4, 5 and 6. A MINOR: every change is a +consumer-side settle decision, so the envelope stays at `schema_version: 1`. The vendored +conformance suite moves to v1.7.0 (new sections `identity_verdict` and `pulsar_exhaustion`, both +passing; `conformance-known-gaps.json` stays empty). + +### Changed (behaviour) +- **A body that does not decode, or has no identity, is poison on every transport (ADR-0036 + decisions 4 and 6).** A JSON object that has neither a valid string `job` nor a valid string + `urn` (ADR-0032 section 6) used to be an *unknown URN*, and on SQS, AMQP, Pulsar, Azure Service + Bus, Artemis and the memory transport a body that did not decode was decoded leniently into an + empty envelope and retried as an unknown URN, finally dead-lettered as `reason: "unknown_urn"` + without the original body. Both are now **poison** and take the poison route of Go's + `routePoison`: they are never handed to a handler and never take the unknown-URN strategies + (those remain for a valid envelope whose URN has no handler). With the DLQ on (`dead_letter=True` + or `on_unknown_urn=DEAD_LETTER`) the message is written to the DLQ at once, with no retry, as the + ADR-0035 record (`dead_letter.reason: "poison"`, the original body preserved: a strict JSON + object spliced byte for byte, anything else as `raw`, base64 plus `raw_encoding` when it is not + valid UTF-8) and acknowledged after the write; `on_unknown_urn=DELETE` drops it (logged); with + the DLQ off `RELEASE` is replaced by the bounded `FAIL` path (re-published with `attempts + 1` + up to `max_attempts`, then dropped and logged). An operator who relied on a poison body being + retried through the unknown-URN strategy, or on a `reason: "unknown_urn"` record for it, sees the + new behaviour. The classification is `EnvelopeCodec.poison_cause(raw)` (new, pure): the typed + decode of the Go reference plus the identity rule; `decode` / `decode_lenient` are unchanged. +- **Pulsar poison records carry `reason: "poison"`** (they were `unknown_urn`). The Pulsar + transport also reports the reconciled `bq-attempts` / redelivery count to the runtime + (`ReceivedMessage.native_attempts`, new), as does SQS (`ApproximateReceiveCount - 1`), so a poison + body, which cannot carry `attempts`, is still bounded by `max_attempts`. +- **Redis: an identity-less body is dead-lettered unconditionally (ADR-0036 decision 4, + broker-bindings section 1.1 "Identity-less body").** In both layouts (legacy `:processing` + and Laravel-compatible zset) a body that decodes but names no identity (including a bare `null`, + which used to follow the bounded `FAIL` path) is handed over as transport poison: it is written + to `.dlq` as the ADR-0035 record with `reason: "poison"` in every configuration + (`on_unknown_urn=DELETE` and dead-lettering off change nothing), never handed to a handler and + never retried. Nothing else in the Redis transport changes. +- **Kafka without retry tiers: a poison record no longer blocks the partition (ADR-0036 decision + 5, broker-bindings section 6.5 degrade-to-delete).** With no `retry_tiers_ms`, dead-lettering off + and `on_unknown_urn` not `DELETE`, the bounded retry of a poison record (including an + identity-less body) needs a tier and raised, leaving the record uncommitted with `bq-attempts` + never advancing. It is now committed, dropped and logged, and the partition advances. A failing + handler and `RELEASE` of a valid envelope keep the section 6.5 rule (they still raise + `BabelQueueError` without tiers). The start-up warning for a tierless transport now says so; enable + `dead_letter` to keep such records in `.dlq`. +- A message dropped after its retries are exhausted with no DLQ configured is now logged + (`babelqueue.app`, WARNING) instead of being dropped silently. + +- **Pulsar: a failed DLQ write or retry publish is negatively acknowledged (Go `Rewind`).** + `PulsarTransport` is now a `Rewinder` (`negative_acknowledge`), so the unsettled original is + redelivered by the broker instead of staying un-acknowledged for the life of the consumer. A + graceful-shutdown release of an in-flight Pulsar message is now a nack as well rather than a + re-publish (on a `Shared` subscription the native redelivery count advances; on `Exclusive` / + `Failover` it does not, which is harmless: `bq-attempts` carries the attempt limit). + +### Fixed +- **Bytes that are not valid UTF-8 no longer crash the AMQP, Pulsar and Artemis consumers.** They + were decoded strictly in `pop`; they are now kept losslessly (`surrogateescape`). Invalid UTF-8 + is poison wherever it sits, **also inside a JSON string** (`EnvelopeCodec.poison_cause`, + error-handling section 7): it is never handed to a handler and is dead-lettered as the ADR-0035 + base64 record, like on Redis and Kafka. + +### Added +- `EnvelopeCodec.poison_cause(raw, *, require_identity=True)`; `Transport.classifies_poison` + (Kafka and Redis classify poison themselves and opt out of the runtime's classification); + `ReceivedMessage.native_attempts`. +- Conformance runners for `identity_verdict` (the core verdict on all 97 bodies, plus the fate on + the memory transport) and `pulsar_exhaustion` (all 29 rows, driven through the real + `PulsarTransport` and runtime over a duck-typed client), and per-transport poison tests. +- `ci.yml` / `release.yml` pin conformance `1.7.0` (the `fetch.sh` SHA-256 is unchanged). + ## [1.22.0] - 2026-10-24 Implements ADR-0033 (Redis) and ADR-0034 (Kafka tiers, backoff, poison). The envelope stays at diff --git a/README.md b/README.md index 9a11a3e..4632522 100644 --- a/README.md +++ b/README.md @@ -259,9 +259,31 @@ the DLQ on (`dead_letter` or `on_unknown_urn=DEAD_LETTER`) it is dead-lettered o written to `.dlq` (which must exist) and its offset is committed once that write is confirmed; a failed write leaves it uncommitted and rewound. With the DLQ off nothing is written: `on_unknown_urn=DELETE` drops it, any other strategy retries it through the retry tiers up to -`max_attempts` and then drops it (without tiers the record is redelivered in place, as above). On -the Redis legacy (`:processing`) layout a body that Go's typed decode rejects (neither a JSON -object nor `null`, or wrong member types) is dead-lettered raw on its first +`max_attempts` and then drops it. **Without retry tiers** that bounded retry has no tier to use, so +(with the DLQ off and `on_unknown_urn` not `DELETE`) the poison record is committed, dropped and logged +instead of blocking the partition (section 6.5 degrade-to-delete, ADR-0036 decision 5); enable +`dead_letter` to keep it in `.dlq`. A failing handler without tiers still cannot be retried, as above. + +### Poison messages on every other transport + +A message body that does not decode into an envelope (not a JSON object, malformed, a wrongly typed +member, or bytes that are not valid UTF-8, even inside a JSON string) or that has neither a valid +string `job` nor a valid string `urn` is **poison** on every transport (ADR-0036 decisions 2, 4 and +6; conformance `identity_verdict`). It is never handed to a handler and never takes the +unknown-URN strategies, which are for a valid envelope whose URN has no handler: + +- DLQ on (`dead_letter` or `on_unknown_urn=DEAD_LETTER`): written to `.dlq` at once, with no + retry, as the ADR-0035 record (`dead_letter.reason: "poison"`, the original body preserved, + base64 plus `raw_encoding` when it is not UTF-8), then acknowledged. A failed write leaves it + unacknowledged. +- `on_unknown_urn=DELETE`: acknowledged and dropped (logged). +- DLQ off: the bounded `FAIL` path (`RELEASE` is replaced by it): retried up to `max_attempts`, then + dropped and logged. The count comes from the broker where it has one (SQS receive count, Pulsar + `bq-attempts` / redelivery count), otherwise from a bare `{"attempts": n}` body the retry leaves. + +On Redis (both layouts) an identity-less body is dead-lettered as poison in **every** configuration, +including `on_unknown_urn=DELETE` and dead-lettering off (broker-bindings section 1.1). On the legacy +(`:processing`) layout a body that Go's typed decode rejects is dead-lettered raw on its first delivery and removed from `:processing`. ### Sharing a Redis queue with Laravel diff --git a/src/babelqueue/_envelope_shape.py b/src/babelqueue/_envelope_shape.py index fb07fbc..d2de61c 100644 --- a/src/babelqueue/_envelope_shape.py +++ b/src/babelqueue/_envelope_shape.py @@ -15,7 +15,7 @@ decoder read them) and every occurrence of a duplicated key is checked; any other member (a differently-cased name or a non-string ``urn`` alias included) is only checked for syntax. A non-string ``job`` is **not** poison either: the identity resolver ignores it (ADR-0032 section 6), - so the body is readable and simply carries no identity (conformance ``kafka_poison``, v1.6.0). + so the body is readable and simply carries no identity (conformance ``kafka_poison``, v1.6.0, and ``identity_verdict``, v1.7.0). A well-typed but non-conformant envelope is not poison: Go's consume path never calls ``Accepts``. """ diff --git a/src/babelqueue/app.py b/src/babelqueue/app.py index a910e32..178a7f9 100644 --- a/src/babelqueue/app.py +++ b/src/babelqueue/app.py @@ -326,15 +326,37 @@ def dispatch(self, received: ReceivedMessage) -> None: # §3.7/§4.7/§5.7/§6.7/§7.7): the raw body goes to the DLQ byte-identical, so there is # no envelope to read here. gated = received.route_urn.startswith(_UNSUPPORTED_SCHEMA_ROUTE_PREFIX) - envelope = {} if gated else EnvelopeCodec.decode_lenient(received.body) + # A body that does not decode, or has no identity, is poison (ADR-0036 decisions 4 and + # 6): never decoded leniently into an empty envelope, never handed to a handler and + # never routed as an unknown URN. A transport that already hands its poison over under + # a poison route (Kafka, Redis) is not classified twice. A delivery that carries an + # out-of-band routing URN is exempt from the identity half of the rule only. + poison_cause = ( + None + if gated or getattr(self.transport, "classifies_poison", False) + else EnvelopeCodec.poison_cause( + received.body, require_identity=not received.route_urn + ) + ) + envelope = ( + {} + if gated or poison_cause is not None + else EnvelopeCodec.decode_lenient(received.body) + ) urn = str(envelope.get("job") or envelope.get("urn") or "") if received.route_urn and envelope: # Transport-supplied routing override; body untouched. Only for a body that # decodes: an undecodable one (decode() yields {}) stays on the poison path. urn = received.route_urn - handler = self._handlers.get(urn) if urn else None + handler = ( + self._handlers.get(urn) if urn and poison_cause is None else None + ) try: + if poison_cause is not None: + self._phase = _SETTLING + self._route_body_poison(received, poison_cause) + return if handler is None: self._phase = _SETTLING self._route_unknown(urn, received, envelope) @@ -344,15 +366,19 @@ def dispatch(self, received: ReceivedMessage) -> None: self._phase = _SETTLING except Exception as exc: # noqa: BLE001 - one bad message must not kill the loop self._phase = _SETTLING - if received.route_urn.startswith(_UNSUPPORTED_SCHEMA_ROUTE_PREFIX): - # A version-gated delivery never takes the generic retry path, even when its - # dead-letter write fails: a retry re-publish may rebuild the version header - # from the (valid) body and walk the message past the gate into a handler. - # It is left unacked; the broker (or a rewind, for cumulative-offset brokers - # such as Kafka) brings it back and the gate rejects it again. + if poison_cause is not None or received.route_urn.startswith( + _UNSUPPORTED_SCHEMA_ROUTE_PREFIX + ): + # A version-gated or poison delivery never takes the generic retry path, even + # when its dead-letter write fails: a retry re-publish may rebuild the version + # header from the (valid) body and walk the message past the gate into a + # handler, and a poison body has no envelope to retry. It is left unacked; the + # broker (or a rewind, for cumulative-offset brokers such as Kafka) brings it + # back and the same decision is taken again. logger.error( - "Dead-lettering a version-gated message (route %r) on queue %r failed; it was NOT " - "retried or acknowledged; it will come back and the dead-letter write will be retried.", + "Dead-lettering a version-gated or poison message (route %r) on queue %r " + "failed; it was NOT retried or acknowledged; it will come back and the " + "dead-letter write will be retried.", received.route_urn, received.queue, exc_info=exc, @@ -579,13 +605,16 @@ def _warn_if_retries_cannot_redeliver(self) -> None: through a tier. That only happens once a handler fails, so it is a warning (not a start-up error). """ - tiers = getattr(self.transport, "retry_tiers_ms", None) - if isinstance(tiers, list) and not tiers and self.max_attempts > 1 and isinstance(self.transport, Redeliverer): + if self._retry_tiers_unavailable() and self.max_attempts > 1: logger.warning( "The Kafka transport has no retry tiers but max_attempts=%d: a failing handler " "cannot be retried -- the record is left uncommitted and redelivered after a " "pause, so it keeps failing in place. Configure retry_tiers_ms (or set " - "max_attempts=1) to retry through the tiers.", + "max_attempts=1) to retry through the tiers. A poison record (a body that does " + "not decode, or has no identity) is different: with dead-lettering off and " + "on_unknown_urn not 'delete' it is committed and dropped with a logged warning " + "(broker-bindings section 6.5, degrade-to-delete), so it never blocks the " + "partition; enable dead_letter to keep it in .dlq instead.", self.max_attempts, ) @@ -657,6 +686,19 @@ def _route_poison(self, received: ReceivedMessage) -> None: if self.dead_letter_enabled or self.on_unknown_urn == UnknownUrnStrategy.DEAD_LETTER: self._settle_version_gated(received, always=True) return + if self._retry_tiers_unavailable(): + # broker-bindings §6.5 / ADR-0036 decision 5: no retry tier, no dead-lettering and not + # ``delete`` degrades to ``delete``. The bounded retry of a poison record needs a tier + # and would raise, leaving the record uncommitted with ``bq-attempts`` never advancing, + # so the partition would block. Commit, drop, and say so. + logger.warning( + "Dropping a poison record on queue %r (route %r): the Kafka transport has no retry " + "tiers and dead-lettering is off, so it can be neither retried nor dead-lettered " + "(broker-bindings section 6.5, degrade-to-delete). The offset was committed.", + received.queue, str(route), + ) + self._ack_settled(received) + return attempts = int(getattr(route, "attempts", 0) or 0) self._retry_or_dead_letter( received, @@ -664,8 +706,60 @@ def _route_poison(self, received: ReceivedMessage) -> None: UnknownUrnError("The record is poison: it carries no decodable envelope."), ) - def _settle_transport_poison(self, received: ReceivedMessage) -> None: - """Dead-letter a Redis poison reservation (broker-bindings §1.1 / §1.3, ADR-0033, ADR-0035). + def _route_body_poison(self, received: ReceivedMessage, cause: str) -> None: + """Settle a body that does not decode into an envelope, or has no identity, on SQS, AMQP, + Pulsar, Azure Service Bus, Artemis, the memory transport and any other transport that does + not classify poison itself (ADR-0036 decisions 2, 4 and 6; Go ``routePoison``). + + The body is never handed to a handler and never takes the unknown-URN strategies: + + * ``on_unknown_urn=DELETE`` drops it (acknowledged and logged); + * with the DLQ on (``dead_letter`` or the DEAD_LETTER strategy) it is written to the DLQ as + the ADR-0035 record with ``reason: "poison"`` (original body preserved) and acknowledged + after the write, with no retry; + * with the DLQ off it takes the bounded FAIL path (``RELEASE`` is replaced by it, a poison + body would otherwise be released forever): re-published with ``attempts + 1`` up to + ``max_attempts``, then dropped. The count is the larger of the transport's own delivery + count and the ``attempts`` of the bare body a previous retry left behind. + + A failed DLQ write propagates: :meth:`dispatch` leaves the delivery unacknowledged.""" + if self.on_unknown_urn == UnknownUrnStrategy.DELETE: + logger.warning( + "Dropping a poison message on queue %r (on_unknown_urn=delete): %s.", + received.queue, cause, + ) + self._ack_settled(received) + return + attempts = self._poison_attempts(received) + if self.dead_letter_enabled or self.on_unknown_urn == UnknownUrnStrategy.DEAD_LETTER: + self._settle_transport_poison(received, cause, attempts) + return + self._retry_or_dead_letter( + received, + {"attempts": attempts}, + UnknownUrnError("The message is poison: " + cause), + ) + + @staticmethod + def _poison_attempts(received: ReceivedMessage) -> int: + """Deliveries a poison body has already had: the larger of the transport's own count and + the integer ``attempts`` of the body (a bare ``{"attempts": N}`` left by a retry).""" + body_attempts = parse_envelope(received.body).get("attempts") + if isinstance(body_attempts, bool) or not isinstance(body_attempts, int): + body_attempts = 0 + return max(int(received.native_attempts or 0), body_attempts, 0) + + def _retry_tiers_unavailable(self) -> bool: + """Whether the transport is a Kafka transport with no retry tiers configured.""" + tiers = getattr(self.transport, "retry_tiers_ms", None) + return isinstance(tiers, list) and not tiers and isinstance(self.transport, Redeliverer) + + def _settle_transport_poison( + self, received: ReceivedMessage, cause: Optional[str] = None, attempts: int = 0 + ) -> None: + """Dead-letter a poison delivery (broker-bindings §1.1 / §1.3, ADR-0033, ADR-0035): a Redis + poison reservation (unconditional) or, via :meth:`_route_body_poison`, a poison body of any + other transport (``cause`` and ``attempts`` are then known). The DLQ write is an unconditional MUST, so it ignores ``dead_letter_enabled``. The record follows ADR-0035 like every other poison record: a strict JSON object body is spliced byte @@ -676,8 +770,8 @@ def _settle_transport_poison(self, received: ReceivedMessage) -> None: {}, "poison", received.queue, - 0, - error="Undecodable BabelQueue body", + attempts, + error="Undecodable BabelQueue body" + (": " + cause if cause else ""), exception=BabelQueueError.__name__, )["dead_letter"] self._write_dead_letter( @@ -730,7 +824,12 @@ def _retry_or_dead_letter( self._dead_letter(received, envelope, reason, exc) return - # Retries exhausted, no DLQ configured — drop it (ack so it leaves the queue). + # Retries exhausted, no DLQ configured — drop it (ack so it leaves the queue), loudly. + logger.warning( + "Dropping a message on queue %r after %d attempts: retries are exhausted and no " + "dead-letter queue is configured.", + received.queue, attempts, + ) self._ack_settled(received) def _dead_letter( diff --git a/src/babelqueue/artemis_transport.py b/src/babelqueue/artemis_transport.py index 2ca8d00..71bf0f2 100644 --- a/src/babelqueue/artemis_transport.py +++ b/src/babelqueue/artemis_transport.py @@ -233,7 +233,8 @@ def _unsupported_schema_version(properties: Any) -> Optional[str]: def _payload(message: Any) -> str: body = getattr(message, "body", None) if isinstance(body, bytes): - return body.decode("utf-8") + # Lossless: see ``PulsarTransport._payload`` (ADR-0035 poison record, no crash). + return body.decode("utf-8", "surrogateescape") return str(body) if body is not None else "" # -- Transport ---------------------------------------------------------- diff --git a/src/babelqueue/codec.py b/src/babelqueue/codec.py index 0c61496..4e53886 100644 --- a/src/babelqueue/codec.py +++ b/src/babelqueue/codec.py @@ -16,7 +16,7 @@ import uuid from typing import Any, Dict, List, Mapping, Optional, Tuple -from ._envelope_shape import CORE_META_KEYS +from ._envelope_shape import CORE_META_KEYS, envelope_violation from .exceptions import BabelQueueError, EnvelopeKeyError SCHEMA_VERSION = 1 @@ -89,6 +89,15 @@ def has_forbidden_keys(envelope: Mapping[str, Any]) -> bool: return _has_forbidden_keys(envelope) +def has_invalid_utf8(text: str) -> bool: + """Whether ``text`` carries a byte that was not valid UTF-8 when the transport read it. + + Transports read a body losslessly (``errors="surrogateescape"``), so every invalid byte + appears as a lone surrogate in ``U+DC80..U+DCFF``. A ``\\udcff`` *escape* in the JSON text is + plain ASCII and is not flagged.""" + return any("\udc80" <= ch <= "\udcff" for ch in text) + + def _is_valid_urn(value: Any) -> bool: """A usable identity: a string that is not blank under the ASCII set (ADR-0032 §5/§6). @@ -299,6 +308,42 @@ def decode_lenient(raw: str) -> Dict[str, Any]: parsed = _parse_object(raw) return {} if parsed is None else _fold_identity(parsed) + @staticmethod + def poison_cause(raw: str, *, require_identity: bool = True) -> Optional[str]: + """Why ``raw`` is a **poison** body, or ``None`` when it decodes into an envelope with an + identity (ADR-0036 decision 4, conformance ``identity_verdict``). + + A body is poison when it is not exactly one JSON object (malformed, BOM, trailing tokens, + ``NaN`` / ``Infinity``, a bare ``null``, an array, a scalar), when a known member has the + wrong type (the trigger set of :func:`~babelqueue._envelope_shape.envelope_violation`, the + Go reference's typed ``Decode``), when the core parser cannot read it, or, with + ``require_identity``, when it has neither a valid string ``job`` nor a valid string + ``urn`` (ADR-0032 section 6: a ``job`` that is not a string is ignored, a blank one falls + back to ``urn``, a ``urn`` that is not a non-blank string is ignored; the last value of a + repeated member wins). A poison body is never handed to a handler and never takes the + unknown-URN strategies, which are for a valid envelope whose URN has no handler. + + ``require_identity=False`` keeps only the structural checks, for a delivery that a + transport routes on an out-of-band identity property (the Pulsar ``bq-job``), which is + outside the identity rule. Pure: no warning is logged and nothing is mutated.""" + if has_invalid_utf8(raw): + # error-handling.md section 7: a body that is not valid UTF-8 is poison even when its + # invalid bytes sit inside a JSON string, where ``json.loads`` would accept the lone + # surrogate they were escaped to. It must never reach a handler or be re-encoded. + return "the body is not valid UTF-8" + try: + violation = envelope_violation(raw) + if violation is not None: + return violation + parsed = json.loads(raw) + except (ValueError, RecursionError, TypeError) as exc: + return "the core decoder cannot parse the body: " + str(exc) + if not isinstance(parsed, dict): + return "the JSON root is not an object" + if require_identity and not (_is_valid_urn(parsed.get("job")) or _is_valid_urn(parsed.get("urn"))): + return "the body carries no valid 'job' or 'urn' identity (ADR-0032 section 6)" + return None + @staticmethod def urn(envelope: Mapping[str, Any]) -> str: """The message URN: canonical ``job``, with ``urn`` accepted as an alias. diff --git a/src/babelqueue/kafka_transport.py b/src/babelqueue/kafka_transport.py index 1159509..349afb7 100644 --- a/src/babelqueue/kafka_transport.py +++ b/src/babelqueue/kafka_transport.py @@ -28,8 +28,10 @@ ``raw_encoding: "base64"`` when it is not valid UTF-8), and a ``dead_letter`` block with ``reason: "poison"`` is added (the key is preserved; the DLQ is the configured one or ``.dlq``, which must exist). With the DLQ off it is NOT written to a DLQ (Go ``routePoison``): ``on_unknown_urn=DELETE`` drops it, any other -strategy takes the bounded FAIL path (retried through the tiers up to ``max_attempts``, then dropped). -The record is handed over raw under a +strategy takes the bounded FAIL path (retried through the tiers up to ``max_attempts``, then dropped); with +no retry tiers configured as well, that path has no tier to retry through, so the record is committed, dropped +and logged (the §6.5 degrade-to-delete rule, ADR-0036 decision 5) instead of blocking the partition. The record is +handed over raw under a ``PoisonRoute`` and wrapped by ``publish_settle`` only when it is written to a queue other than the one it was read from, so a ``redrive`` restore into the DLQ it popped it from (any DLQ name) is byte for byte and never double-wraps. The version gate (§6.7) still @@ -161,6 +163,10 @@ def _brokers_from_url(url: str) -> str: class KafkaTransport(Transport): + #: Classifies poison itself (``_poll_one``, ADR-0034 section 10): the runtime does not classify + #: the body a second time. + classifies_poison = True + def __init__( self, url: str = "kafka://localhost:9092", diff --git a/src/babelqueue/pika_transport.py b/src/babelqueue/pika_transport.py index 096a00b..ddfb1e3 100644 --- a/src/babelqueue/pika_transport.py +++ b/src/babelqueue/pika_transport.py @@ -115,7 +115,9 @@ def pop(self, queue: str, timeout: float = 1.0) -> Optional[ReceivedMessage]: if timeout and timeout > 0: self._connection.sleep(timeout) return None - text = body.decode("utf-8") if isinstance(body, (bytes, bytearray)) else str(body) + # Lossless: bytes that are not UTF-8 stay recoverable (``surrogateescape``) so a poison + # body is dead-lettered as the ADR-0035 base64 record instead of crashing the consumer. + text = body.decode("utf-8", "surrogateescape") if isinstance(body, (bytes, bytearray)) else str(body) headers = _delivery_headers(props) return ReceivedMessage(body=text, queue=queue, handle=method.delivery_tag, headers=headers) diff --git a/src/babelqueue/pulsar_transport.py b/src/babelqueue/pulsar_transport.py index 189cec0..ceb5623 100644 --- a/src/babelqueue/pulsar_transport.py +++ b/src/babelqueue/pulsar_transport.py @@ -230,6 +230,17 @@ def _parse_attempts(raw: Any) -> Optional[int]: value = int(raw) return value if value <= _MAX_ATTEMPTS else None + @staticmethod + def _native_attempts(redelivery_count: Any, properties: Optional[Dict[str, str]] = None) -> int: + """``max(bq-attempts, RedeliveryCount)`` (§5.5; 0-based): the reconciled count of a delivery + that has no decodable body attempts. An invalid or absent ``bq-attempts`` counts as 0.""" + from_property = PulsarTransport._parse_attempts((properties or {}).get("bq-attempts")) or 0 + try: + native = int(redelivery_count) + except (TypeError, ValueError): + native = 0 + return max(from_property, native, 0) + @staticmethod def _reconcile( body: str, @@ -320,7 +331,9 @@ def _payload(message: Any) -> str: getter = getattr(message, "data", None) data = getter() if callable(getter) else getter if isinstance(data, bytes): - return data.decode("utf-8") + # Lossless: bytes that are not UTF-8 stay recoverable (``surrogateescape``) so a poison + # body is dead-lettered as the ADR-0035 base64 record instead of crashing the consumer. + return data.decode("utf-8", "surrogateescape") return str(data) if data is not None else "" # -- Transport ---------------------------------------------------------- @@ -370,13 +383,33 @@ def pop(self, queue: str, timeout: float = 1.0) -> Optional[ReceivedMessage]: properties, self._on_routing_mismatch, ) - return ReceivedMessage(body=body, queue=queue, handle=message, route_urn=route_urn) + return ReceivedMessage( + body=body, + queue=queue, + handle=message, + route_urn=route_urn, + # The §5.5 reconciled count without a body: what a poison body (no envelope to carry + # ``attempts``) is counted by (ADR-0036 decision 2). + native_attempts=self._native_attempts(self._redelivery_count(message), properties), + ) def ack(self, message: ReceivedMessage) -> None: if message.handle is None: return self._consumer(message.queue).acknowledge(message.handle) + def rewind(self, message: ReceivedMessage) -> None: + """:class:`~babelqueue.transport.Rewinder`: negatively acknowledge ``message`` so the broker + redelivers it (Go ``Rewind`` = ``Nack``; ADR-0036 decision 1: a failed DLQ write or retry + publish leaves the original un-acknowledged for redelivery instead of leaving it unsettled + for the life of the consumer). On a ``Shared`` subscription the native redelivery count + advances, which the §5.5 reconciliation reads back; on ``Exclusive`` / ``Failover`` it does + not, which is harmless: the attempt limit is carried by the ``bq-attempts`` property and + the body, never by the nack.""" + if message.handle is None: + return + self._consumer(message.queue).negative_acknowledge(message.handle) + def close(self) -> None: # pragma: no cover - resource cleanup for resource in (*self._producers.values(), *self._consumers.values()): try: diff --git a/src/babelqueue/redis_transport.py b/src/babelqueue/redis_transport.py index 5b3212f..54d38b4 100644 --- a/src/babelqueue/redis_transport.py +++ b/src/babelqueue/redis_transport.py @@ -67,8 +67,7 @@ from typing import Any, Callable, Dict, NamedTuple, Optional, Tuple, Union from urllib.parse import parse_qs, urlsplit -from ._envelope_shape import envelope_violation -from .codec import parse_envelope +from .codec import EnvelopeCodec, parse_envelope from .exceptions import BabelQueueError from .transport import NON_NUMERIC_ATTEMPTS_ROUTE, ReceivedMessage, Transport @@ -232,6 +231,10 @@ class _LegacyHandle(NamedTuple): class RedisTransport(Transport): + #: Classifies poison itself (undecodable, invalid UTF-8, identity-less): see ``_pop_legacy`` / + #: ``_zset_message``; the runtime does not classify the body a second time. + classifies_poison = True + def __init__( self, url: str = "redis://localhost:6379/0", @@ -441,18 +444,19 @@ def _pop_legacy(self, queue: str, budget: float) -> Optional[ReceivedMessage]: # The stored value may be a header frame (publish_with_headers) or a bare wire # envelope. The handle keeps the raw stored value so the settle LREM matches. unframed, headers = _unframe(text) - # A body the Go reference's typed ``Decode`` rejects (not valid JSON, not an object, - # or a member of the wrong type such as a string ``attempts``/``meta``) can never - # be decoded or retried: hand it over as poison so the runtime dead-letters the raw - # bytes on this first delivery instead of rebuilding and retrying it. An object - # without an identity is NOT poison (it keeps the bounded FAIL path), and neither is - # a bare ``null``. (``envelope_violation`` parses the body once; ``dispatch`` parses - # it again for the handler -- one extra parse of an already small body.) - # A stored value carrying invalid UTF-8 is poison even when it decodes (broker-bindings - # section 1.1): never delivered, retried or re-published, only dead-lettered. + # A body that does not decode into an envelope (not valid JSON, not an object, a + # member of the wrong type such as a string ``attempts``/``meta``: what the Go + # reference's typed ``Decode`` rejects) or that has neither a valid ``job`` nor a + # valid ``urn`` (ADR-0036 decision 4, broker-bindings section 1.1 "Identity-less + # body") can never be decoded, retried or routed: hand it over as poison so the + # runtime dead-letters the raw bytes on this first delivery, in every + # configuration, instead of rebuilding and retrying it. (``poison_cause`` parses the + # body once; ``dispatch`` parses it again for the handler -- one extra parse of an + # already small body.) A stored value carrying invalid UTF-8 is poison even when it + # decodes (section 1.1): never delivered, retried or re-published, only dead-lettered. route = ( "" - if not _has_invalid_utf8(text) and envelope_violation(unframed) is None + if not _has_invalid_utf8(text) and EnvelopeCodec.poison_cause(unframed) is None else NON_NUMERIC_ATTEMPTS_ROUTE ) return ReceivedMessage( @@ -515,7 +519,13 @@ def _zset_message(queue: str, job: str, reserved: str) -> ReceivedMessage: # The handler gets the RAW job; the reserved member (attempts + 1, or the raw job for a # poison body) is only the ack handle, so a failure counts the attempt exactly once. # Invalid UTF-8 is poison even when the reserve script could decode and re-encode it. - route = NON_NUMERIC_ATTEMPTS_ROUTE if job == reserved or _has_invalid_utf8(job) else "" + # An identity-less body (ADR-0036 decision 4) is poison on this layout too, in every + # configuration; the reserved member (not the raw job) is what the dead-letter script removes. + route = ( + NON_NUMERIC_ATTEMPTS_ROUTE + if job == reserved or _has_invalid_utf8(job) or EnvelopeCodec.poison_cause(job) is not None + else "" + ) return ReceivedMessage(body=job, queue=queue, handle=reserved, route_urn=route) def _migrate(self, ready: str) -> None: diff --git a/src/babelqueue/sqs_transport.py b/src/babelqueue/sqs_transport.py index 7f078c6..d154fcd 100644 --- a/src/babelqueue/sqs_transport.py +++ b/src/babelqueue/sqs_transport.py @@ -317,6 +317,9 @@ def pop(self, queue: str, timeout: float = 1.0) -> Optional[ReceivedMessage]: queue=queue, handle=handle, headers=headers, + # SQS counts deliveries itself: a poison body has no envelope to carry ``attempts``, + # so the runtime reads the native count (``ApproximateReceiveCount - 1``) instead. + native_attempts=self._native_attempts(receive_count), route_urn=( UNSUPPORTED_SCHEMA_ROUTE_PREFIX + quote(unsupported, safe="") if unsupported is not None @@ -324,6 +327,14 @@ def pop(self, queue: str, timeout: float = 1.0) -> Optional[ReceivedMessage]: ), ) + @staticmethod + def _native_attempts(receive_count: Any) -> int: + """``ApproximateReceiveCount - 1`` (the 0-based delivery count), 0 when absent or invalid.""" + try: + return max(int(receive_count) - 1, 0) + except (ValueError, TypeError): + return 0 + @staticmethod def _unsupported_schema_version(attributes: Any) -> Optional[str]: """The §3.7 step-3 gate on the ``bq-schema-version`` message attribute: the offending raw diff --git a/src/babelqueue/transport.py b/src/babelqueue/transport.py index 989f4f8..08cc849 100644 --- a/src/babelqueue/transport.py +++ b/src/babelqueue/transport.py @@ -48,6 +48,13 @@ class ReceivedMessage: #: to a :class:`SettlePublisher` on its own re-publishes (retry, DLQ) so the key is not lost; #: it never synthesises one. Not part of the envelope; ``None`` for every other transport. key: Optional[bytes] = None + #: The delivery count the broker itself reports for this delivery (0-based: ``0`` on the first + #: delivery), for transports that have one (SQS ``ApproximateReceiveCount - 1``, the Pulsar + #: ``bq-attempts`` property floored by the ``RedeliveryCount``). A **poison** body has no + #: decodable envelope to carry ``attempts``, so the runtime counts its bounded retry from the + #: larger of this value and the ``attempts`` of the (bare) body it re-publishes. ``0`` for + #: every transport that does not report one. + native_attempts: int = 0 class PoisonRoute(str): @@ -80,6 +87,13 @@ def __new__( class Transport(ABC): """Minimal broker contract: publish a raw body, pop one, acknowledge it.""" + #: ``True`` for a transport that already classifies poison bodies itself and hands them over + #: under a poison route (Kafka, Redis). The runtime then does not classify the body again; for + #: every other transport it applies the poison rule of ADR-0036 decisions 4 and 6 in + #: :meth:`BabelQueue.dispatch` (a body that does not decode, or has no identity, is never + #: handed to a handler and takes the poison route). + classifies_poison: bool = False + @abstractmethod def publish(self, queue: str, body: str) -> None: """Append an already-encoded envelope to ``queue``.""" diff --git a/tests/conformance/CONFORMANCE_VERSION b/tests/conformance/CONFORMANCE_VERSION index dc1e644..bd8bf88 100644 --- a/tests/conformance/CONFORMANCE_VERSION +++ b/tests/conformance/CONFORMANCE_VERSION @@ -1 +1 @@ -1.6.0 +1.7.0 diff --git a/tests/conformance/manifest.json b/tests/conformance/manifest.json index 63f7ca1..5d06e20 100644 --- a/tests/conformance/manifest.json +++ b/tests/conformance/manifest.json @@ -1,6 +1,6 @@ { "schema_version": 1, - "conformance_version": "1.6.0", + "conformance_version": "1.7.0", "sections": { "cases": { "scope": "core" @@ -79,6 +79,12 @@ }, "poison_dlq": { "scope": "core" + }, + "identity_verdict": { + "scope": "core" + }, + "pulsar_exhaustion": { + "scope": "pulsar" } }, "exceptions": [ @@ -5798,5 +5804,816 @@ } } ] + }, + "identity_verdict": { + "description": "Body classification outside Kafka (ADR-0036 decision 4; ADR-0032 section 6; error-handling.md section 3). Scope: `core`, because the decoder verdict is the same for every binding. For each case, feed `body` (a JSON string; it is the raw message body, so a case may hold invalid JSON, a BOM or duplicate keys) to the SDK core decoder and assert the verdict `expect`: `ok` - the body decodes into an envelope with a valid identity; `poison` - it does not (a decode error, or no valid identity). On a transport other than Kafka a `poison` body is never handed to a handler: it takes the poison route (broker-bindings.md section 5.5 and error-handling.md section 7; ADR-0035 record with `reason: poison`), not the unknown-URN strategies of error-handling.md section 3, which remain for a valid envelope whose URN has no handler. The case set starts with EVERY `kafka_poison` case (same names, bodies and order, the first 92 cases), with `no_identity` mapped to `poison` and `ok` / `poison` unchanged, so the two sections stay in step; the 5 cases after them are extra rows that pin the ADR-0036 decision 4 bodies `kafka_poison` does not hold (a blank or non-string `job` next to a valid `urn`, and a tab-only `job`) (a body with a valid identity but a wrongly typed known member is `poison` in both); a CI oracle derives the first 92 cases from `kafka_poison`, fails on any difference, and checks the extra rows with its own identity rule. The duplicate-key rule (the LAST value of a repeated top-level key wins) and the identity rule are those of `kafka_poison`: a `job` that is not a string is ignored; a blank (empty or whitespace-only) string `job` falls back to `urn`; a `urn` that is not a non-blank string is ignored; when neither is valid the body has no identity. Rows that pin the ADR-0036 decision 4 table (a `job` of `\"\"`, `\" \\t\"`, `null` or `{}` next to a valid `urn` is `ok`; see also the extra rows): a valid envelope (`full valid`) and a body whose only valid identity is `urn` (`urn alias string`), a blank `job` with a valid `urn` (`job blank with valid urn alias`), a non-string `job` with a valid `urn` (`job number with valid urn alias`) are `ok`; `{}` (`empty object`), a non-string `job` without `urn` (`job number`, `job null`), a non-string `urn` without `job` (`urn alias number`), non-object top-level values (`null literal`, `empty array`, `string`, `number`) and a non-JSON or empty body (`not json`, `empty`) are `poison`. The edge that a transport routes on an out-of-band identity property (Pulsar / Kafka `bq-job`) is outside this section.", + "cases": [ + { + "name": "empty", + "body": "", + "expect": "poison" + }, + { + "name": "whitespace only", + "body": " ", + "expect": "poison" + }, + { + "name": "not json", + "body": "not-json", + "expect": "poison" + }, + { + "name": "truncated", + "body": "{\"job\":\"urn:a\"", + "expect": "poison" + }, + { + "name": "empty array", + "body": "[]", + "expect": "poison" + }, + { + "name": "array", + "body": "[1]", + "expect": "poison" + }, + { + "name": "string", + "body": "\"str\"", + "expect": "poison" + }, + { + "name": "number", + "body": "123", + "expect": "poison" + }, + { + "name": "boolean", + "body": "true", + "expect": "poison" + }, + { + "name": "null literal", + "body": "null", + "expect": "poison" + }, + { + "name": "empty object", + "body": "{}", + "expect": "poison" + }, + { + "name": "full valid", + "body": "{\"job\":\"urn:babel:orders:created\",\"trace_id\":\"t\",\"data\":{},\"meta\":{\"id\":\"i\",\"queue\":\"q\",\"lang\":\"php\",\"schema_version\":1,\"created_at\":1700000000000},\"attempts\":0}", + "expect": "ok" + }, + { + "name": "job number", + "body": "{\"job\":5}", + "expect": "poison" + }, + { + "name": "job null", + "body": "{\"job\":null}", + "expect": "poison" + }, + { + "name": "urn alias number", + "body": "{\"urn\":5}", + "expect": "poison" + }, + { + "name": "urn alias string", + "body": "{\"urn\":\"urn:x\"}", + "expect": "ok" + }, + { + "name": "urn alias padded", + "body": "{\"urn\":\" urn:x \"}", + "expect": "ok" + }, + { + "name": "trace_id number", + "body": "{\"trace_id\":7}", + "expect": "poison" + }, + { + "name": "data array", + "body": "{\"data\":[]}", + "expect": "poison" + }, + { + "name": "data string", + "body": "{\"data\":\"x\"}", + "expect": "poison" + }, + { + "name": "data null", + "body": "{\"data\":null}", + "expect": "poison" + }, + { + "name": "meta string", + "body": "{\"meta\":\"x\"}", + "expect": "poison" + }, + { + "name": "meta array", + "body": "{\"meta\":[]}", + "expect": "poison" + }, + { + "name": "meta null", + "body": "{\"meta\":null}", + "expect": "poison" + }, + { + "name": "meta.id number", + "body": "{\"meta\":{\"id\":5}}", + "expect": "poison" + }, + { + "name": "meta.queue number", + "body": "{\"meta\":{\"queue\":5}}", + "expect": "poison" + }, + { + "name": "meta.lang boolean", + "body": "{\"meta\":{\"lang\":true}}", + "expect": "poison" + }, + { + "name": "meta.created_at string", + "body": "{\"meta\":{\"created_at\":\"2026\"}}", + "expect": "poison" + }, + { + "name": "meta.created_at fractional", + "body": "{\"meta\":{\"created_at\":1.5}}", + "expect": "poison" + }, + { + "name": "meta.created_at integer", + "body": "{\"meta\":{\"created_at\":1700000000000}}", + "expect": "poison" + }, + { + "name": "meta.created_at exponent", + "body": "{\"meta\":{\"created_at\":1e3}}", + "expect": "poison" + }, + { + "name": "meta.created_at decimal zero", + "body": "{\"meta\":{\"created_at\":1.0}}", + "expect": "poison" + }, + { + "name": "meta.created_at int64 overflow", + "body": "{\"meta\":{\"created_at\":9223372036854775808}}", + "expect": "poison" + }, + { + "name": "meta.created_at int64 max", + "body": "{\"meta\":{\"created_at\":9223372036854775807}}", + "expect": "poison" + }, + { + "name": "meta.created_at negative", + "body": "{\"meta\":{\"created_at\":-1}}", + "expect": "poison" + }, + { + "name": "meta.created_at negative zero", + "body": "{\"meta\":{\"created_at\":-0}}", + "expect": "poison" + }, + { + "name": "meta.schema_version string", + "body": "{\"meta\":{\"schema_version\":\"1\"}}", + "expect": "poison" + }, + { + "name": "meta.schema_version two", + "body": "{\"meta\":{\"schema_version\":2}}", + "expect": "poison" + }, + { + "name": "meta.schema_version decimal zero", + "body": "{\"meta\":{\"schema_version\":1.0}}", + "expect": "poison" + }, + { + "name": "meta.schema_version null", + "body": "{\"meta\":{\"schema_version\":null}}", + "expect": "poison" + }, + { + "name": "meta.schema_version boolean", + "body": "{\"meta\":{\"schema_version\":true}}", + "expect": "poison" + }, + { + "name": "attempts fractional", + "body": "{\"attempts\":1.5}", + "expect": "poison" + }, + { + "name": "attempts string", + "body": "{\"attempts\":\"3\"}", + "expect": "poison" + }, + { + "name": "attempts integer", + "body": "{\"attempts\":2}", + "expect": "poison" + }, + { + "name": "attempts exponent", + "body": "{\"attempts\":1e0}", + "expect": "poison" + }, + { + "name": "attempts upper exponent", + "body": "{\"attempts\":1E2}", + "expect": "poison" + }, + { + "name": "attempts negative zero", + "body": "{\"attempts\":-0}", + "expect": "poison" + }, + { + "name": "attempts overflow", + "body": "{\"attempts\":9223372036854775808}", + "expect": "poison" + }, + { + "name": "attempts null", + "body": "{\"attempts\":null}", + "expect": "poison" + }, + { + "name": "dead_letter string", + "body": "{\"dead_letter\":\"x\"}", + "expect": "poison" + }, + { + "name": "dead_letter null", + "body": "{\"dead_letter\":null}", + "expect": "poison" + }, + { + "name": "dead_letter empty object", + "body": "{\"dead_letter\":{}}", + "expect": "poison" + }, + { + "name": "dead_letter.attempts string", + "body": "{\"dead_letter\":{\"attempts\":\"3\"}}", + "expect": "poison" + }, + { + "name": "dead_letter.reason number", + "body": "{\"dead_letter\":{\"reason\":5}}", + "expect": "poison" + }, + { + "name": "dead_letter.error number", + "body": "{\"dead_letter\":{\"error\":5}}", + "expect": "poison" + }, + { + "name": "dead_letter nullable strings", + "body": "{\"dead_letter\":{\"error\":null,\"exception\":null}}", + "expect": "poison" + }, + { + "name": "dead_letter.failed_at string", + "body": "{\"dead_letter\":{\"failed_at\":\"x\"}}", + "expect": "poison" + }, + { + "name": "dead_letter.failed_at fractional", + "body": "{\"dead_letter\":{\"failed_at\":1.5}}", + "expect": "poison" + }, + { + "name": "dead_letter.original_queue array", + "body": "{\"dead_letter\":{\"original_queue\":[]}}", + "expect": "poison" + }, + { + "name": "dead_letter.lang object", + "body": "{\"dead_letter\":{\"lang\":{}}}", + "expect": "poison" + }, + { + "name": "dead_letter full", + "body": "{\"dead_letter\":{\"error\":\"e\",\"exception\":\"x\",\"failed_at\":1,\"original_queue\":\"q\",\"attempts\":1,\"lang\":\"go\",\"reason\":\"r\"}}", + "expect": "poison" + }, + { + "name": "upper-case job key with number is an unknown member (names are case-sensitive)", + "body": "{\"JOB\":5}", + "expect": "poison" + }, + { + "name": "mixed-case meta.id key with number is an unknown member (names are case-sensitive)", + "body": "{\"Meta\":{\"ID\":5}}", + "expect": "poison" + }, + { + "name": "upper-case job key with string", + "body": "{\"JOB\":\"urn:x\"}", + "expect": "poison" + }, + { + "name": "trailing garbage", + "body": "{\"job\":\"x\"} trailing", + "expect": "poison" + }, + { + "name": "two objects", + "body": "{\"job\":\"x\"}{\"job\":\"y\"}", + "expect": "poison" + }, + { + "name": "unknown keys", + "body": "{\"x\":1}", + "expect": "poison" + }, + { + "name": "forbidden keys with odd types", + "body": "{\"timestamp\":\"x\",\"meta\":{\"max_retries\":\"x\",\"attempts\":\"y\",\"source\":5,\"ts\":\"z\"}}", + "expect": "poison" + }, + { + "name": "nested lookalikes in data", + "body": "{\"data\":{\"a\":[1,2,{\"b\":null}],\"meta\":5,\"job\":5},\"extra\":{\"meta\":5}}", + "expect": "poison" + }, + { + "name": "duplicate job string then number", + "body": "{\"job\":\"urn:x\",\"job\":5}", + "expect": "poison" + }, + { + "name": "duplicate job number then string", + "body": "{\"job\":5,\"job\":\"urn:x\"}", + "expect": "ok" + }, + { + "name": "byte order mark", + "body": "\ufeff{\"job\":\"x\"}", + "expect": "poison" + }, + { + "name": "leading zero number", + "body": "{\"attempts\":01}", + "expect": "poison" + }, + { + "name": "NaN literal", + "body": "{\"attempts\":NaN}", + "expect": "poison" + }, + { + "name": "single quotes", + "body": "{'a':1}", + "expect": "poison" + }, + { + "name": "meta duplicate-case key", + "body": "{\"meta\":{\"id\":\"a\",\"queue\":\"b\",\"lang\":\"c\",\"schema_version\":1,\"created_at\":1,\"ID\":\"dup\"}}", + "expect": "poison" + }, + { + "name": "attempts huge exponent", + "body": "{\"attempts\":1e400}", + "expect": "poison" + }, + { + "name": "data with huge number", + "body": "{\"data\":{\"n\":1e400}}", + "expect": "poison" + }, + { + "name": "job number with valid urn alias", + "body": "{\"job\":5,\"urn\":\"urn:x\"}", + "expect": "ok" + }, + { + "name": "job blank with valid urn alias", + "body": "{\"job\":\" \",\"urn\":\"urn:x\"}", + "expect": "ok" + }, + { + "name": "job valid with urn number", + "body": "{\"job\":\"urn:x\",\"urn\":5}", + "expect": "ok" + }, + { + "name": "job blank without urn", + "body": "{\"job\":\" \"}", + "expect": "poison" + }, + { + "name": "job and urn both empty strings", + "body": "{\"job\":\"\",\"urn\":\"\"}", + "expect": "poison" + }, + { + "name": "job and urn both arrays", + "body": "{\"job\":[],\"urn\":[]}", + "expect": "poison" + }, + { + "name": "job object without urn", + "body": "{\"job\":{}}", + "expect": "poison" + }, + { + "name": "identity with null data", + "body": "{\"job\":\"urn:x\",\"data\":null}", + "expect": "ok" + }, + { + "name": "identity with null meta", + "body": "{\"job\":\"urn:x\",\"meta\":null}", + "expect": "ok" + }, + { + "name": "identity with null attempts", + "body": "{\"job\":\"urn:x\",\"attempts\":null}", + "expect": "ok" + }, + { + "name": "identity with null dead_letter", + "body": "{\"job\":\"urn:x\",\"dead_letter\":null}", + "expect": "ok" + }, + { + "name": "identity with integer attempts", + "body": "{\"job\":\"urn:x\",\"attempts\":2}", + "expect": "ok" + }, + { + "name": "identity with wrong typed attempts", + "body": "{\"job\":\"urn:x\",\"attempts\":\"3\"}", + "expect": "poison" + }, + { + "name": "identity with wrong typed data", + "body": "{\"job\":\"urn:x\",\"data\":[]}", + "expect": "poison" + }, + { + "name": "job empty string with valid urn", + "body": "{\"job\":\"\",\"urn\":\"urn:x\"}", + "expect": "ok" + }, + { + "name": "job space and tab with valid urn", + "body": "{\"job\":\" \\t\",\"urn\":\"urn:x\"}", + "expect": "ok" + }, + { + "name": "job null with valid urn", + "body": "{\"job\":null,\"urn\":\"urn:x\"}", + "expect": "ok" + }, + { + "name": "job object with valid urn", + "body": "{\"job\":{},\"urn\":\"urn:x\"}", + "expect": "ok" + }, + { + "name": "job space and tab without urn", + "body": "{\"job\":\" \\t\"}", + "expect": "poison" + } + ] + }, + "pulsar_exhaustion": { + "description": "Pulsar max-attempts settle decision (ADR-0036 decisions 1 and 2, amends ADR-0014; broker-bindings.md section 5.5 \"Max attempts and poison\"). Scope: `pulsar`. A pure function, no broker: for each case compute the outcome the consumer takes for a delivery and assert `expect`. Case `{name, kind, body_attempts?, bq_attempts?, redelivery_count, max_attempts?, dead_letter, on_unknown_urn?, expect}`. `kind`: `failed` (the handler throws), `unknown_urn` (an unmapped URN with no hook; strategy `fail`), `poison` (the body does not decode into an envelope, including an object with no valid identity, see `identity_verdict`). Inputs: `body_attempts` is the envelope's top-level `attempts` (absent for `poison`, which has no decodable body: its input is `bq_attempts` and `redelivery_count` only); `bq_attempts` is the `bq-attempts` message property, a string (absent = not set); `redelivery_count` is the native Pulsar RedeliveryCount (0-based); `max_attempts` absent = the default 3 (the total number of deliveries, the first included; a value below 1 is a SHOULD-level configuration error and is not a case); `dead_letter` is a boolean (dead-lettering on or off) and `off` has two meanings: in the Go and Python runtimes it is the dead-letter SETTING only, the DLQ target stays writable and the `dead_letter` strategy still writes a record; in the PHP `Consume\\Dispatcher` \"on\" means a `DeadLetterPublisher` is configured, so `off` means NO publisher and nothing can be written; `on_unknown_urn` absent = `fail`. Derivation: `reconciled` = `bq_attempts` when present and valid, else `body_attempts`, then `max(reconciled, redelivery_count)` (pulsar.attempts_reconciliation); `counted = reconciled + 1`. Decision. `failed` and `unknown_urn` (strategy `fail`): `counted < max_attempts` is `retry`; otherwise dead-lettering on is `dlq` (`reason` `failed` or `unknown_urn`, `dead_letter_attempts = counted`) and off is `drop` (acknowledge and log). `poison`: `on_unknown_urn: delete` is `drop`; `on_unknown_urn: dead_letter` is `dlq` with `reason: poison` whatever `dead_letter` says (the strategy writes the record where a DLQ writer exists; see `outcome_no_writer` below); dead-lettering on (strategy `fail` or `release`) is `dlq` with `reason: poison` and NO retry (the ADR-0035 record, `dead_letter_attempts` is not compared); dead-lettering off takes the bounded `fail` path (`release` is replaced by it) - `counted < max_attempts` is `retry`, otherwise `drop`. `unknown_urn` with `on_unknown_urn: delete` is `drop`; with `dead_letter` it is a direct `dlq` with `reason: unknown_urn` whatever `dead_letter` says (`dead_letter_attempts` is not compared; see `outcome_no_writer` below). `expect.outcome_no_writer` (optional, `drop`): present only on a row whose result differs for an SDK without a DLQ writer (the PHP `Dispatcher` without a publisher: `on_unknown_urn: dead_letter` is acknowledged, dropped and logged; owner decision 2026-10-25, ADR-0036 decision 3). A runner for such an SDK asserts `expect.outcome_no_writer` instead of `expect.outcome`; every other runner asserts `expect.outcome` and ignores the field; when the field is absent both kinds of SDK assert `expect.outcome`. On Kafka the PHP `Dispatcher` rethrows instead of dropping (outside this Pulsar-scoped section). Rows with any `on_unknown_urn` value (they are the rows with that key) apply only to an SDK that has the strategy option (the Go and Python runtimes and the PHP `Consume\\Dispatcher`, ADR-0036 decision 3); a runner for the Java, .NET or Node consumer, which have no such option, skips them and reports them as `skip`, not `pass`. `unknown_urn` with `release` is not a case: on Pulsar the Go and Python runtimes release it immediately without counting (the `unknown_urn_release_delay` known gap of broker-bindings.md section 5.5). Not covered here (no broker-less test): the DLQ write itself, the acknowledgement after a successful write, the negative acknowledgement after a failed one, and the PHP queue name of ADR-0036 decision 3; each SDK covers them in its own tests.", + "cases": [ + { + "name": "failed-reconciled-0-dlq-on-retries", + "kind": "failed", + "body_attempts": 0, + "redelivery_count": 0, + "dead_letter": true, + "expect": { + "outcome": "retry" + } + }, + { + "name": "failed-reconciled-1-dlq-on-retries", + "kind": "failed", + "body_attempts": 1, + "redelivery_count": 0, + "dead_letter": true, + "expect": { + "outcome": "retry" + } + }, + { + "name": "failed-reconciled-2-dlq-on-dead-letters", + "kind": "failed", + "body_attempts": 2, + "redelivery_count": 0, + "dead_letter": true, + "expect": { + "outcome": "dlq", + "reason": "failed", + "dead_letter_attempts": 3 + } + }, + { + "name": "failed-reconciled-2-dlq-off-drops", + "kind": "failed", + "body_attempts": 2, + "redelivery_count": 0, + "dead_letter": false, + "expect": { + "outcome": "drop" + } + }, + { + "name": "failed-reconciled-7-dlq-on-counts-past-the-cap", + "kind": "failed", + "body_attempts": 7, + "redelivery_count": 0, + "dead_letter": true, + "expect": { + "outcome": "dlq", + "reason": "failed", + "dead_letter_attempts": 8 + } + }, + { + "name": "failed-max-1-dlq-on-first-delivery-dead-letters", + "kind": "failed", + "body_attempts": 0, + "redelivery_count": 0, + "max_attempts": 1, + "dead_letter": true, + "expect": { + "outcome": "dlq", + "reason": "failed", + "dead_letter_attempts": 1 + } + }, + { + "name": "failed-max-1-dlq-off-first-delivery-drops", + "kind": "failed", + "body_attempts": 0, + "redelivery_count": 0, + "max_attempts": 1, + "dead_letter": false, + "expect": { + "outcome": "drop" + } + }, + { + "name": "failed-max-5-reconciled-3-dlq-off-retries", + "kind": "failed", + "body_attempts": 3, + "redelivery_count": 0, + "max_attempts": 5, + "dead_letter": false, + "expect": { + "outcome": "retry" + } + }, + { + "name": "failed-max-5-reconciled-4-dlq-off-drops", + "kind": "failed", + "body_attempts": 4, + "redelivery_count": 0, + "max_attempts": 5, + "dead_letter": false, + "expect": { + "outcome": "drop" + } + }, + { + "name": "unknown-urn-reconciled-1-dlq-on-retries", + "kind": "unknown_urn", + "body_attempts": 1, + "redelivery_count": 0, + "dead_letter": true, + "expect": { + "outcome": "retry" + } + }, + { + "name": "unknown-urn-reconciled-2-dlq-on-dead-letters", + "kind": "unknown_urn", + "body_attempts": 2, + "redelivery_count": 0, + "dead_letter": true, + "expect": { + "outcome": "dlq", + "reason": "unknown_urn", + "dead_letter_attempts": 3 + } + }, + { + "name": "unknown-urn-reconciled-2-dlq-off-drops", + "kind": "unknown_urn", + "body_attempts": 2, + "redelivery_count": 0, + "dead_letter": false, + "expect": { + "outcome": "drop" + } + }, + { + "name": "poison-reconciled-0-dlq-on-dead-letters-without-retry", + "kind": "poison", + "redelivery_count": 0, + "dead_letter": true, + "expect": { + "outcome": "dlq", + "reason": "poison" + } + }, + { + "name": "poison-reconciled-0-dlq-off-retries", + "kind": "poison", + "redelivery_count": 0, + "dead_letter": false, + "expect": { + "outcome": "retry" + } + }, + { + "name": "poison-reconciled-2-dlq-off-drops", + "kind": "poison", + "redelivery_count": 2, + "dead_letter": false, + "expect": { + "outcome": "drop" + } + }, + { + "name": "poison-delete-dlq-on-drops", + "kind": "poison", + "redelivery_count": 0, + "dead_letter": true, + "on_unknown_urn": "delete", + "expect": { + "outcome": "drop" + } + }, + { + "name": "poison-delete-dlq-off-drops", + "kind": "poison", + "redelivery_count": 0, + "dead_letter": false, + "on_unknown_urn": "delete", + "expect": { + "outcome": "drop" + } + }, + { + "name": "poison-release-dlq-on-dead-letters-without-retry", + "kind": "poison", + "redelivery_count": 0, + "dead_letter": true, + "on_unknown_urn": "release", + "expect": { + "outcome": "dlq", + "reason": "poison" + } + }, + { + "name": "poison-release-dlq-off-reconciled-0-takes-the-bounded-fail-path", + "kind": "poison", + "redelivery_count": 0, + "dead_letter": false, + "on_unknown_urn": "release", + "expect": { + "outcome": "retry" + } + }, + { + "name": "poison-release-dlq-off-reconciled-2-drops-not-released-forever", + "kind": "poison", + "redelivery_count": 2, + "dead_letter": false, + "on_unknown_urn": "release", + "expect": { + "outcome": "drop" + } + }, + { + "name": "poison-dead-letter-strategy-dlq-off-dead-letters", + "kind": "poison", + "redelivery_count": 0, + "dead_letter": false, + "on_unknown_urn": "dead_letter", + "expect": { + "outcome": "dlq", + "reason": "poison", + "outcome_no_writer": "drop" + } + }, + { + "name": "unknown-urn-delete-dlq-on-drops", + "kind": "unknown_urn", + "body_attempts": 0, + "redelivery_count": 0, + "dead_letter": true, + "on_unknown_urn": "delete", + "expect": { + "outcome": "drop" + } + }, + { + "name": "unknown-urn-dead-letter-strategy-dlq-off-dead-letters-directly", + "kind": "unknown_urn", + "body_attempts": 0, + "redelivery_count": 0, + "dead_letter": false, + "on_unknown_urn": "dead_letter", + "expect": { + "outcome": "dlq", + "reason": "unknown_urn", + "outcome_no_writer": "drop" + } + }, + { + "name": "poison-dead-letter-strategy-dlq-on-dead-letters", + "kind": "poison", + "redelivery_count": 0, + "dead_letter": true, + "on_unknown_urn": "dead_letter", + "expect": { + "outcome": "dlq", + "reason": "poison" + } + }, + { + "name": "unknown-urn-dead-letter-strategy-dlq-on-dead-letters-directly", + "kind": "unknown_urn", + "body_attempts": 0, + "redelivery_count": 0, + "dead_letter": true, + "on_unknown_urn": "dead_letter", + "expect": { + "outcome": "dlq", + "reason": "unknown_urn" + } + }, + { + "name": "reconciliation-bq-attempts-string-over-body", + "kind": "failed", + "body_attempts": 0, + "bq_attempts": "2", + "redelivery_count": 0, + "dead_letter": true, + "expect": { + "outcome": "dlq", + "reason": "failed", + "dead_letter_attempts": 3 + } + }, + { + "name": "reconciliation-redelivery-count-floors-body", + "kind": "failed", + "body_attempts": 1, + "redelivery_count": 2, + "dead_letter": true, + "expect": { + "outcome": "dlq", + "reason": "failed", + "dead_letter_attempts": 3 + } + }, + { + "name": "reconciliation-bq-attempts-below-cap-retries", + "kind": "unknown_urn", + "body_attempts": 0, + "bq_attempts": "1", + "redelivery_count": 0, + "dead_letter": true, + "expect": { + "outcome": "retry" + } + }, + { + "name": "reconciliation-poison-bq-attempts-dlq-off-drops", + "kind": "poison", + "bq_attempts": "2", + "redelivery_count": 0, + "dead_letter": false, + "expect": { + "outcome": "drop" + } + } + ] } } diff --git a/tests/conformance/schema/results.example.json b/tests/conformance/schema/results.example.json index 9130d29..cd422f6 100644 --- a/tests/conformance/schema/results.example.json +++ b/tests/conformance/schema/results.example.json @@ -1,7 +1,7 @@ { "sdk": "example-sdk/redis", "version": "1.9.0", - "conformance_version": "1.6.0", + "conformance_version": "1.7.0", "generated_at": "2026-10-18T09:30:00Z", "bindings": [ "redis", diff --git a/tests/test_app.py b/tests/test_app.py index 5af81d5..327b048 100644 --- a/tests/test_app.py +++ b/tests/test_app.py @@ -5,6 +5,7 @@ from __future__ import annotations +import json import unittest from babelqueue import BabelQueue, EnvelopeCodec, UnknownUrnStrategy @@ -423,9 +424,10 @@ def test_unknown_urn_release_still_releases_verbatim(self) -> None: self.assertEqual(app.transport.size("orders.dlq"), 0) self.assertEqual(app.transport._queues["orders"].popleft(), body) # noqa: SLF001 - def test_ungated_poison_is_bounded_under_release(self) -> None: - # Not version-gated: an undecodable body and a body naming no URN can never gain a handler - # either, so RELEASE takes the bounded FAIL path for them too (Go routePoison parity). + def test_poison_is_dead_lettered_at_once_under_release_when_the_dlq_is_on(self) -> None: + # ADR-0036 decisions 2/4/6: a body that does not decode, or names no identity, is poison. + # With the DLQ on it is written as the ADR-0035 record (reason ``poison``) on its first + # delivery, with no retry, whatever the unknown-URN strategy says. no_urn = EnvelopeCodec.encode(EnvelopeCodec.make("urn:babel:orders:created", {}, queue="orders")) no_urn = no_urn.replace('"urn:babel:orders:created"', '""') for name, body in (("undecodable", "{not json"), ("no-urn", no_urn)): @@ -434,9 +436,26 @@ def test_ungated_poison_is_bounded_under_release(self) -> None: "memory://", queue="orders", max_attempts=3, dead_letter=True, on_unknown_urn=UnknownUrnStrategy.RELEASE, ) - self.assertEqual(self._redeliver_until_settled(app, "orders", "", body), 3) + self.assertEqual(self._redeliver_until_settled(app, "orders", "", body), 1) self.assertEqual(app.transport.size("orders"), 0) self.assertEqual(app.transport.size("orders.dlq"), 1) + record = json.loads(app.transport._queues["orders.dlq"][0]) # noqa: SLF001 + self.assertEqual(record["dead_letter"]["reason"], "poison") + + def test_poison_is_bounded_under_release_when_the_dlq_is_off(self) -> None: + # Dead-lettering off: RELEASE is replaced by the bounded FAIL path (a poison body would + # otherwise be released forever): max_attempts deliveries, then a logged drop. + no_urn = EnvelopeCodec.encode(EnvelopeCodec.make("urn:babel:orders:created", {}, queue="orders")) + no_urn = no_urn.replace('"urn:babel:orders:created"', '""') + for name, body in (("undecodable", "{not json"), ("no-urn", no_urn)): + with self.subTest(body=name): + app = BabelQueue( + "memory://", queue="orders", max_attempts=3, on_unknown_urn=UnknownUrnStrategy.RELEASE, + ) + with self.assertLogs("babelqueue.app", level="WARNING"): + self.assertEqual(self._redeliver_until_settled(app, "orders", "", body), 3) + self.assertEqual(app.transport.size("orders"), 0) + self.assertEqual(app.transport.size("orders.dlq"), 0) def test_unsupported_broker_scheme_raises(self) -> None: from babelqueue import BabelQueueError diff --git a/tests/test_behaviour_conformance.py b/tests/test_behaviour_conformance.py index b737691..db80a39 100644 --- a/tests/test_behaviour_conformance.py +++ b/tests/test_behaviour_conformance.py @@ -29,6 +29,8 @@ import redis_backend from test_kafka_transport import FakeConsumer, FakeProducer, OffsetConsumer, OffsetMessage, ReportingProducer from test_kafka_transport import transport as poison_transport +from test_pulsar_transport import FakeMessage as PulsarFakeMessage +from test_pulsar_transport import Timeout as PulsarTimeout from test_sqs_transport import FakeSQS from babelqueue import BabelQueue, BabelQueueError, EnvelopeCodec @@ -40,6 +42,7 @@ from babelqueue._poison_record import build_poison_record from babelqueue.kafka_transport import DEFAULT_RETRY_TIERS_MS, POISON_ROUTE, KafkaTransport, retry_topic from babelqueue.pika_transport import PikaTransport +from babelqueue.pulsar_transport import PulsarTransport from babelqueue.redis_transport import RedisTransport, _LegacyHandle from babelqueue.schema import validate_schema from babelqueue.sqs_transport import SqsTransport @@ -1296,6 +1299,145 @@ def _eval_kafka_poison(case: Mapping[str, Any]) -> None: _ensure(message.body == body, "a poison record is handed over raw") +def _eval_identity_verdict(case: Mapping[str, Any]) -> None: + """``identity_verdict``: the core decoder's verdict outside Kafka (ADR-0036 decision 4), then + the fate on a transport that does not classify poison itself (the memory transport): a + ``poison`` body never reaches a handler and is dead-lettered as the ADR-0035 record with + ``reason: "poison"`` and the original body preserved; an ``ok`` body reaches its handler.""" + body = case["body"] + verdict = "ok" if EnvelopeCodec.poison_cause(body) is None else "poison" + _ensure(verdict == case["expect"], f"verdict: {verdict!r} != {case['expect']!r}") + + calls: List[Any] = [] + app = BabelQueue("memory://", queue="orders", dead_letter=True, max_attempts=3) + if verdict == "ok": + urn = EnvelopeCodec.urn(EnvelopeCodec.decode(body)) # must not raise + _ensure(urn != "", "an ok body resolves an identity") + app.register(urn, lambda data, meta: calls.append(urn)) + else: + app.register("urn:babel:orders:created", lambda data, meta: calls.append("called")) + app.transport.publish("orders", body) + app.consume("orders", max_messages=1, timeout=0.01, handle_signals=False) + + if verdict == "ok": + _ensure(len(calls) == 1, "an ok body must reach its handler once") + _ensure(app.transport.size("orders.dlq") == 0, "an ok body is not dead-lettered") + return + _ensure(calls == [], "a poison body must never reach a handler") + _ensure(app.transport.size("orders") == 0, "a poison body must not be retried (dead-lettering is on)") + _ensure(app.transport.size("orders.dlq") == 1, "a poison body must be dead-lettered once") + record_text = app.transport._queues["orders.dlq"][0] # noqa: SLF001 + record = json.loads(record_text) + block = record["dead_letter"] + _ensure(block["reason"] == "poison", f"reason: {block['reason']!r} is not 'poison'") + _ensure(block["original_queue"] == "orders", "original_queue") + if "raw" in record and "raw_encoding" not in record: + _ensure(record["raw"] == body, "the wrapper must carry the original body") + else: # spliced: every original character is kept, only a dead_letter member is appended + trimmed = body.strip(" \t\r\n") + _ensure(record_text.startswith(trimmed[:-1]), "the spliced record must keep the original body") + + +class _TopicProducer: + def __init__(self, sent: List[Tuple[str, str, Any]], topic: str) -> None: + self._sent, self._topic = sent, topic + + def send(self, content: Any, properties: Any = None) -> None: + text = content.decode("utf-8") if isinstance(content, (bytes, bytearray)) else str(content) + self._sent.append((self._topic, text, properties)) + + +class _OnceConsumer: + def __init__(self, message: Any) -> None: + self._messages = [message] + self.acked: List[Any] = [] + + def receive(self, timeout_millis: Any = None) -> Any: + if not self._messages: + raise PulsarTimeout() + return self._messages.pop(0) + + def acknowledge(self, message: Any) -> None: + self.acked.append(message) + + +class _PulsarClient: + def __init__(self, consumer: _OnceConsumer) -> None: + self.consumer = consumer + self.sent: List[Tuple[str, str, Any]] = [] + + def create_producer(self, topic: str) -> _TopicProducer: + return _TopicProducer(self.sent, topic) + + def subscribe(self, topic: str, subscription: str, **kwargs: Any) -> _OnceConsumer: + return self.consumer + + +def _eval_pulsar_exhaustion(case: Mapping[str, Any]) -> None: + """``pulsar_exhaustion``: the settle decision of ADR-0036 decisions 1 and 2, driven through the + real ``PulsarTransport`` (reconciliation) and runtime (the decision) over a duck-typed client. + Python has the ``on_unknown_urn`` option, so every row runs (none is skipped).""" + kind = case["kind"] + expect = case["expect"] + max_attempts = case.get("max_attempts", 3) + properties = {"bq-attempts": case["bq_attempts"]} if "bq_attempts" in case else None + if kind == "poison": + body = "not json" + else: + envelope = EnvelopeCodec.make("urn:babel:orders:created", {"k": 1}, queue="orders") + envelope["attempts"] = case["body_attempts"] + body = EnvelopeCodec.encode(envelope) + reconciled = case.get("body_attempts", 0) + if "bq_attempts" in case: + reconciled = int(case["bq_attempts"]) + reconciled = max(reconciled, case["redelivery_count"]) + counted = reconciled + 1 + + consumer = _OnceConsumer(PulsarFakeMessage(body, case["redelivery_count"], properties)) + client = _PulsarClient(consumer) + app = BabelQueue( + transport=PulsarTransport("pulsar://localhost:6650", client=client), + queue="orders", + max_attempts=max_attempts, + dead_letter=case["dead_letter"], + on_unknown_urn=case.get("on_unknown_urn", "fail"), + ) + calls: List[str] = [] + if kind == "failed": + def fail(data: Any, meta: Any) -> None: + calls.append("called") + raise RuntimeError("boom") + + app.register("urn:babel:orders:created", fail) + else: + app.register("urn:babel:other", lambda data, meta: calls.append("called")) + app.consume("orders", max_messages=1, timeout=0.01, handle_signals=False) + + _ensure(calls == ([] if kind != "failed" else ["called"]), f"handler calls: {calls!r}") + topics = [topic for topic, _, _ in client.sent] + if topics == ["orders.dlq"]: + outcome = "dlq" + elif topics == ["orders"]: + outcome = "retry" + elif topics == []: + outcome = "drop" + else: + raise AssertionError(f"unexpected publishes: {topics!r}") + _ensure(outcome == expect["outcome"], f"outcome: {outcome!r} != {expect['outcome']!r}") + _ensure(len(consumer.acked) == 1, "the original is acknowledged exactly once") + if outcome == "retry": + retried = json.loads(client.sent[0][1]) + _ensure(retried["attempts"] == counted, f"retry attempts {retried['attempts']!r} != {counted!r}") + if outcome == "dlq": + block = json.loads(client.sent[0][1])["dead_letter"] + _ensure(block["reason"] == expect["reason"], f"reason: {block['reason']!r} != {expect['reason']!r}") + if "dead_letter_attempts" in expect: + _ensure( + block["attempts"] == expect["dead_letter_attempts"], + f"dead_letter.attempts {block['attempts']!r} != {expect['dead_letter_attempts']!r}", + ) + + def _top_level_names(text: str) -> List[str]: """Every top-level member name of the JSON object ``text`` (duplicates kept, last-wins not applied).""" names: List[List[str]] = [] @@ -1421,8 +1563,8 @@ def test_a_non_string_job_is_warned_about_and_a_blank_one_is_not(self) -> None: class RunnerCaseIdTest(unittest.TestCase): - #: ``section -> number of case ids`` of the vendored conformance 1.6.0 manifest. - EXPECTED_COUNTS = {"trim": 35, "job_type": 11, "traceparent": 14, "delay": 9, "backoff": 29, "redis_layout": 59, "redis_delay": 8, "retry_backoff": 18, "kafka_poison": 92, "poison_dlq": 43, "idempotency": 7, "redis": 1, "rabbitmq": 1} + #: ``section -> number of case ids`` of the vendored conformance 1.7.0 manifest. + EXPECTED_COUNTS = {"identity_verdict": 97, "pulsar_exhaustion": 29, "trim": 35, "job_type": 11, "traceparent": 14, "delay": 9, "backoff": 29, "redis_layout": 59, "redis_delay": 8, "retry_backoff": 18, "kafka_poison": 92, "poison_dlq": 43, "idempotency": 7, "redis": 1, "rabbitmq": 1} def test_case_id_counts_match_the_manifest(self) -> None: for section, count in self.EXPECTED_COUNTS.items(): @@ -1451,7 +1593,7 @@ def test_known_gap_entries_are_unique(self) -> None: self.assertEqual(sum(len(c) for c in KNOWN_GAPS.values()), len(keys)) def test_there_are_no_known_gaps(self) -> None: - # Every conformance 1.6.0 case is implemented (r1 Faz 08): the gap list stays empty, and a + # Every conformance 1.7.0 case is implemented (r1 Faz 08; conformance 1.7.0, post-R1 Faz 06): the gap list stays empty, and a # new entry is a deliberate, reviewed change to this test as well. self.assertEqual({}, KNOWN_GAPS) @@ -1482,6 +1624,37 @@ def test_kafka_poison_cases(self) -> None: _run_cases(self, "kafka_poison", _eval_kafka_poison) +class IdentityVerdictConformanceTest(unittest.TestCase): + def test_identity_verdict_cases(self) -> None: + _run_cases(self, "identity_verdict", _eval_identity_verdict) + + def test_the_first_cases_are_every_kafka_poison_case_with_no_identity_mapped_to_poison(self) -> None: + verdicts = [case for _, case in iter_cases(MANIFEST, "identity_verdict")] + kafka = [case for _, case in iter_cases(MANIFEST, "kafka_poison")] + self.assertGreaterEqual(len(verdicts), len(kafka)) + for source, derived in zip(kafka, verdicts): + with self.subTest(case=source["name"]): + self.assertEqual(source["name"], derived["name"]) + self.assertEqual(source["body"], derived["body"]) + self.assertEqual("poison" if source["expect"] == "no_identity" else source["expect"], derived["expect"]) + + def test_the_kafka_verdict_agrees_with_the_core_verdict(self) -> None: + for case_id, case in iter_cases(MANIFEST, "kafka_poison"): + with self.subTest(case=case_id): + core = "ok" if EnvelopeCodec.poison_cause(case["body"]) is None else "poison" + self.assertEqual("poison" if case["expect"] != "ok" else "ok", core) + + +class PulsarExhaustionConformanceTest(unittest.TestCase): + def test_pulsar_exhaustion_cases(self) -> None: + _run_cases(self, "pulsar_exhaustion", _eval_pulsar_exhaustion) + + def test_the_strategy_rows_are_not_skipped(self) -> None: + # Python has the on_unknown_urn option: every row runs (a skip would be a pass by omission). + rows = [c for _, c in iter_cases(MANIFEST, "pulsar_exhaustion") if "on_unknown_urn" in c] + self.assertEqual(10, len(rows)) + + class PoisonDlqConformanceTest(unittest.TestCase): def test_poison_dlq_cases(self) -> None: _run_cases(self, "poison_dlq", _eval_poison_dlq) diff --git a/tests/test_conformance.py b/tests/test_conformance.py index 89c9bde..598f86e 100644 --- a/tests/test_conformance.py +++ b/tests/test_conformance.py @@ -101,6 +101,8 @@ def test_cases(self) -> None: "redis_delay": [("test_behaviour_conformance", "RedisDelayConformanceTest", "test_redis_delay_cases")], "kafka_poison": [("test_behaviour_conformance", "KafkaPoisonConformanceTest", "test_kafka_poison_cases")], "poison_dlq": [("test_behaviour_conformance", "PoisonDlqConformanceTest", "test_poison_dlq_cases")], + "identity_verdict": [("test_behaviour_conformance", "IdentityVerdictConformanceTest", "test_identity_verdict_cases")], + "pulsar_exhaustion": [("test_behaviour_conformance", "PulsarExhaustionConformanceTest", "test_pulsar_exhaustion_cases")], # The broker bindings are evidenced per runner, not per case (every method must pass). "sqs": [("test_sqs_conformance", "SqsConformanceTest", m) for m in ( "test_attribute_projection", "test_attempts_reconciliation", "test_schema_version_gate")], diff --git a/tests/test_kafka_retry.py b/tests/test_kafka_retry.py index b82186e..ebb9406 100644 --- a/tests/test_kafka_retry.py +++ b/tests/test_kafka_retry.py @@ -114,6 +114,62 @@ def _boom(data, meta): self.assertEqual(producer.produced, []) self.assertTrue(any("NOT acknowledged" in m for m in logs.output)) + def _tierless_poison(self, bodies, **app_kwargs): + records = [OffsetMessage(raw, [], offset=i) for i, raw in enumerate(bodies)] + consumer = OffsetConsumer(records) + producer = ReportingProducer() + kafka = KafkaTransport(producer=producer, consumer_factory=lambda _q: consumer) # no tiers + app = BabelQueue(transport=kafka, queue="orders", **app_kwargs) + app._rewind_pause = 0.0 # noqa: SLF001 + app.register(URN, lambda data, meta: None) + return app, consumer, producer + + def test_a_poison_record_without_tiers_and_without_a_dlq_is_committed_and_dropped(self): + # ADR-0036 decision 5 / broker-bindings section 6.5 (degrade-to-delete): the bounded retry + # of a poison record needs a tier and used to raise, leaving the record uncommitted with + # ``bq-attempts`` never advancing, so the partition blocked. Now it is committed and + # dropped (logged), and the partition moves on to the next record. + poison = [b"not json", b"[]", b'{"foo":1}', b"\xff\xfe not utf-8", b'{"job":"u","attempts":"x"}'] + for strategy in ("fail", "release"): + with self.subTest(strategy=strategy): + valid = body().encode("utf-8") + app, consumer, producer = self._tierless_poison( + [*poison, valid], max_attempts=3, on_unknown_urn=strategy, + ) + with self.assertLogs("babelqueue.app", level="WARNING") as logs: + self.assertEqual(app.consume(max_messages=len(poison) + 1, handle_signals=False), 6) + self.assertEqual(consumer.seeks, []) # nothing rewound + self.assertEqual(consumer.commits, [0, 1, 2, 3, 4, 5]) # the partition advanced + self.assertEqual(producer.produced, []) # not retried, not dead-lettered + dropped = [m for m in logs.output if "Dropping a poison record" in m] + self.assertEqual(len(dropped), len(poison)) + self.assertTrue(all("degrade-to-delete" in m for m in dropped)) + + def test_a_poison_record_without_tiers_is_still_dead_lettered_when_the_dlq_is_on(self): + app, consumer, producer = self._tierless_poison([b"not json"], max_attempts=3, dead_letter=True) + app.consume(max_messages=1, handle_signals=False) + (record,) = producer.produced + self.assertEqual(record["topic"], "orders.dlq") + self.assertEqual(consumer.commits, [0]) + + def test_a_poison_record_without_tiers_is_dropped_by_the_delete_strategy_too(self): + app, consumer, producer = self._tierless_poison([b"not json"], max_attempts=3, on_unknown_urn="delete") + app.consume(max_messages=1, handle_signals=False) + self.assertEqual(producer.produced, []) + self.assertEqual(consumer.commits, [0]) + + def test_handler_failure_without_tiers_keeps_the_section_6_5_rule(self): + # Only poison degrades to delete: a valid envelope whose handler fails still cannot be + # retried without a tier (covered above); an unknown URN under ``release`` likewise. + consumer = OffsetConsumer([OffsetMessage(body().encode("utf-8").replace(URN.encode(), b"urn:other"), [], offset=0)]) + kafka = KafkaTransport(producer=ReportingProducer(), consumer_factory=lambda _q: consumer) + app = BabelQueue(transport=kafka, queue="orders", max_attempts=3, on_unknown_urn="release") + app._rewind_pause = 0.0 # noqa: SLF001 + with self.assertLogs("babelqueue.app", level="ERROR"): + app.consume(max_messages=1, handle_signals=False) + self.assertEqual(consumer.commits, []) + self.assertEqual(consumer.seeks, [("orders", 0, 0)]) + def test_no_warning_when_tiers_are_configured(self): kafka = _tier_transport(FakeProducer()) self.assertEqual(self._warnings(kafka, max_attempts=3), []) diff --git a/tests/test_poison_routing.py b/tests/test_poison_routing.py new file mode 100644 index 0000000..e715cbd --- /dev/null +++ b/tests/test_poison_routing.py @@ -0,0 +1,511 @@ +"""Poison routing outside Redis and Kafka (ADR-0036 decisions 2, 4 and 6). + +A body that does not decode into an envelope, or names no identity, is **poison**: on the memory +transport, SQS, AMQP, Pulsar, Azure Service Bus and Artemis it is never decoded leniently into an +empty envelope, never handed to a handler and never routed as an unknown URN. It takes the poison +route of Go's ``routePoison``: dead-lettered as the ADR-0035 record (``reason: "poison"``, original +body preserved) when the DLQ is on, dropped by ``on_unknown_urn: delete``, and bounded by +``max_attempts`` when the DLQ is off. No broker and no client library are needed: the clients are +injected fakes. +""" + +from __future__ import annotations + +import base64 +import json +import unittest +from typing import Any, List, Tuple + +from babelqueue import BabelQueue, EnvelopeCodec, UnknownUrnStrategy +from babelqueue.asb_transport import AsbTransport +from babelqueue.artemis_transport import ArtemisTransport +from babelqueue.pika_transport import PikaTransport +from babelqueue.pulsar_transport import PulsarTransport +from babelqueue.sqs_transport import SqsTransport +from babelqueue.transport import InMemoryTransport, ReceivedMessage + +from test_artemis_transport import FakeConnection as ArtemisConnection +from test_artemis_transport import FakeMessage as ArtemisMessage +from test_artemis_transport import FakeReceiver as ArtemisReceiver +from test_asb_transport import FakeClient as AsbClient +from test_asb_transport import FakeReceived as AsbReceived +from test_asb_transport import FakeReceiver as AsbReceiver +from test_pulsar_transport import FakeClient as PulsarClient +from test_pulsar_transport import FakeConsumer as PulsarConsumer +from test_pulsar_transport import FakeMessage as PulsarMessage +from test_pulsar_transport import FakeProducer as PulsarProducer +from test_sqs_transport import FakeSQS + +URN = "urn:babel:orders:created" + +#: Bodies that are poison outside Kafka (conformance ``identity_verdict``): undecodable, not an +#: object, a wrongly typed member, and an object with no identity. +POISON_BODIES = [ + "not json", + "{not json", + "", + "[]", + "null", + "5", + "{}", + '{"foo":1}', + '{"job":5}', + '{"job":" ","attempts":0}', + '{"job":"%s","attempts":"3"}' % URN, +] + + +def valid_body(attempts: int = 0, urn: str = URN) -> str: + env = EnvelopeCodec.make(urn, {"order_id": 1}, queue="orders") + env["attempts"] = attempts + return EnvelopeCodec.encode(env) + + +def assert_poison_record(case: unittest.TestCase, record: str, body: str, queue: str = "orders") -> None: + """``record`` is the ADR-0035 DLQ record of the poison ``body`` (original bytes preserved).""" + parsed = json.loads(record) + block = parsed["dead_letter"] + case.assertEqual("poison", block["reason"]) + case.assertEqual(queue, block["original_queue"]) + if "raw" in parsed: + if parsed.get("raw_encoding") == "base64": + case.assertEqual(body.encode("utf-8", "surrogateescape"), base64.b64decode(parsed["raw"])) + else: + case.assertEqual(body, parsed["raw"]) + else: # spliced strict JSON object: every original member is kept + original = json.loads(body) + case.assertEqual(original, {k: v for k, v in parsed.items() if k != "dead_letter"}) + + +class MemoryPoisonTest(unittest.TestCase): + def app(self, **kwargs: Any) -> Tuple[BabelQueue, List[Any]]: + calls: List[Any] = [] + app = BabelQueue("memory://", queue="orders", max_attempts=3, **kwargs) + app.register(URN, lambda data, meta: calls.append(data)) + return app, calls + + def feed(self, app: BabelQueue, body: str, limit: int = 10) -> int: + """Publish ``body`` and consume until the queue is empty; returns the deliveries.""" + app.transport.publish("orders", body) + deliveries = 0 + while app.transport.size("orders") and deliveries < limit: + app.consume("orders", max_messages=1, timeout=0.01, handle_signals=False) + deliveries += 1 + return deliveries + + def dlq(self, app: BabelQueue) -> List[str]: + return list(app.transport._queues["orders.dlq"]) # noqa: SLF001 + + def test_with_the_dlq_on_poison_is_dead_lettered_at_once_whatever_the_strategy(self) -> None: + for strategy in (UnknownUrnStrategy.FAIL, UnknownUrnStrategy.RELEASE, UnknownUrnStrategy.DEAD_LETTER): + for body in POISON_BODIES: + with self.subTest(strategy=strategy, body=body): + app, calls = self.app(dead_letter=True, on_unknown_urn=strategy) + self.assertEqual(1, self.feed(app, body)) # no retry + self.assertEqual([], calls) + (record,) = self.dlq(app) + assert_poison_record(self, record, body) + + def test_the_dead_letter_strategy_writes_the_record_even_with_dead_lettering_off(self) -> None: + app, calls = self.app(on_unknown_urn=UnknownUrnStrategy.DEAD_LETTER) + self.assertEqual(1, self.feed(app, "not json")) + (record,) = self.dlq(app) + assert_poison_record(self, record, "not json") + self.assertEqual([], calls) + + def test_delete_drops_poison_with_a_log_and_writes_nothing(self) -> None: + for dead_letter in (False, True): + with self.subTest(dead_letter=dead_letter): + app, calls = self.app(dead_letter=dead_letter, on_unknown_urn=UnknownUrnStrategy.DELETE) + with self.assertLogs("babelqueue.app", level="WARNING"): + self.assertEqual(1, self.feed(app, "not json")) + self.assertEqual(0, app.transport.size("orders.dlq")) + self.assertEqual([], calls) + + def test_with_the_dlq_off_poison_is_bounded_by_max_attempts_then_dropped(self) -> None: + for strategy in (UnknownUrnStrategy.FAIL, UnknownUrnStrategy.RELEASE): + for body in ("not json", '{"foo":1}'): + with self.subTest(strategy=strategy, body=body): + app, calls = self.app(on_unknown_urn=strategy) + with self.assertLogs("babelqueue.app", level="WARNING") as logs: + self.assertEqual(3, self.feed(app, body)) + self.assertEqual(0, app.transport.size("orders")) + self.assertEqual(0, app.transport.size("orders.dlq")) + self.assertEqual([], calls) + self.assertTrue(any("Dropping a message" in m for m in logs.output)) + + def test_a_valid_envelope_with_no_handler_is_still_an_unknown_urn_not_poison(self) -> None: + app, _ = self.app(dead_letter=True) + self.assertEqual(3, self.feed(app, valid_body(urn="urn:babel:other"))) + (record,) = self.dlq(app) + self.assertEqual("unknown_urn", json.loads(record)["dead_letter"]["reason"]) + + def test_a_valid_envelope_with_a_handler_is_not_poison(self) -> None: + app, calls = self.app(dead_letter=True) + self.assertEqual(1, self.feed(app, valid_body())) + self.assertEqual([{"order_id": 1}], calls) + self.assertEqual(0, app.transport.size("orders.dlq")) + + def test_a_urn_alias_and_a_blank_job_with_a_urn_are_identities(self) -> None: + for body in ('{"urn":"%s"}' % URN, '{"job":" ","urn":"%s"}' % URN, '{"job":5,"urn":"%s"}' % URN): + with self.subTest(body=body): + app, calls = self.app(dead_letter=True) + self.assertEqual(1, self.feed(app, body)) + self.assertEqual(1, len(calls)) + self.assertEqual(0, app.transport.size("orders.dlq")) + + def test_an_out_of_band_route_exempts_a_delivery_from_the_identity_half_of_the_rule(self) -> None: + # ADR-0036 decision 4, scope: a delivery routed on a transport property (Pulsar bq-job) is + # outside the identity rule; an undecodable body is still poison. + app, calls = self.app(dead_letter=True) + app.dispatch(ReceivedMessage(body='{"attempts":0}', queue="orders", route_urn=URN)) + self.assertEqual(1, len(calls)) # routed by the property, not poison + app.dispatch(ReceivedMessage(body="not json", queue="orders", route_urn=URN)) + self.assertEqual(1, len(calls)) + (record,) = self.dlq(app) + assert_poison_record(self, record, "not json") + + def test_a_failed_dead_letter_write_leaves_the_poison_message_unacknowledged(self) -> None: + class FailingDlq(InMemoryTransport): + def __init__(self) -> None: + super().__init__() + self.acked = 0 + + def publish(self, queue: str, body: str) -> None: + if queue.endswith(".dlq"): + raise RuntimeError("dlq down") + super().publish(queue, body) + + def ack(self, message: ReceivedMessage) -> None: + self.acked += 1 + + transport = FailingDlq() + app = BabelQueue(transport=transport, queue="orders", max_attempts=3, dead_letter=True) + app.register(URN, lambda data, meta: None) + transport.publish("orders", "not json") + with self.assertLogs("babelqueue.app", level="ERROR"): + app.consume("orders", max_messages=1, timeout=0.01, handle_signals=False) + self.assertEqual(0, transport.acked) # never acknowledged, never retried + self.assertEqual(0, transport.size("orders")) # nothing re-published as a retry + + def test_the_core_decoder_is_not_changed_by_the_poison_rule(self) -> None: + self.assertEqual({}, EnvelopeCodec.decode("not json")) + self.assertEqual({}, EnvelopeCodec.decode_lenient("[1]")) + self.assertIsNone(EnvelopeCodec.poison_cause(valid_body())) + self.assertIsNotNone(EnvelopeCodec.poison_cause('{"foo":1}')) + self.assertIsNone(EnvelopeCodec.poison_cause('{"foo":1}', require_identity=False)) + self.assertIsNotNone(EnvelopeCodec.poison_cause("not json", require_identity=False)) + + +class SqsPoisonTest(unittest.TestCase): + URL = "http://fake/orders" + + def setUp(self) -> None: + self.fake = FakeSQS() + self.transport = SqsTransport("sqs://", client=self.fake, queue_url_prefix="http://fake") + + def test_a_poison_message_is_dead_lettered_with_the_original_body_and_deleted(self) -> None: + for body in ("not json", '{"foo":1}'): + with self.subTest(body=body): + self.fake.seed(self.URL, body, 1) + app = BabelQueue(transport=self.transport, queue="orders", max_attempts=3, dead_letter=True) + app.register(URN, lambda data, meta: None) + app.consume("orders", max_messages=1, timeout=0, handle_signals=False) + record = self.fake.sent[-1] + self.assertEqual("http://fake/orders.dlq", record["QueueUrl"]) + assert_poison_record(self, record["MessageBody"], body) + self.assertEqual(1, len(self.fake.deleted)) + self.assertEqual([], self.fake.visibility_changes) # no retry + self.fake.sent.clear() + self.fake.deleted.clear() + + def test_the_native_receive_count_bounds_the_retry_when_the_dlq_is_off(self) -> None: + # SQS keeps the body and counts deliveries itself: the runtime reads the count off the + # delivery (an undecodable body cannot carry ``attempts``) and stops at max_attempts. + self.fake.seed(self.URL, "not json", 1) + app = BabelQueue(transport=self.transport, queue="orders", max_attempts=3) + app.register(URN, lambda data, meta: None) + deliveries = 0 + with self.assertLogs("babelqueue.app", level="WARNING"): + while self.fake.visible.get(self.URL) and deliveries < 10: + app.consume("orders", max_messages=1, timeout=0, handle_signals=False) + deliveries += 1 + self.assertEqual(3, deliveries) + self.assertEqual(2, len(self.fake.visibility_changes)) # retried twice, dropped the third time + self.assertEqual(1, len(self.fake.deleted)) + self.assertEqual([], self.fake.sent) + + def test_the_delivery_carries_the_native_attempts(self) -> None: + self.fake.seed(self.URL, "not json", 4) + message = self.transport.pop("orders", timeout=0) + assert message is not None + self.assertEqual(3, message.native_attempts) + self.assertEqual("not json", message.body) + + +class PulsarPoisonTest(unittest.TestCase): + def make(self, message: Any, **kwargs: Any) -> Tuple[BabelQueue, PulsarClient, PulsarConsumer]: + consumer = PulsarConsumer([message]) + client = PulsarClient(producer=PulsarProducer(), consumer=consumer) + app = BabelQueue( + transport=PulsarTransport("pulsar://localhost:6650", client=client), + queue="orders", + max_attempts=3, + **kwargs, + ) + app.register(URN, lambda data, meta: None) + return app, client, consumer + + def test_a_poison_body_is_dead_lettered_with_reason_poison(self) -> None: + app, client, consumer = self.make(PulsarMessage("not json"), dead_letter=True) + app.consume("orders", max_messages=1, timeout=0, handle_signals=False) + ((content, _),) = client.producer.sent + assert_poison_record(self, content.decode("utf-8"), "not json") + self.assertEqual(1, len(consumer.acked)) + + def test_bytes_that_are_not_utf_8_survive_losslessly_as_base64(self) -> None: + class RawBytes(PulsarMessage): + def data(self) -> bytes: + return b"\xff\xfe{garbage" + + app, client, consumer = self.make(RawBytes(""), dead_letter=True) + app.consume("orders", max_messages=1, timeout=0, handle_signals=False) # no UnicodeDecodeError + ((content, _),) = client.producer.sent + parsed = json.loads(content.decode("utf-8")) + self.assertEqual("base64", parsed["raw_encoding"]) + self.assertEqual(b"\xff\xfe{garbage", base64.b64decode(parsed["raw"])) + self.assertEqual("poison", parsed["dead_letter"]["reason"]) + self.assertEqual(1, len(consumer.acked)) + + def test_the_reconciled_attempts_count_a_poison_body(self) -> None: + # bq-attempts "2" + max_attempts 3, DLQ off: counted 3 -> dropped, not retried. + app, client, consumer = self.make(PulsarMessage("not json", 0, {"bq-attempts": "2"})) + with self.assertLogs("babelqueue.app", level="WARNING"): + app.consume("orders", max_messages=1, timeout=0, handle_signals=False) + self.assertEqual([], client.producer.sent) + self.assertEqual(1, len(consumer.acked)) + + def test_a_failed_dead_letter_write_is_negatively_acknowledged_for_redelivery(self) -> None: + # ADR-0036 decision 1: the original is acknowledged only after the DLQ write succeeded; a + # failed write nacks it (Go ``Rewind`` = ``Nack``) so the broker redelivers it. + class FailingProducer(PulsarProducer): + def send(self, content: Any, properties: Any = None) -> None: + raise RuntimeError("dlq down") + + message = PulsarMessage("not json") + consumer = PulsarConsumer([message]) + client = PulsarClient(producer=FailingProducer(), consumer=consumer) + app = BabelQueue( + transport=PulsarTransport("pulsar://localhost:6650", client=client), + queue="orders", max_attempts=3, dead_letter=True, + ) + app._rewind_pause = 0.0 # noqa: SLF001 + with self.assertLogs("babelqueue.app", level="ERROR"): + app.consume("orders", max_messages=1, timeout=0, handle_signals=False) + self.assertEqual([message], consumer.nacked) + self.assertEqual([], consumer.acked) + + def test_the_transport_is_a_rewinder(self) -> None: + from babelqueue.transport import Rewinder + + self.assertIsInstance(PulsarTransport("pulsar://x", client=PulsarClient()), Rewinder) + + def test_a_retry_re_publishes_a_bare_body_that_carries_the_count(self) -> None: + app, client, consumer = self.make(PulsarMessage("not json", 1)) + app.consume("orders", max_messages=1, timeout=0, handle_signals=False) + ((content, properties),) = client.producer.sent + self.assertEqual({"attempts": 2}, json.loads(content.decode("utf-8"))) + self.assertEqual("2", properties["bq-attempts"]) + self.assertEqual(1, len(consumer.acked)) + + +#: A *valid* envelope whose JSON string ``data.s`` holds the invalid UTF-8 byte ``0xff``: the bytes +#: are not UTF-8, yet ``json.loads`` accepts the lone surrogate they are read as. +INVALID_UTF8_IN_STRING = ( + b'{"job":"' + URN.encode() + b'","trace_id":"t","data":{"s":"\xff"},"meta":{},"attempts":0}' +) + + +def assert_base64_poison(case: unittest.TestCase, record: str, raw: bytes) -> None: + parsed = json.loads(record) + case.assertEqual("poison", parsed["dead_letter"]["reason"]) + case.assertEqual("base64", parsed["raw_encoding"]) + case.assertEqual(raw, base64.b64decode(parsed["raw"])) + + +class InvalidUtf8InsideAJsonStringTest(unittest.TestCase): + """error-handling.md section 7: invalid UTF-8 is poison even when it sits inside a JSON string, + where the lossless read leaves a lone surrogate ``json.loads`` accepts. It never reaches a + handler and its original bytes are dead-lettered as the ADR-0035 base64 record.""" + + def test_the_core_verdict(self) -> None: + text = INVALID_UTF8_IN_STRING.decode("utf-8", "surrogateescape") + self.assertIsNotNone(EnvelopeCodec.poison_cause(text)) + self.assertIsNotNone(EnvelopeCodec.poison_cause(text, require_identity=False)) + # A ``\udcff`` *escape* is plain ASCII in the text and stays a valid envelope. + escaped = '{"job":"%s","data":{"s":"\\udcff"},"attempts":0}' % URN + self.assertIsNone(EnvelopeCodec.poison_cause(escaped)) + + def run_app(self, transport: Any) -> List[str]: + calls: List[str] = [] + app = BabelQueue(transport=transport, queue="orders", max_attempts=3, dead_letter=True) + app.register(URN, lambda data, meta: calls.append("called")) + app.consume("orders", max_messages=1, timeout=0, handle_signals=False) + return calls + + def test_amqp(self) -> None: + transport, channel = AmqpPoisonTest().transport(INVALID_UTF8_IN_STRING) + capture = _PublishCapture(transport) + self.assertEqual([], self.run_app(transport)) + ((queue, record),) = capture.written + self.assertEqual("orders.dlq", queue) + assert_base64_poison(self, record, INVALID_UTF8_IN_STRING) + self.assertEqual([7], channel.acked) + + def test_pulsar(self) -> None: + class RawBytes(PulsarMessage): + def data(self) -> bytes: + return INVALID_UTF8_IN_STRING + + consumer = PulsarConsumer([RawBytes("")]) + client = PulsarClient(producer=PulsarProducer(), consumer=consumer) + self.assertEqual([], self.run_app(PulsarTransport("pulsar://x", client=client))) + ((content, _),) = client.producer.sent + assert_base64_poison(self, content.decode("utf-8"), INVALID_UTF8_IN_STRING) + self.assertEqual(1, len(consumer.acked)) + + def test_artemis(self) -> None: + receiver = ArtemisReceiver([ArtemisMessage(INVALID_UTF8_IN_STRING)]) + transport = ArtemisTransport(connection=ArtemisConnection(receiver=receiver)) + capture = _PublishCapture(transport) + self.assertEqual([], self.run_app(transport)) + ((_, record),) = capture.written + assert_base64_poison(self, record, INVALID_UTF8_IN_STRING) + self.assertEqual(1, receiver.accepted) + + +class _PublishCapture: + """Replaces ``transport.publish`` so a transport whose publish needs a client library (or a + live connection) can still be asked what the runtime wrote.""" + + def __init__(self, transport: Any) -> None: + self.written: List[Tuple[str, str]] = [] + transport.publish = self.publish + + def publish(self, queue: str, body: str) -> None: + self.written.append((queue, body)) + + +class AsbPoisonTest(unittest.TestCase): + def test_a_poison_message_is_dead_lettered_and_completed(self) -> None: + for body in ("not json", '{"foo":1}'): + with self.subTest(body=body): + received = AsbReceived(body) + receiver = AsbReceiver([received]) + transport = AsbTransport(client=AsbClient(receiver=receiver)) + capture = _PublishCapture(transport) + app = BabelQueue(transport=transport, queue="orders", max_attempts=3, dead_letter=True) + app.register(URN, lambda data, meta: None) + app.consume("orders", max_messages=1, timeout=0, handle_signals=False) + ((queue, record),) = capture.written + self.assertEqual("orders.dlq", queue) + assert_poison_record(self, record, body) + self.assertEqual([received], receiver.completed) + + +class ArtemisPoisonTest(unittest.TestCase): + def test_a_poison_message_is_dead_lettered_and_accepted(self) -> None: + for body in ("not json", '{"foo":1}'): + with self.subTest(body=body): + receiver = ArtemisReceiver([ArtemisMessage(body)]) + transport = ArtemisTransport(connection=ArtemisConnection(receiver=receiver)) + capture = _PublishCapture(transport) + app = BabelQueue(transport=transport, queue="orders", max_attempts=3, dead_letter=True) + app.register(URN, lambda data, meta: None) + app.consume("orders", max_messages=1, timeout=0, handle_signals=False) + ((queue, record),) = capture.written + self.assertEqual("orders.dlq", queue) + assert_poison_record(self, record, body) + self.assertEqual(1, receiver.accepted) + + def test_bytes_that_are_not_utf_8_do_not_crash_the_consumer(self) -> None: + receiver = ArtemisReceiver([ArtemisMessage(b"\xff\xfe{garbage")]) + transport = ArtemisTransport(connection=ArtemisConnection(receiver=receiver)) + capture = _PublishCapture(transport) + app = BabelQueue(transport=transport, queue="orders", max_attempts=3, dead_letter=True) + app.consume("orders", max_messages=1, timeout=0, handle_signals=False) + ((_, record),) = capture.written + parsed = json.loads(record) + self.assertEqual("base64", parsed["raw_encoding"]) + self.assertEqual(b"\xff\xfe{garbage", base64.b64decode(parsed["raw"])) + + +class _FakeChannel: + is_closed = False + + def __init__(self, messages: List[Tuple[Any, Any, Any]]) -> None: + self._messages = messages + self.acked: List[Any] = [] + + def queue_declare(self, **kwargs: Any) -> None: + return None + + def basic_get(self, queue: str, auto_ack: bool = False) -> Tuple[Any, Any, Any]: + return self._messages.pop(0) if self._messages else (None, None, None) + + def basic_ack(self, delivery_tag: Any) -> None: + self.acked.append(delivery_tag) + + +class _Method: + delivery_tag = 7 + + +class AmqpPoisonTest(unittest.TestCase): + def transport(self, body: bytes) -> Tuple[PikaTransport, _FakeChannel]: + channel = _FakeChannel([(_Method(), None, body)]) + transport = PikaTransport.__new__(PikaTransport) + transport._pika = None + transport._url = "amqp://x" + transport._declared = set() + transport._connection = type("C", (), {"is_closed": False, "sleep": lambda self, t: None})() + transport._channel = channel + return transport, channel + + def test_a_poison_message_is_dead_lettered_and_acked(self) -> None: + transport, channel = self.transport(b"not json") + capture = _PublishCapture(transport) + app = BabelQueue(transport=transport, queue="orders", max_attempts=3, dead_letter=True) + app.register(URN, lambda data, meta: None) + app.consume("orders", max_messages=1, timeout=0, handle_signals=False) + ((queue, record),) = capture.written + self.assertEqual("orders.dlq", queue) + assert_poison_record(self, record, "not json") + self.assertEqual([7], channel.acked) + + def test_bytes_that_are_not_utf_8_do_not_crash_the_consumer(self) -> None: + transport, channel = self.transport(b"\xff\xfe{garbage") + capture = _PublishCapture(transport) + app = BabelQueue(transport=transport, queue="orders", max_attempts=3, dead_letter=True) + app.consume("orders", max_messages=1, timeout=0, handle_signals=False) + ((_, record),) = capture.written + parsed = json.loads(record) + self.assertEqual("base64", parsed["raw_encoding"]) + self.assertEqual(b"\xff\xfe{garbage", base64.b64decode(parsed["raw"])) + self.assertEqual([7], channel.acked) + + +class TransportsThatClassifyPoisonThemselvesTest(unittest.TestCase): + def test_only_kafka_and_redis_opt_out_of_the_runtime_classification(self) -> None: + from babelqueue.kafka_transport import KafkaTransport + from babelqueue.redis_transport import RedisTransport + + self.assertTrue(KafkaTransport.classifies_poison) + self.assertTrue(RedisTransport.classifies_poison) + for cls in (InMemoryTransport, SqsTransport, PikaTransport, PulsarTransport, AsbTransport, ArtemisTransport): + with self.subTest(transport=cls.__name__): + self.assertFalse(cls.classifies_poison) + + +if __name__ == "__main__": # pragma: no cover + unittest.main() diff --git a/tests/test_pulsar_transport.py b/tests/test_pulsar_transport.py index 2debdff..f85de7d 100644 --- a/tests/test_pulsar_transport.py +++ b/tests/test_pulsar_transport.py @@ -86,6 +86,7 @@ class FakeConsumer: def __init__(self, messages=()) -> None: self._messages = list(messages) self.acked: list = [] + self.nacked: list = [] self.subscribe_kwargs: dict = {} def receive(self, timeout_millis=None): @@ -96,6 +97,9 @@ def receive(self, timeout_millis=None): def acknowledge(self, message) -> None: self.acked.append(message) + def negative_acknowledge(self, message) -> None: + self.nacked.append(message) + class FakeClient: def __init__(self, producer=None, consumer=None) -> None: diff --git a/tests/test_redis_transport.py b/tests/test_redis_transport.py index c172a4f..a474a89 100644 --- a/tests/test_redis_transport.py +++ b/tests/test_redis_transport.py @@ -451,17 +451,50 @@ def test_a_body_go_typed_decode_rejects_is_poison_on_the_first_delivery(self): ((_, args),) = client.settles assert_poison_record(self, args[5], garbage, "orders") # never handled or retried - def test_a_bare_null_is_not_poison(self): + def test_a_bare_null_is_poison_it_has_no_identity(self): + # ADR-0036 decision 4: a body that decodes but names no identity is poison (conformance + # ``identity_verdict``: ``null`` is ``poison``), no longer the bounded FAIL path. client = self.Client("null") message = RedisTransport("redis://", client=client).pop("orders", timeout=0) - self.assertEqual(message.route_urn, "") + self.assertEqual(message.route_urn, NON_NUMERIC_ATTEMPTS_ROUTE) + + def test_an_identity_less_object_is_poison(self): + # ADR-0036 decision 4 / broker-bindings section 1.1 "Identity-less body": an object with + # neither a valid ``job`` nor a valid ``urn`` is dead-lettered as poison, never handled. + for body in ('{"attempts":0}', "{}", '{"job":" ","attempts":0}', '{"job":5,"urn":null}'): + with self.subTest(body=body): + client = self.Client(body) + message = RedisTransport("redis://", client=client).pop("orders", timeout=0) + self.assertEqual(message.route_urn, NON_NUMERIC_ATTEMPTS_ROUTE) - def test_an_identity_less_object_is_not_poison(self): - # {} / an object without job or urn still takes the bounded FAIL path, not the poison DLQ. - client = self.Client('{"attempts":0}') + def test_a_urn_alias_is_an_identity_not_poison(self): + client = self.Client('{"urn":"urn:a","attempts":0}') message = RedisTransport("redis://", client=client).pop("orders", timeout=0) self.assertEqual(message.route_urn, "") + def test_an_identity_less_object_is_dead_lettered_in_every_configuration(self): + for kwargs in ( + {}, # dead-lettering off (the default) + {"dead_letter": True, "on_unknown_urn": UnknownUrnStrategy.DELETE}, + {"on_unknown_urn": UnknownUrnStrategy.DELETE}, + {"on_unknown_urn": UnknownUrnStrategy.RELEASE}, + ): + with self.subTest(**{k: str(v) for k, v in kwargs.items()}): + body = '{"foo":1,"attempts":0}' + handled = [] + client = self.Client(body) + app = BabelQueue( + transport=RedisTransport("redis://", client=client, retry_after=90), + queue="orders", max_attempts=3, **kwargs, + ) + app.register(URN, lambda data, meta: handled.append(1)) + self.assertEqual(app.consume("orders", max_messages=1, timeout=0), 1) + ((_, args),) = client.settles + self.assertEqual(args[2], "orders.dlq") + assert_poison_record(self, args[5], body, "orders") + self.assertEqual(handled, []) + self.assertEqual(client.stored, []) # never retried + def test_a_valid_envelope_is_not_flagged(self): client = self.Client(make_body("orders")) message = RedisTransport("redis://", client=client).pop("orders", timeout=0) @@ -792,6 +825,34 @@ def test_poison_is_dead_lettered_even_with_the_delete_unknown_urn_strategy(self) self.assert_dlq_poison_records(poison) self.assertEqual(self.client.zcard(self.reserved), 0) + def test_identity_less_body_is_dead_lettered_in_every_configuration_and_never_handled(self): + # ADR-0036 decision 4: the zset layout too. The reserved member (attempts bumped when the + # body has a numeric ``attempts``) is the ack handle; the DLQ record keeps the raw job. + bodies = ['{"foo":1,"attempts":0}', "{}", "null", '{"job":" ","attempts":0}', '{"foo":1}'] + configs = ( + {"dead_letter": False}, + {"dead_letter": False, "on_unknown_urn": UnknownUrnStrategy.DELETE}, + {"dead_letter": True, "on_unknown_urn": UnknownUrnStrategy.DELETE}, + {"dead_letter": False, "on_unknown_urn": UnknownUrnStrategy.RELEASE}, + ) + for kwargs in configs: + for body in bodies: + with self.subTest(body=body, **{k: str(v) for k, v in kwargs.items()}): + handled: list = [] + self.tr.publish(self.queue, body) + app = self.app(max_attempts=3, **kwargs) + app.register(URN, lambda data, meta: handled.append(1)) + message = self.tr.pop(self.queue, timeout=0) + self.assertEqual(message.route_urn, NON_NUMERIC_ATTEMPTS_ROUTE) + self.assertEqual(message.body, body) # the raw job, not the reserved member + app.dispatch(message) + (record,) = self.client.lrange(f"{self.ready}.dlq", 0, -1) + assert_poison_record(self, record, body, self.queue) + self.assertEqual(handled, []) + for key in (self.ready, self.reserved, self.delayed): + self.assertEqual(self.client.exists(key), 0, key) # never retried + self.client.delete(f"{self.ready}.dlq", f"{self.ready}.dlq:notify", self.notify) + def test_a_genuine_unknown_urn_is_still_deleted_with_the_delete_strategy(self): self.tr.publish(self.queue, make_body(self.queue)) app = self.app(dead_letter=True, on_unknown_urn=UnknownUrnStrategy.DELETE) @@ -1123,6 +1184,32 @@ def test_undecodable_body_is_dead_lettered_with_the_dlq_off_too(self): assert_poison_record(self, record, "[1]", self.queue) self.assertEqual(self.client.llen(self.processing), 0) + def test_identity_less_body_is_dead_lettered_in_every_configuration_and_never_handled(self): + # ADR-0036 decision 4 / broker-bindings section 1.1 "Identity-less body": unconditional, + # so ``on_unknown_urn: delete`` and dead-lettering off change nothing, and no retry. + bodies = ['{"foo":1,"attempts":0}', "{}", "null", '{"job":" ","attempts":0}'] + configs = ( + {"dead_letter": False}, + {"dead_letter": False, "on_unknown_urn": UnknownUrnStrategy.DELETE}, + {"dead_letter": True, "on_unknown_urn": UnknownUrnStrategy.DELETE}, + {"dead_letter": False, "on_unknown_urn": UnknownUrnStrategy.RELEASE}, + ) + for kwargs in configs: + for body in bodies: + with self.subTest(body=body, **{k: str(v) for k, v in kwargs.items()}): + handled: list = [] + self.client.rpush(self.queue, body) + app = self.app(max_attempts=3, **kwargs) + app.register(URN, lambda data, meta: handled.append(1)) + self.assertEqual(app.consume(self.queue, max_messages=1, timeout=0), 1) + (record,) = self.client.lrange(f"{self.queue}.dlq", 0, -1) + assert_poison_record(self, record, body, self.queue) + self.assertEqual(handled, []) + self.assertEqual(self.client.llen(self.queue), 0) # never retried + self.assertEqual(self.client.llen(self.processing), 0) + self.assertEqual(self.client.zcard(self.deadlines), 0) + self.client.delete(f"{self.queue}.dlq") + def test_publish_with_headers_stores_a_frame_and_pop_unframes(self): body = make_body(self.queue) self.tr.publish_with_headers(self.queue, body, {"traceparent": "00-deadbeef"}) From 6daed5d89d91ba3f3e0606a7ce506039c823e61b Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Muhammet=20=C5=9Eafak?= Date: Sun, 25 Oct 2026 10:15:16 +0300 Subject: [PATCH 2/2] Release v1.23.0 --- CHANGELOG.md | 2 ++ pyproject.toml | 2 +- src/babelqueue/__init__.py | 2 +- 3 files changed, 4 insertions(+), 2 deletions(-) diff --git a/CHANGELOG.md b/CHANGELOG.md index f02d5a3..9c1e52c 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -9,6 +9,8 @@ The envelope wire format is versioned separately by `meta.schema_version` ## [Unreleased] +## [1.23.0] - 2026-10-25 + Implements ADR-0036 (post-R1 parity) decisions 2, 4, 5 and 6. A MINOR: every change is a consumer-side settle decision, so the envelope stays at `schema_version: 1`. The vendored conformance suite moves to v1.7.0 (new sections `identity_verdict` and `pulsar_exhaustion`, both diff --git a/pyproject.toml b/pyproject.toml index ccf3379..842b4f5 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -4,7 +4,7 @@ build-backend = "hatchling.build" [project] name = "babelqueue" -version = "1.22.0" +version = "1.23.0" description = "Polyglot Queues, Simplified — the Python core: the canonical BabelQueue wire-envelope codec, contracts and dead-letter helpers." readme = "README.md" requires-python = ">=3.9" diff --git a/src/babelqueue/__init__.py b/src/babelqueue/__init__.py index 6760a15..8e6ad41 100644 --- a/src/babelqueue/__init__.py +++ b/src/babelqueue/__init__.py @@ -42,7 +42,7 @@ Transport, ) -__version__ = "1.22.0" +__version__ = "1.23.0" __all__ = [ "BabelQueue",