From e6e272caabeb6b6606e379ddb4679400a5ce1878 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Muhammet=20=C5=9Eafak?= Date: Sun, 25 Oct 2026 17:54:05 +0300 Subject: [PATCH 1/3] Leave SQS and ASB poison to the native mechanism when dead-lettering is off (ADR-0037) With dead-lettering off, version-gated and poison bodies on SQS and Azure Service Bus are no longer dropped; they are left to RedrivePolicy or MaxDeliveryCount. AsbTransport gains a narrow Abandoner used only when a version-gate DLQ write fails. Pin conformance v1.8.0 and run the version_gate sections. --- .github/workflows/ci.yml | 2 +- .github/workflows/release.yml | 4 +- CHANGELOG.md | 49 ++ README.md | 16 +- src/babelqueue/__init__.py | 4 + src/babelqueue/app.py | 82 +- src/babelqueue/asb_transport.py | 37 + src/babelqueue/sqs_transport.py | 16 + src/babelqueue/transport.py | 38 + tests/conformance/CONFORMANCE_VERSION | 2 +- tests/conformance/manifest.json | 706 +++++++++++++++++- tests/conformance/schema/results.example.json | 2 +- tests/test_app.py | 114 +++ tests/test_asb_conformance.py | 95 ++- tests/test_asb_transport.py | 149 +++- tests/test_behaviour_conformance.py | 112 ++- tests/test_conformance.py | 4 + tests/test_poison_routing.py | 98 ++- tests/test_sqs_conformance.py | 77 +- tests/test_sqs_transport.py | 47 +- 20 files changed, 1608 insertions(+), 46 deletions(-) diff --git a/.github/workflows/ci.yml b/.github/workflows/ci.yml index f140d54..7c78d9f 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.7.0 + BQ_CONFORMANCE_VERSION: 1.8.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 0e46e36..f64df0c 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.7.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.7.0 + BQ_CONFORMANCE_VERSION: 1.8.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 9c1e52c..974e506 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -9,6 +9,55 @@ The envelope wire format is versioned separately by `meta.schema_version` ## [Unreleased] +Implements conformance **1.8.0** (vendored from the `v1.8.0` release asset, verified with +`fetch.sh --check`; `BQ_CONFORMANCE_VERSION` is now `1.8.0` in `ci.yml` and `release.yml`). The +envelope stays at `schema_version: 1`; nothing here touches the wire (ADR-0037). The new +`version_gate_pulsar`, `version_gate_artemis`, `version_gate_sqs` and `version_gate_asb` sections run +and pass (17 cases each; Python has the `on_unknown_urn` option, so no row is skipped). A MINOR; the +version number is set at release time. + +### Changed (behaviour) +- **With the DLQ off, SQS and Azure Service Bus no longer drop a version-gated message or retry a + poison body; they are left to the broker's native mechanism (ADR-0037 decision 1 rule 3 and + decision 2 "Go and Python"; broker-bindings 3.5, 4.5, 5.7; owner decision 2026-10-25).** + - *Version-gated message* (an unsupported `bq-schema-version`), `dead_letter` off and a strategy + other than `DELETE` or `DEAD_LETTER`: it used to be acknowledged and dropped. It is now **not** + acknowledged. SQS: no `DeleteMessage`; visibility expiry redelivers it (the gate rejects it + again) and the queue's `RedrivePolicy` moves it to the native DLQ after `maxReceiveCount`. Azure + Service Bus: an explicit native `DeadLetter` (reason `unsupported schema version`) into + `$DeadLetterQueue` instead of `Complete`. + - *Poison body* (invalid JSON or an identity-less object), `dead_letter` off and a strategy other + than `DELETE` or `DEAD_LETTER`: it used to take the bounded `FAIL` path (re-publish an empty + envelope up to `max_attempts`, then drop). On SQS and Azure Service Bus it is now settled + natively **on the first delivery** exactly as above (Azure Service Bus reason `poison`): no + re-publish, no retry, no ack, body unchanged, the event is logged. Every other transport + (Pulsar, Artemis, AMQP, Kafka, Redis, memory) keeps its behaviour; `DELETE` (ack and drop), + `dead_letter=True` and the `DEAD_LETTER` strategy (the raw record, then ack) are unchanged on + SQS and Azure Service Bus too. + - A deployment that relied on SQS/Azure Service Bus silently dropping these messages now needs a + `RedrivePolicy` (SQS) or accepts them in `$DeadLetterQueue` (Azure Service Bus), or sets + `dead_letter=True` / `on_unknown_urn=DELETE`. + +### Added +- **`babelqueue.transport.NativeSettler`**, an optional `Transport` capability + (`settle_native(message, reason, description)`). With the DLQ off the runtime calls it instead of + acknowledging a version-gated message (and, on the poison route, instead of the bounded retry) on + a transport that offers it; the decision follows the capability, never the transport's name. + `SqsTransport.settle_native` is a deliberate no-op on the wire; `AsbTransport.settle_native` + dead-letters natively (falling back to `abandon_message` for a receiver without + `dead_letter_message`). Transports without it are unchanged. +- **`babelqueue.transport.Abandoner` and `AsbTransport.abandon`.** When the dead-letter write of a + *version-gated* message fails, Azure Service Bus now `Abandon`s the peek-lock (broker-bindings 4.5, + "A failed write `Abandon`s it") so the broker redelivers it at once, instead of leaving it locked until + the lock lapses (up to 5 minutes). Only that branch abandons (owner decision 2026-10-25): `AsbTransport` + is deliberately **not** a `Rewinder`, so a graceful-shutdown release and a failed retry re-publish keep + their previous handling (a copy is re-published and the original completed), and a failed dead-letter + write of a poison body is still left to its lock. Tests pin all three. The `version_gate_asb` runner + asserts the `Abandon` on its `unsettled` rows. +- Conformance `version_gate_*` runners (`version_gate_pulsar` / `version_gate_artemis` through the + runtime, `version_gate_sqs` / `version_gate_asb` through the real transports) and their entries in + the results writer. + ## [1.23.0] - 2026-10-25 Implements ADR-0036 (post-R1 parity) decisions 2, 4, 5 and 6. A MINOR: every change is a diff --git a/README.md b/README.md index 4632522..059656f 100644 --- a/README.md +++ b/README.md @@ -158,8 +158,13 @@ untrimmed; Azure Service Bus and Artemis also accept the integer `1`). A message anything else is not decoded: it reaches the runtime on the reserved route `babelqueue:unsupported-schema-version:` (the version percent-escaped) and is **never retried** — it goes straight to the dead-letter queue (the incoming body, byte-identical, -with no `dead_letter` block) when `dead_letter` is enabled, and is dropped otherwise, whatever `max_attempts` is. -It does not depend on a broker `RedrivePolicy`. If the dead-letter write fails the message is +with no `dead_letter` block) when `dead_letter` is enabled (or `on_unknown_urn=DEAD_LETTER`), and +`on_unknown_urn=DELETE` drops it, whatever `max_attempts` is. With `dead_letter` off, **SQS and Azure +Service Bus no longer drop it** (ADR-0037): it is left to the broker's native mechanism and logged +(SQS: no `DeleteMessage`, so visibility expiry redelivers it — the gate rejects it again — and the +queue's `RedrivePolicy` moves it to the native DLQ after `maxReceiveCount`; Azure Service Bus: an +explicit native `DeadLetter` with reason `unsupported schema version` into `$DeadLetterQueue` +instead of `Complete`). Kafka, Artemis and Pulsar still acknowledge and drop it. If the dead-letter write fails the message is left unacknowledged (an error is logged), is never retried and never reaches a handler; the dead-letter write is attempted again when the broker brings it back, which depends on the broker: @@ -199,6 +204,13 @@ a warning. A **per-message delay** at publish time is standard queue; a larger value, or any per-message delay on a FIFO queue, raises `BabelQueueError` instead of being truncated or dropped). +**Poison bodies.** A body that does not decode, or has no identity, is never retried by the SDK +(ADR-0037): `dead_letter=True` writes the ADR-0035 record to `.dlq` and then deletes it, +`on_unknown_urn=DELETE` deletes it, and with the DLQ off it is **not deleted and not re-sent**: the +transport's `settle_native` capability leaves it to SQS on the first delivery (visibility expiry, +then the queue's `RedrivePolicy`; the event is logged). Azure Service Bus does the same with a native +`DeadLetter` (reason `poison`). Every other transport keeps the bounded retry up to `max_attempts`. + **Poison-loop risk:** a message that always fails is redelivered at once until `max_attempts` is reached. If the receive count is unavailable the SDK cannot advance attempts at all. **Configure a native `RedrivePolicy`** diff --git a/src/babelqueue/__init__.py b/src/babelqueue/__init__.py index 8e6ad41..b162189 100644 --- a/src/babelqueue/__init__.py +++ b/src/babelqueue/__init__.py @@ -30,10 +30,12 @@ from .replay import HEADER_REPLAY_BYPASS, bypass_external_effects, is_replay from .routing import UnknownUrnStrategy from .transport import ( + Abandoner, BatchPopper, DeadLetterWriter, HeaderPublisher, InMemoryTransport, + NativeSettler, ReceivedMessage, Redeliverer, Rewinder, @@ -55,9 +57,11 @@ "Transport", "InMemoryTransport", "ReceivedMessage", + "Abandoner", "BatchPopper", "DeadLetterWriter", "HeaderPublisher", + "NativeSettler", "Redeliverer", "Rewinder", "RoutedPublisher", diff --git a/src/babelqueue/app.py b/src/babelqueue/app.py index 178a7f9..cf255fa 100644 --- a/src/babelqueue/app.py +++ b/src/babelqueue/app.py @@ -43,8 +43,10 @@ def on_order_created(data, meta): from .routing import UnknownUrnStrategy from .transport import ( NON_NUMERIC_ATTEMPTS_ROUTE, + Abandoner, DeadLetterWriter, HeaderPublisher, + NativeSettler, PoisonRoute, ReceivedMessage, Redeliverer, @@ -383,7 +385,10 @@ def dispatch(self, received: ReceivedMessage) -> None: received.queue, exc_info=exc, ) - self._rewind_unsettled(received) + if received.route_urn.startswith(_UNSUPPORTED_SCHEMA_ROUTE_PREFIX): + self._abandon_unsettled(received) + else: + self._rewind_unsettled(received) return self._retry_or_dead_letter(received, envelope, exc) else: @@ -528,6 +533,49 @@ def _ack_settled(self, received: ReceivedMessage) -> None: exc_info=True, ) + def _settle_native(self, received: ReceivedMessage, reason: str, description: str) -> bool: + """Hand ``received`` to the transport's :class:`~babelqueue.transport.NativeSettler` + capability and report whether it did: ``False`` means the transport has none and the caller + keeps its previous settle. The event is logged and never retried by the SDK; a failing call + is logged too and leaves the message unacknowledged (the broker redelivers it).""" + if not isinstance(self.transport, NativeSettler): + return False + logger.warning( + "Left to the broker's native mechanism (not acknowledged, not retried): queue %r, " + "route %r, reason %s.", + received.queue, received.route_urn, reason, + ) + try: + self.transport.settle_native(received, reason, description) + except Exception: # noqa: BLE001 - reported, never kills the loop + logger.error( + "Native settle of a message on queue %r failed; it stays unacknowledged and will " + "be redelivered by the broker.", + received.queue, + exc_info=True, + ) + return True + + def _abandon_unsettled(self, received: ReceivedMessage) -> None: + """A version-gated delivery whose dead-letter write failed: a :class:`Rewinder` transport is + rewound (:meth:`_rewind_unsettled`); an :class:`~babelqueue.transport.Abandoner` transport + (Azure Service Bus) abandons its lock so the broker redelivers it at once, after the same + short pause. Any other transport leaves it to its lock or visibility timeout.""" + if isinstance(self.transport, Rewinder) or not isinstance(self.transport, Abandoner): + self._rewind_unsettled(received) + return + try: + self.transport.abandon(received) + except Exception: # noqa: BLE001 - reported, never kills the loop + logger.error( + "Failed to abandon an unsettled version-gated message (route %r) on queue %r.", + received.route_urn, + received.queue, + exc_info=True, + ) + return + self._stop.wait(self._rewind_pause) + def _rewind_unsettled(self, received: ReceivedMessage) -> None: """Make a deliberately unsettled message come back (cumulative-offset brokers such as Kafka would otherwise lose it when a later record on the partition is committed), after a short @@ -717,10 +765,15 @@ def _route_body_poison(self, received: ReceivedMessage, cause: str) -> None: * 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. + * with the DLQ off, on a transport with a native mechanism (SQS, Azure Service Bus: the + :class:`~babelqueue.transport.NativeSettler` capability) it is settled natively on the first + delivery -- not acknowledged, not re-published, not retried, body unchanged (ADR-0037 + decision 2); + * with the DLQ off on any other transport 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: @@ -734,6 +787,11 @@ def _route_body_poison(self, received: ReceivedMessage, cause: str) -> None: if self.dead_letter_enabled or self.on_unknown_urn == UnknownUrnStrategy.DEAD_LETTER: self._settle_transport_poison(received, cause, attempts) return + # DLQ off, strategy neither delete nor a writable dead_letter: a transport with a native + # mechanism (SQS, Azure Service Bus) settles the poison body natively on the first delivery + # (ADR-0037 decision 2, "Go and Python") instead of the bounded retry below. + if self._settle_native(received, "poison", "the body does not decode into a valid envelope: " + cause): + return self._retry_or_dead_letter( received, {"attempts": attempts}, @@ -783,7 +841,9 @@ def _dead_letter_target(self, received: ReceivedMessage) -> str: return self.dead_letter_queue or (received.queue + self.dead_letter_suffix) def _settle_version_gated(self, received: ReceivedMessage, always: bool = False) -> None: - """Dead-letter a version-gated delivery directly (no retry); drop it when no DLQ is set. + """Dead-letter a version-gated delivery directly (no retry). With no DLQ set it is left to the + transport's native mechanism when the transport has the :class:`~babelqueue.transport.NativeSettler` + capability (SQS, Azure Service Bus; ADR-0037 decision 1 rule 3) and dropped otherwise. ``always`` writes the DLQ record in every configuration, ignoring ``dead_letter_enabled``. A Kafka poison record reaches it with the DLQ already on (:meth:`_route_poison`). The Redis @@ -798,7 +858,15 @@ def _settle_version_gated(self, received: ReceivedMessage, always: bool = False) if always or self.dead_letter_enabled or self.on_unknown_urn == UnknownUrnStrategy.DEAD_LETTER: self._write_dead_letter(received, self._dead_letter_target(received), received.body) return - # No DLQ to hold it: the drop is deliberate, but it must not be silent. + # No DLQ to hold it (broker-bindings §5.7 rule 3). A transport with a native mechanism (SQS, + # Azure Service Bus) leaves it to that mechanism instead of dropping it (ADR-0037 decision 1). + if self._settle_native( + received, + "unsupported schema version", + "the message declares a schema version this consumer does not support", + ): + return + # Any other transport: the drop is deliberate, but it must not be silent. logger.warning( "Dropping an undeliverable message on queue %r (route %r, %d body characters): it can " "never be handled and no dead-letter queue is configured.", diff --git a/src/babelqueue/asb_transport.py b/src/babelqueue/asb_transport.py index 6ab259b..ff17d53 100644 --- a/src/babelqueue/asb_transport.py +++ b/src/babelqueue/asb_transport.py @@ -12,6 +12,10 @@ ``DeliveryCount`` (1-based), reconciled onto the envelope as ``attempts = DeliveryCount - 1``. A message left un-acked has its lock expire and is redelivered (at-least-once). +A version-gated message (``bq-schema-version`` the consumer does not support) or a poison body is, with +dead-lettering off, not completed: :meth:`AsbTransport.settle_native` dead-letters it natively into +``$DeadLetterQueue`` (ADR-0037). + This implements §4 of the broker-bindings contract. The envelope is unchanged (``schema_version`` stays 1); Azure Service Bus is purely additive. @@ -269,6 +273,39 @@ def ack(self, message: ReceivedMessage) -> None: return self._receiver(message.queue).complete_message(message.handle) + def abandon(self, message: ReceivedMessage) -> None: + """:class:`~babelqueue.transport.Abandoner`: abandon the peek-lock on ``message`` so the + broker redelivers it right away (``AbandonMessage``) instead of when the lock lapses. + Nothing is completed or dead-lettered and the body is untouched. The runtime calls it only + when the dead-letter write of a version-gated message failed (broker-bindings 4.5, "A failed + write ``Abandon``s it"). This transport is deliberately **not** a + :class:`~babelqueue.transport.Rewinder`, so a graceful-shutdown release and a failed retry + re-publish keep their previous handling (copy re-publish and ``Complete``). A message + without a handle, or a receiver without ``abandon_message`` (a test fake), leaves the lock to + lapse.""" + if message.handle is None: + return + abandon = getattr(self._receiver(message.queue), "abandon_message", None) + if abandon is not None: + abandon(message.handle) + + def settle_native(self, message: ReceivedMessage, reason: str, description: str) -> None: + """Dead-letter ``message`` natively (:class:`~babelqueue.transport.NativeSettler`, ADR-0037 + decisions 1 and 2): a version-gated delivery or a poison body that the runtime may neither + process nor drop, with dead-lettering off, is moved by an explicit ``DeadLetter`` (reason + ``"unsupported schema version"`` or ``"poison"``, plus the description) into the entity's + ``$DeadLetterQueue``. Nothing is completed, re-published or retried, and the body is + unchanged. A receiver without ``dead_letter_message`` abandons the lock instead (also a + conformant native settle: ``MaxDeliveryCount`` takes over).""" + if message.handle is None: + return + receiver = self._receiver(message.queue) + dead_letter = getattr(receiver, "dead_letter_message", None) + if dead_letter is not None: + dead_letter(message.handle, reason=reason, error_description=description) + return + self.abandon(message) + def close(self) -> None: # pragma: no cover - resource cleanup for resource in (*self._senders.values(), *self._receivers.values(), self._client): try: diff --git a/src/babelqueue/sqs_transport.py b/src/babelqueue/sqs_transport.py index d154fcd..aec9227 100644 --- a/src/babelqueue/sqs_transport.py +++ b/src/babelqueue/sqs_transport.py @@ -25,6 +25,10 @@ non-``bq-`` attributes back onto :attr:`~babelqueue.transport.ReceivedMessage.headers`. The frozen envelope is untouched (GR-1). +A version-gated message (``bq-schema-version`` the consumer does not support) or a poison body is, with +dead-lettering off, not deleted: :meth:`SqsTransport.settle_native` leaves it to SQS (visibility +expiry, then the queue's ``RedrivePolicy``; ADR-0037). + This implements §3 of the broker-bindings contract. The envelope is unchanged (``schema_version`` stays 1); SQS is purely additive. @@ -365,6 +369,18 @@ def ack(self, message: ReceivedMessage) -> None: QueueUrl=self._resolve_url(message.queue), ReceiptHandle=message.handle ) + def settle_native(self, message: ReceivedMessage, reason: str, description: str) -> None: + """Leave ``message`` to SQS itself (:class:`~babelqueue.transport.NativeSettler`, ADR-0037 + decisions 1 and 2): a version-gated delivery or a poison body that the runtime may neither + process nor drop, with dead-lettering off. Deliberately does nothing on the wire: no + ``DeleteMessage``, no ``ChangeMessageVisibility``, no re-send. The message becomes visible + again when its visibility timeout lapses (the gate or the poison check rejects it again) and + the queue's ``RedrivePolicy`` moves it to the native DLQ once ``maxReceiveCount`` is reached; + without a ``RedrivePolicy`` it is redelivered until the retention period ends. The runtime + has already logged the event.""" + if message.handle: + self._uncounted_handles.discard(message.handle) + def redeliver(self, message: ReceivedMessage, body: str, delay: float) -> None: """Release ``message`` for redelivery after ``delay`` seconds (broker-bindings.md §3.5; :class:`~babelqueue.transport.Redeliverer`). diff --git a/src/babelqueue/transport.py b/src/babelqueue/transport.py index 08cc849..c3c0a34 100644 --- a/src/babelqueue/transport.py +++ b/src/babelqueue/transport.py @@ -229,6 +229,44 @@ def rewind(self, message: ReceivedMessage) -> None: ... +@runtime_checkable +class Abandoner(Protocol): + """Optional :class:`Transport` capability: hand an unsettled *version-gated* delivery back to the + broker at once (Azure Service Bus ``Abandon``; broker-bindings.md section 4.5 "A failed write + ``Abandon``s it"; ADR-0037 decision 1 rule 5). + + The runtime calls :meth:`abandon` only when the dead-letter write of a version-gated delivery + failed, instead of leaving it locked until the lock lapses. It is deliberately narrower than + :class:`Rewinder`: a transport that offers only this capability keeps its previous graceful + shutdown release and retry re-publish failure handling (copy re-publish and ack).""" + + def abandon(self, message: ReceivedMessage) -> None: + """Release the broker lock on ``message`` so the broker redelivers it right away.""" + ... + + +@runtime_checkable +class NativeSettler(Protocol): + """Optional :class:`Transport` capability: leave a delivery the runtime can neither process nor + safely drop to the broker's own mechanism, without acknowledging or deleting it and without + re-publishing or retrying it (ADR-0037 decisions 1 and 2). + + The runtime calls :meth:`settle_native` instead of acknowledging when dead-lettering is off (and + the strategy is neither ``delete`` nor a writable ``dead_letter``) for a version-gated delivery + (broker-bindings.md section 5.7 rule 3) and for a poison body (error-handling.md section 7). + SQS implements it as "do nothing" (no ``DeleteMessage``: visibility expiry redelivers the message + and the queue's ``RedrivePolicy`` moves it to the native DLQ after ``maxReceiveCount``); Azure + Service Bus implements it with an explicit native ``DeadLetter(reason, description)`` into + ``$DeadLetterQueue``. The body is never altered. A transport that does not implement it keeps + the previous settle (acknowledge and drop), so the behaviour follows the capability, never the + transport's name.""" + + def settle_native(self, message: ReceivedMessage, reason: str, description: str) -> None: + """Leave ``message`` to the broker's native mechanism; ``reason`` and ``description`` say + why (e.g. ``"unsupported schema version"``, ``"poison"``).""" + ... + + #: Reserved ``route_urn`` of a record whose envelope carries a non-numeric ``attempts`` (a foreign #: producer). It sits under the version-gated prefix, so the runtime dead-letters it directly, raw #: and never retried (retrying needs a numeric attempt counter); ``redrive`` restores it to the DLQ diff --git a/tests/conformance/CONFORMANCE_VERSION b/tests/conformance/CONFORMANCE_VERSION index bd8bf88..27f9cd3 100644 --- a/tests/conformance/CONFORMANCE_VERSION +++ b/tests/conformance/CONFORMANCE_VERSION @@ -1 +1 @@ -1.7.0 +1.8.0 diff --git a/tests/conformance/manifest.json b/tests/conformance/manifest.json index 5d06e20..4b2accb 100644 --- a/tests/conformance/manifest.json +++ b/tests/conformance/manifest.json @@ -1,6 +1,6 @@ { "schema_version": 1, - "conformance_version": "1.7.0", + "conformance_version": "1.8.0", "sections": { "cases": { "scope": "core" @@ -85,6 +85,18 @@ }, "pulsar_exhaustion": { "scope": "pulsar" + }, + "version_gate_pulsar": { + "scope": "pulsar" + }, + "version_gate_artemis": { + "scope": "artemis" + }, + "version_gate_sqs": { + "scope": "sqs" + }, + "version_gate_asb": { + "scope": "asb" } }, "exceptions": [ @@ -6615,5 +6627,697 @@ } } ] + }, + "version_gate_pulsar": { + "description": "Version-gate settle rule for Apache Pulsar (broker-bindings.md section 5.7, property `bq-schema-version`) (ADR-0037 decision 1; broker-bindings.md section 5.7 \"Version-gate settle rule\"). Scope: `pulsar`. A pure function, no broker: for each case compute how the consumer settles a delivery whose schema-version property is present, not null, not blank and not the supported version, and assert `expect`. Which property values are version-gated is NOT decided here: the `schema_version_gate` block of this binding already locks the trigger. A runner feeds the delivery a property value that gates (any of that block's rejected values, for example `2`). Case `{name, on_unknown_urn?, dead_letter, dlq_write?, bq_attempts?, expect}`. `on_unknown_urn`: `fail` | `delete` | `release` | `dead_letter`; absent = `fail`. A row with any value other than `fail` applies only to an SDK whose consumer has the strategy option; the Java, .NET and Node Pulsar consumers have none (they behave as `fail`), so a runner for one of them reports those rows as `skip` citing ADR-0037, not as `pass`. `release` behaves as `fail` for a version-gated delivery (it is never released). The version gate reads only the explicitly configured `on_unknown_urn` strategy; with none configured it is `fail`. An `onUnknownUrn` hook, any strategy a hook implies (a hook-only Java or Node SQS consumer would otherwise be `delete`) and a hook's precedence over the strategy (the .NET SQS consumer) do not affect it: the hook is for an unknown URN only, so a hook-only consumer settles these rows as `fail`. `dead_letter`: `on` | `off` (dead-lettering on or off, each SDK's existing meaning; for the `dead_letter` strategy with `off` see `expect.outcome_no_writer`). `dlq_write`: `ok` | `fails`, whether the write of the raw record to `.dlq` succeeds; absent = `ok` (ADR-0037). It is read only on rows whose `expect.outcome` writes a record (`dlq_raw` or `unsettled`); it is present on those rows, and also on the `dead_letter` strategy rows with `dead_letter: off`, where an SDK with a writable DLQ target writes and an SDK without a writer asserts `outcome_no_writer` instead. `bq_attempts`: optional, the `bq-attempts` property as a string; present only to show that the outcome does not depend on it (the rule applies on the first delivery, with no retry, whatever the attempts, the native redelivery count or the max-attempts setting say). `expect.outcome`: `ack` | `dlq_raw` (the raw record is written to `.dlq`, then the delivery is acknowledged) | `unsettled` (the write failed: the delivery is neither acknowledged nor retried by the SDK and the failure is reported; negative acknowledgement). With dead-lettering off and a strategy other than `dead_letter` the outcome is `ack` (acknowledged and dropped, logged). `expect.record` (only on `dlq_raw` rows): `incoming_body`, the record written to `.dlq` has a body equal to the incoming body byte for byte and carries no `dead_letter` block (no re-encoding, no ADR-0035 wrapping). A runner checks it against every body of `incoming_bodies`, which are deliberately not in canonical form, so a writer that decodes and re-encodes fails. `expect.outcome_no_writer`: optional; a runner for an SDK in which `dead_letter: off` means NO DLQ writer exists (no publisher, sender or queue URL) asserts it instead of `expect.outcome`. It is present exactly on the rows with strategy `dead_letter` and `dead_letter: off`, where the two kinds differ (here: `ack`). Absent = both kinds assert `expect.outcome`. In every case the delivery is never decoded, no handler and no `onUnknownUrn` hook is called, and nothing is retried. The Laravel driver for this binding (if any) falls under the ADR-0037 decision 4 exception and reports every case of this section as `skip` with reason `ADR-0037 decision 4`; a package with no consumer for this binding (produce-only) follows the rule of the existing `sqs` / `artemis` binding sections (see README, \"Added in 1.8.0\"). Out of scope: Kafka (a version-gated record takes the section 6.5 poison path) and RabbitMQ.", + "incoming_bodies": [ + { + "name": "pretty-printed-escaped", + "body": "{\n \"job\": \"urn:babelqueue:order.created\",\n \"data\": { \"name\": \"caf\\u00e9\", \"path\": \"a\\/b\" },\n \"meta\": { \"schema_version\": 2 }\n}\n" + }, + { + "name": "key-order-and-number-forms", + "body": "{\"meta\": {\"schema_version\": 2}, \"data\": {\"z\": 1, \"a\": 2, \"ratio\": 1.50, \"exp\": 1E2, \"big\": 12345678901234567890}, \"job\": \"urn:babelqueue:order.created\"}" + }, + { + "name": "not-json", + "body": "plain text, not JSON \t " + } + ], + "cases": [ + { + "name": "default-dlq-on-write-ok", + "dead_letter": "on", + "dlq_write": "ok", + "expect": { + "outcome": "dlq_raw", + "record": "incoming_body" + } + }, + { + "name": "default-dlq-on-write-fails", + "dead_letter": "on", + "dlq_write": "fails", + "expect": { + "outcome": "unsettled" + } + }, + { + "name": "fail-dlq-on-write-ok", + "on_unknown_urn": "fail", + "dead_letter": "on", + "dlq_write": "ok", + "expect": { + "outcome": "dlq_raw", + "record": "incoming_body" + } + }, + { + "name": "fail-dlq-on-write-fails", + "on_unknown_urn": "fail", + "dead_letter": "on", + "dlq_write": "fails", + "expect": { + "outcome": "unsettled" + } + }, + { + "name": "fail-dlq-on-write-ok-bq-attempts-7", + "on_unknown_urn": "fail", + "dead_letter": "on", + "dlq_write": "ok", + "bq_attempts": "7", + "expect": { + "outcome": "dlq_raw", + "record": "incoming_body" + } + }, + { + "name": "release-dlq-on-write-ok", + "on_unknown_urn": "release", + "dead_letter": "on", + "dlq_write": "ok", + "expect": { + "outcome": "dlq_raw", + "record": "incoming_body" + } + }, + { + "name": "release-dlq-on-write-fails", + "on_unknown_urn": "release", + "dead_letter": "on", + "dlq_write": "fails", + "expect": { + "outcome": "unsettled" + } + }, + { + "name": "delete-dlq-on-acks", + "on_unknown_urn": "delete", + "dead_letter": "on", + "expect": { + "outcome": "ack" + } + }, + { + "name": "delete-dlq-off-acks", + "on_unknown_urn": "delete", + "dead_letter": "off", + "expect": { + "outcome": "ack" + } + }, + { + "name": "dead-letter-dlq-on-write-ok", + "on_unknown_urn": "dead_letter", + "dead_letter": "on", + "dlq_write": "ok", + "expect": { + "outcome": "dlq_raw", + "record": "incoming_body" + } + }, + { + "name": "dead-letter-dlq-on-write-fails", + "on_unknown_urn": "dead_letter", + "dead_letter": "on", + "dlq_write": "fails", + "expect": { + "outcome": "unsettled" + } + }, + { + "name": "default-dlq-off-ack", + "dead_letter": "off", + "expect": { + "outcome": "ack" + } + }, + { + "name": "fail-dlq-off-ack", + "on_unknown_urn": "fail", + "dead_letter": "off", + "expect": { + "outcome": "ack" + } + }, + { + "name": "fail-dlq-off-ack-bq-attempts-7", + "on_unknown_urn": "fail", + "dead_letter": "off", + "bq_attempts": "7", + "expect": { + "outcome": "ack" + } + }, + { + "name": "release-dlq-off-ack", + "on_unknown_urn": "release", + "dead_letter": "off", + "expect": { + "outcome": "ack" + } + }, + { + "name": "dead-letter-dlq-off-write-ok", + "on_unknown_urn": "dead_letter", + "dead_letter": "off", + "dlq_write": "ok", + "expect": { + "outcome": "dlq_raw", + "record": "incoming_body", + "outcome_no_writer": "ack" + } + }, + { + "name": "dead-letter-dlq-off-write-fails", + "on_unknown_urn": "dead_letter", + "dead_letter": "off", + "dlq_write": "fails", + "expect": { + "outcome": "unsettled", + "outcome_no_writer": "ack" + } + } + ] + }, + "version_gate_artemis": { + "description": "Version-gate settle rule for Apache ActiveMQ Artemis (broker-bindings.md section 7.7, property `bq_schema_version`) (ADR-0037 decision 1; broker-bindings.md section 5.7 \"Version-gate settle rule\"). Scope: `artemis`. A pure function, no broker: for each case compute how the consumer settles a delivery whose schema-version property is present, not null, not blank and not the supported version, and assert `expect`. Which property values are version-gated is NOT decided here: the `schema_version_gate` block of this binding already locks the trigger. A runner feeds the delivery a property value that gates (any of that block's rejected values, for example `2`). Case `{name, on_unknown_urn?, dead_letter, dlq_write?, bq_attempts?, expect}`. `on_unknown_urn`: `fail` | `delete` | `release` | `dead_letter`; absent = `fail`. Every row is asserted: the Java, .NET, Node, Go and Python consumers of this binding have the strategy option, so none of its rows may be skipped for lack of it. `release` behaves as `fail` for a version-gated delivery (it is never released). The version gate reads only the explicitly configured `on_unknown_urn` strategy; with none configured it is `fail`. An `onUnknownUrn` hook, any strategy a hook implies (a hook-only Java or Node SQS consumer would otherwise be `delete`) and a hook's precedence over the strategy (the .NET SQS consumer) do not affect it: the hook is for an unknown URN only, so a hook-only consumer settles these rows as `fail`. `dead_letter`: `on` | `off` (dead-lettering on or off, each SDK's existing meaning; for the `dead_letter` strategy with `off` see `expect.outcome_no_writer`). `dlq_write`: `ok` | `fails`, whether the write of the raw record to `.dlq` succeeds; absent = `ok` (ADR-0037). It is read only on rows whose `expect.outcome` writes a record (`dlq_raw` or `unsettled`); it is present on those rows, and also on the `dead_letter` strategy rows with `dead_letter: off`, where an SDK with a writable DLQ target writes and an SDK without a writer asserts `outcome_no_writer` instead. `bq_attempts`: optional, the `bq-attempts` property as a string; present only to show that the outcome does not depend on it (the rule applies on the first delivery, with no retry, whatever the attempts, the native redelivery count or the max-attempts setting say). `expect.outcome`: `ack` | `dlq_raw` (the raw record is written to `.dlq`, then the delivery is acknowledged) | `unsettled` (the write failed: the delivery is neither acknowledged nor retried by the SDK and the failure is reported; JMS `session.recover()` / AMQP `released`). With dead-lettering off and a strategy other than `dead_letter` the outcome is `ack` (acknowledged and dropped, logged). `expect.record` (only on `dlq_raw` rows): `incoming_body`, the record written to `.dlq` has a body equal to the incoming body byte for byte and carries no `dead_letter` block (no re-encoding, no ADR-0035 wrapping). A runner checks it against every body of `incoming_bodies`, which are deliberately not in canonical form, so a writer that decodes and re-encodes fails. `expect.outcome_no_writer`: optional; a runner for an SDK in which `dead_letter: off` means NO DLQ writer exists (no publisher, sender or queue URL) asserts it instead of `expect.outcome`. It is present exactly on the rows with strategy `dead_letter` and `dead_letter: off`, where the two kinds differ (here: `ack`). Absent = both kinds assert `expect.outcome`. In every case the delivery is never decoded, no handler and no `onUnknownUrn` hook is called, and nothing is retried. The Laravel driver for this binding (if any) falls under the ADR-0037 decision 4 exception and reports every case of this section as `skip` with reason `ADR-0037 decision 4`; a package with no consumer for this binding (produce-only) follows the rule of the existing `sqs` / `artemis` binding sections (see README, \"Added in 1.8.0\"). Out of scope: Kafka (a version-gated record takes the section 6.5 poison path) and RabbitMQ.", + "incoming_bodies": [ + { + "name": "pretty-printed-escaped", + "body": "{\n \"job\": \"urn:babelqueue:order.created\",\n \"data\": { \"name\": \"caf\\u00e9\", \"path\": \"a\\/b\" },\n \"meta\": { \"schema_version\": 2 }\n}\n" + }, + { + "name": "key-order-and-number-forms", + "body": "{\"meta\": {\"schema_version\": 2}, \"data\": {\"z\": 1, \"a\": 2, \"ratio\": 1.50, \"exp\": 1E2, \"big\": 12345678901234567890}, \"job\": \"urn:babelqueue:order.created\"}" + }, + { + "name": "not-json", + "body": "plain text, not JSON \t " + } + ], + "cases": [ + { + "name": "default-dlq-on-write-ok", + "dead_letter": "on", + "dlq_write": "ok", + "expect": { + "outcome": "dlq_raw", + "record": "incoming_body" + } + }, + { + "name": "default-dlq-on-write-fails", + "dead_letter": "on", + "dlq_write": "fails", + "expect": { + "outcome": "unsettled" + } + }, + { + "name": "fail-dlq-on-write-ok", + "on_unknown_urn": "fail", + "dead_letter": "on", + "dlq_write": "ok", + "expect": { + "outcome": "dlq_raw", + "record": "incoming_body" + } + }, + { + "name": "fail-dlq-on-write-fails", + "on_unknown_urn": "fail", + "dead_letter": "on", + "dlq_write": "fails", + "expect": { + "outcome": "unsettled" + } + }, + { + "name": "fail-dlq-on-write-ok-bq-attempts-7", + "on_unknown_urn": "fail", + "dead_letter": "on", + "dlq_write": "ok", + "bq_attempts": "7", + "expect": { + "outcome": "dlq_raw", + "record": "incoming_body" + } + }, + { + "name": "release-dlq-on-write-ok", + "on_unknown_urn": "release", + "dead_letter": "on", + "dlq_write": "ok", + "expect": { + "outcome": "dlq_raw", + "record": "incoming_body" + } + }, + { + "name": "release-dlq-on-write-fails", + "on_unknown_urn": "release", + "dead_letter": "on", + "dlq_write": "fails", + "expect": { + "outcome": "unsettled" + } + }, + { + "name": "delete-dlq-on-acks", + "on_unknown_urn": "delete", + "dead_letter": "on", + "expect": { + "outcome": "ack" + } + }, + { + "name": "delete-dlq-off-acks", + "on_unknown_urn": "delete", + "dead_letter": "off", + "expect": { + "outcome": "ack" + } + }, + { + "name": "dead-letter-dlq-on-write-ok", + "on_unknown_urn": "dead_letter", + "dead_letter": "on", + "dlq_write": "ok", + "expect": { + "outcome": "dlq_raw", + "record": "incoming_body" + } + }, + { + "name": "dead-letter-dlq-on-write-fails", + "on_unknown_urn": "dead_letter", + "dead_letter": "on", + "dlq_write": "fails", + "expect": { + "outcome": "unsettled" + } + }, + { + "name": "default-dlq-off-ack", + "dead_letter": "off", + "expect": { + "outcome": "ack" + } + }, + { + "name": "fail-dlq-off-ack", + "on_unknown_urn": "fail", + "dead_letter": "off", + "expect": { + "outcome": "ack" + } + }, + { + "name": "fail-dlq-off-ack-bq-attempts-7", + "on_unknown_urn": "fail", + "dead_letter": "off", + "bq_attempts": "7", + "expect": { + "outcome": "ack" + } + }, + { + "name": "release-dlq-off-ack", + "on_unknown_urn": "release", + "dead_letter": "off", + "expect": { + "outcome": "ack" + } + }, + { + "name": "dead-letter-dlq-off-write-ok", + "on_unknown_urn": "dead_letter", + "dead_letter": "off", + "dlq_write": "ok", + "expect": { + "outcome": "dlq_raw", + "record": "incoming_body", + "outcome_no_writer": "ack" + } + }, + { + "name": "dead-letter-dlq-off-write-fails", + "on_unknown_urn": "dead_letter", + "dead_letter": "off", + "dlq_write": "fails", + "expect": { + "outcome": "unsettled", + "outcome_no_writer": "ack" + } + } + ] + }, + "version_gate_sqs": { + "description": "Version-gate settle rule for Amazon SQS (broker-bindings.md section 3.7, message attribute `bq-schema-version`) (ADR-0037 decision 1; broker-bindings.md section 5.7 \"Version-gate settle rule\"). Scope: `sqs`. A pure function, no broker: for each case compute how the consumer settles a delivery whose schema-version property is present, not null, not blank and not the supported version, and assert `expect`. Which property values are version-gated is NOT decided here: the `schema_version_gate` block of this binding already locks the trigger. A runner feeds the delivery a property value that gates (any of that block's rejected values, for example `2`). Case `{name, on_unknown_urn?, dead_letter, dlq_write?, bq_attempts?, expect}`. `on_unknown_urn`: `fail` | `delete` | `release` | `dead_letter`; absent = `fail`. Every row is asserted: the Java, .NET, Node, Go and Python consumers of this binding have the strategy option, so none of its rows may be skipped for lack of it. `release` behaves as `fail` for a version-gated delivery (it is never released). The version gate reads only the explicitly configured `on_unknown_urn` strategy; with none configured it is `fail`. An `onUnknownUrn` hook, any strategy a hook implies (a hook-only Java or Node SQS consumer would otherwise be `delete`) and a hook's precedence over the strategy (the .NET SQS consumer) do not affect it: the hook is for an unknown URN only, so a hook-only consumer settles these rows as `fail`. `dead_letter`: `on` | `off` (dead-lettering on or off, each SDK's existing meaning; for the `dead_letter` strategy with `off` see `expect.outcome_no_writer`). `dlq_write`: `ok` | `fails`, whether the write of the raw record to `.dlq` succeeds; absent = `ok` (ADR-0037). It is read only on rows whose `expect.outcome` writes a record (`dlq_raw` or `unsettled`); it is present on those rows, and also on the `dead_letter` strategy rows with `dead_letter: off`, where an SDK with a writable DLQ target writes and an SDK without a writer asserts `outcome_no_writer` instead. `bq_attempts`: optional, the `bq-attempts` property as a string; present only to show that the outcome does not depend on it (the rule applies on the first delivery, with no retry, whatever the attempts, the native redelivery count or the max-attempts setting say). `expect.outcome`: `ack` | `dlq_raw` (the raw record is written to `.dlq`, then the delivery is acknowledged) | `unsettled` (the write failed: the delivery is neither acknowledged nor retried by the SDK and the failure is reported; no `DeleteMessage`). With dead-lettering off and a strategy other than `dead_letter` the outcome is `native` (the SDK does not settle it and leaves it to the broker's native mechanism: SQS no `DeleteMessage`, so visibility expiry redelivers it and the queue's `RedrivePolicy` / retention applies; it is reported, not retried by the SDK). `expect.record` (only on `dlq_raw` rows): `incoming_body`, the record written to `.dlq` has a body equal to the incoming body byte for byte and carries no `dead_letter` block (no re-encoding, no ADR-0035 wrapping). A runner checks it against every body of `incoming_bodies`, which are deliberately not in canonical form, so a writer that decodes and re-encodes fails. `expect.outcome_no_writer`: optional; a runner for an SDK in which `dead_letter: off` means NO DLQ writer exists (no publisher, sender or queue URL) asserts it instead of `expect.outcome`. It is present exactly on the rows with strategy `dead_letter` and `dead_letter: off`, where the two kinds differ (here: `native`). Absent = both kinds assert `expect.outcome`. In every case the delivery is never decoded, no handler and no `onUnknownUrn` hook is called, and nothing is retried. The Laravel driver for this binding (if any) falls under the ADR-0037 decision 4 exception and reports every case of this section as `skip` with reason `ADR-0037 decision 4`; a package with no consumer for this binding (produce-only) follows the rule of the existing `sqs` / `artemis` binding sections (see README, \"Added in 1.8.0\"). Out of scope: Kafka (a version-gated record takes the section 6.5 poison path) and RabbitMQ.", + "incoming_bodies": [ + { + "name": "pretty-printed-escaped", + "body": "{\n \"job\": \"urn:babelqueue:order.created\",\n \"data\": { \"name\": \"caf\\u00e9\", \"path\": \"a\\/b\" },\n \"meta\": { \"schema_version\": 2 }\n}\n" + }, + { + "name": "key-order-and-number-forms", + "body": "{\"meta\": {\"schema_version\": 2}, \"data\": {\"z\": 1, \"a\": 2, \"ratio\": 1.50, \"exp\": 1E2, \"big\": 12345678901234567890}, \"job\": \"urn:babelqueue:order.created\"}" + }, + { + "name": "not-json", + "body": "plain text, not JSON \t " + } + ], + "cases": [ + { + "name": "default-dlq-on-write-ok", + "dead_letter": "on", + "dlq_write": "ok", + "expect": { + "outcome": "dlq_raw", + "record": "incoming_body" + } + }, + { + "name": "default-dlq-on-write-fails", + "dead_letter": "on", + "dlq_write": "fails", + "expect": { + "outcome": "unsettled" + } + }, + { + "name": "fail-dlq-on-write-ok", + "on_unknown_urn": "fail", + "dead_letter": "on", + "dlq_write": "ok", + "expect": { + "outcome": "dlq_raw", + "record": "incoming_body" + } + }, + { + "name": "fail-dlq-on-write-fails", + "on_unknown_urn": "fail", + "dead_letter": "on", + "dlq_write": "fails", + "expect": { + "outcome": "unsettled" + } + }, + { + "name": "fail-dlq-on-write-ok-bq-attempts-7", + "on_unknown_urn": "fail", + "dead_letter": "on", + "dlq_write": "ok", + "bq_attempts": "7", + "expect": { + "outcome": "dlq_raw", + "record": "incoming_body" + } + }, + { + "name": "release-dlq-on-write-ok", + "on_unknown_urn": "release", + "dead_letter": "on", + "dlq_write": "ok", + "expect": { + "outcome": "dlq_raw", + "record": "incoming_body" + } + }, + { + "name": "release-dlq-on-write-fails", + "on_unknown_urn": "release", + "dead_letter": "on", + "dlq_write": "fails", + "expect": { + "outcome": "unsettled" + } + }, + { + "name": "delete-dlq-on-acks", + "on_unknown_urn": "delete", + "dead_letter": "on", + "expect": { + "outcome": "ack" + } + }, + { + "name": "delete-dlq-off-acks", + "on_unknown_urn": "delete", + "dead_letter": "off", + "expect": { + "outcome": "ack" + } + }, + { + "name": "dead-letter-dlq-on-write-ok", + "on_unknown_urn": "dead_letter", + "dead_letter": "on", + "dlq_write": "ok", + "expect": { + "outcome": "dlq_raw", + "record": "incoming_body" + } + }, + { + "name": "dead-letter-dlq-on-write-fails", + "on_unknown_urn": "dead_letter", + "dead_letter": "on", + "dlq_write": "fails", + "expect": { + "outcome": "unsettled" + } + }, + { + "name": "default-dlq-off-native", + "dead_letter": "off", + "expect": { + "outcome": "native" + } + }, + { + "name": "fail-dlq-off-native", + "on_unknown_urn": "fail", + "dead_letter": "off", + "expect": { + "outcome": "native" + } + }, + { + "name": "fail-dlq-off-native-bq-attempts-7", + "on_unknown_urn": "fail", + "dead_letter": "off", + "bq_attempts": "7", + "expect": { + "outcome": "native" + } + }, + { + "name": "release-dlq-off-native", + "on_unknown_urn": "release", + "dead_letter": "off", + "expect": { + "outcome": "native" + } + }, + { + "name": "dead-letter-dlq-off-write-ok", + "on_unknown_urn": "dead_letter", + "dead_letter": "off", + "dlq_write": "ok", + "expect": { + "outcome": "dlq_raw", + "record": "incoming_body", + "outcome_no_writer": "native" + } + }, + { + "name": "dead-letter-dlq-off-write-fails", + "on_unknown_urn": "dead_letter", + "dead_letter": "off", + "dlq_write": "fails", + "expect": { + "outcome": "unsettled", + "outcome_no_writer": "native" + } + } + ] + }, + "version_gate_asb": { + "description": "Version-gate settle rule for Azure Service Bus (broker-bindings.md section 4.7, application property `bq-schema-version`) (ADR-0037 decision 1; broker-bindings.md section 5.7 \"Version-gate settle rule\"). Scope: `asb`. A pure function, no broker: for each case compute how the consumer settles a delivery whose schema-version property is present, not null, not blank and not the supported version, and assert `expect`. Which property values are version-gated is NOT decided here: the `schema_version_gate` block of this binding already locks the trigger. A runner feeds the delivery a property value that gates (any of that block's rejected values, for example `2`). Case `{name, on_unknown_urn?, dead_letter, dlq_write?, bq_attempts?, expect}`. `on_unknown_urn`: `fail` | `delete` | `release` | `dead_letter`; absent = `fail`. A row with any value other than `fail` applies only to an SDK whose consumer has the strategy option; the Java, .NET and Node Azure Service Bus consumers have only an `onUnknownUrn` callback and no strategy option (they behave as `fail`), so a runner for one of them reports those rows as `skip` citing ADR-0037, not as `pass`; the Go and Python consumers have the option and assert every row. `release` behaves as `fail` for a version-gated delivery (it is never released). The version gate reads only the explicitly configured `on_unknown_urn` strategy; with none configured it is `fail`. An `onUnknownUrn` hook, any strategy a hook implies (a hook-only Java or Node SQS consumer would otherwise be `delete`) and a hook's precedence over the strategy (the .NET SQS consumer) do not affect it: the hook is for an unknown URN only, so a hook-only consumer settles these rows as `fail`. `dead_letter`: `on` | `off` (dead-lettering on or off, each SDK's existing meaning; for the `dead_letter` strategy with `off` see `expect.outcome_no_writer`). `dlq_write`: `ok` | `fails`, whether the write of the raw record to `.dlq` succeeds; absent = `ok` (ADR-0037). It is read only on rows whose `expect.outcome` writes a record (`dlq_raw` or `unsettled`); it is present on those rows, and also on the `dead_letter` strategy rows with `dead_letter: off`, where an SDK with a writable DLQ target writes and an SDK without a writer asserts `outcome_no_writer` instead. `bq_attempts`: optional, the `bq-attempts` property as a string; present only to show that the outcome does not depend on it (the rule applies on the first delivery, with no retry, whatever the attempts, the native redelivery count or the max-attempts setting say). `expect.outcome`: `ack` | `dlq_raw` (the raw record is written to `.dlq`, then the delivery is acknowledged) | `unsettled` (the write failed: the delivery is neither acknowledged nor retried by the SDK and the failure is reported; `Abandon`). With dead-lettering off and a strategy other than `dead_letter` the outcome is `native` (the SDK does not settle it and leaves it to the broker's native mechanism: an explicit native `DeadLetter(...)` into `$DeadLetterQueue` or `Abandon` until `MaxDeliveryCount`; it is reported, not retried by the SDK). `expect.record` (only on `dlq_raw` rows): `incoming_body`, the record written to `.dlq` has a body equal to the incoming body byte for byte and carries no `dead_letter` block (no re-encoding, no ADR-0035 wrapping). A runner checks it against every body of `incoming_bodies`, which are deliberately not in canonical form, so a writer that decodes and re-encodes fails. `expect.outcome_no_writer`: optional; a runner for an SDK in which `dead_letter: off` means NO DLQ writer exists (no publisher, sender or queue URL) asserts it instead of `expect.outcome`. It is present exactly on the rows with strategy `dead_letter` and `dead_letter: off`, where the two kinds differ (here: `native`). Absent = both kinds assert `expect.outcome`. In every case the delivery is never decoded, no handler and no `onUnknownUrn` hook is called, and nothing is retried. The Laravel driver for this binding (if any) falls under the ADR-0037 decision 4 exception and reports every case of this section as `skip` with reason `ADR-0037 decision 4`; a package with no consumer for this binding (produce-only) follows the rule of the existing `sqs` / `artemis` binding sections (see README, \"Added in 1.8.0\"). Out of scope: Kafka (a version-gated record takes the section 6.5 poison path) and RabbitMQ.", + "incoming_bodies": [ + { + "name": "pretty-printed-escaped", + "body": "{\n \"job\": \"urn:babelqueue:order.created\",\n \"data\": { \"name\": \"caf\\u00e9\", \"path\": \"a\\/b\" },\n \"meta\": { \"schema_version\": 2 }\n}\n" + }, + { + "name": "key-order-and-number-forms", + "body": "{\"meta\": {\"schema_version\": 2}, \"data\": {\"z\": 1, \"a\": 2, \"ratio\": 1.50, \"exp\": 1E2, \"big\": 12345678901234567890}, \"job\": \"urn:babelqueue:order.created\"}" + }, + { + "name": "not-json", + "body": "plain text, not JSON \t " + } + ], + "cases": [ + { + "name": "default-dlq-on-write-ok", + "dead_letter": "on", + "dlq_write": "ok", + "expect": { + "outcome": "dlq_raw", + "record": "incoming_body" + } + }, + { + "name": "default-dlq-on-write-fails", + "dead_letter": "on", + "dlq_write": "fails", + "expect": { + "outcome": "unsettled" + } + }, + { + "name": "fail-dlq-on-write-ok", + "on_unknown_urn": "fail", + "dead_letter": "on", + "dlq_write": "ok", + "expect": { + "outcome": "dlq_raw", + "record": "incoming_body" + } + }, + { + "name": "fail-dlq-on-write-fails", + "on_unknown_urn": "fail", + "dead_letter": "on", + "dlq_write": "fails", + "expect": { + "outcome": "unsettled" + } + }, + { + "name": "fail-dlq-on-write-ok-bq-attempts-7", + "on_unknown_urn": "fail", + "dead_letter": "on", + "dlq_write": "ok", + "bq_attempts": "7", + "expect": { + "outcome": "dlq_raw", + "record": "incoming_body" + } + }, + { + "name": "release-dlq-on-write-ok", + "on_unknown_urn": "release", + "dead_letter": "on", + "dlq_write": "ok", + "expect": { + "outcome": "dlq_raw", + "record": "incoming_body" + } + }, + { + "name": "release-dlq-on-write-fails", + "on_unknown_urn": "release", + "dead_letter": "on", + "dlq_write": "fails", + "expect": { + "outcome": "unsettled" + } + }, + { + "name": "delete-dlq-on-acks", + "on_unknown_urn": "delete", + "dead_letter": "on", + "expect": { + "outcome": "ack" + } + }, + { + "name": "delete-dlq-off-acks", + "on_unknown_urn": "delete", + "dead_letter": "off", + "expect": { + "outcome": "ack" + } + }, + { + "name": "dead-letter-dlq-on-write-ok", + "on_unknown_urn": "dead_letter", + "dead_letter": "on", + "dlq_write": "ok", + "expect": { + "outcome": "dlq_raw", + "record": "incoming_body" + } + }, + { + "name": "dead-letter-dlq-on-write-fails", + "on_unknown_urn": "dead_letter", + "dead_letter": "on", + "dlq_write": "fails", + "expect": { + "outcome": "unsettled" + } + }, + { + "name": "default-dlq-off-native", + "dead_letter": "off", + "expect": { + "outcome": "native" + } + }, + { + "name": "fail-dlq-off-native", + "on_unknown_urn": "fail", + "dead_letter": "off", + "expect": { + "outcome": "native" + } + }, + { + "name": "fail-dlq-off-native-bq-attempts-7", + "on_unknown_urn": "fail", + "dead_letter": "off", + "bq_attempts": "7", + "expect": { + "outcome": "native" + } + }, + { + "name": "release-dlq-off-native", + "on_unknown_urn": "release", + "dead_letter": "off", + "expect": { + "outcome": "native" + } + }, + { + "name": "dead-letter-dlq-off-write-ok", + "on_unknown_urn": "dead_letter", + "dead_letter": "off", + "dlq_write": "ok", + "expect": { + "outcome": "dlq_raw", + "record": "incoming_body", + "outcome_no_writer": "native" + } + }, + { + "name": "dead-letter-dlq-off-write-fails", + "on_unknown_urn": "dead_letter", + "dead_letter": "off", + "dlq_write": "fails", + "expect": { + "outcome": "unsettled", + "outcome_no_writer": "native" + } + } + ] } } diff --git a/tests/conformance/schema/results.example.json b/tests/conformance/schema/results.example.json index cd422f6..e2b7bb1 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.7.0", + "conformance_version": "1.8.0", "generated_at": "2026-10-18T09:30:00Z", "bindings": [ "redis", diff --git a/tests/test_app.py b/tests/test_app.py index 327b048..fdfa545 100644 --- a/tests/test_app.py +++ b/tests/test_app.py @@ -466,3 +466,117 @@ def test_unsupported_broker_scheme_raises(self) -> None: if __name__ == "__main__": unittest.main() + + +class _AckRecordingTransport(InMemoryTransport): + """In-memory transport that records every ack; ``settle_native`` is added by a subclass.""" + + def __init__(self) -> None: + super().__init__() + self.acks = 0 + + def ack(self, message: ReceivedMessage) -> None: + self.acks += 1 + + +class _NativeTransport(_AckRecordingTransport): + """The optional :class:`~babelqueue.transport.NativeSettler` capability (SQS, Azure Service Bus).""" + + def __init__(self, fail: bool = False) -> None: + super().__init__() + self.natives: list = [] + self._fail = fail + + def settle_native(self, message: ReceivedMessage, reason: str, description: str) -> None: + self.natives.append(reason) + if self._fail: + raise RuntimeError("broker unavailable") + + +class NativeSettleTest(unittest.TestCase): + """ADR-0037 decisions 1 and 2: with the DLQ off, a transport with a native mechanism gets + ``settle_native`` instead of an ack (version-gated) or the bounded retry (poison).""" + + GATE = "babelqueue:unsupported-schema-version:2" + RAW = '{"job":"urn:babel:orders:created","data":{},"meta":{"schema_version":2}}' + + def _app(self, transport, **kwargs) -> BabelQueue: + app = BabelQueue(transport=transport, queue="orders", max_attempts=5, **kwargs) + app.register("urn:babel:orders:created", lambda data, meta: None) + return app + + def test_a_version_gated_delivery_is_left_to_the_native_mechanism(self) -> None: + for kwargs in ({}, {"on_unknown_urn": UnknownUrnStrategy.RELEASE}, {"on_unknown_urn": UnknownUrnStrategy.FAIL}): + with self.subTest(**kwargs): + transport = _NativeTransport() + app = self._app(transport, **kwargs) + with self.assertLogs("babelqueue.app", level="WARNING"): + app.dispatch(ReceivedMessage(body=self.RAW, queue="orders", route_urn=self.GATE)) + self.assertEqual(["unsupported schema version"], transport.natives) + self.assertEqual(0, transport.acks) + self.assertEqual(0, transport.size("orders")) + self.assertEqual(0, transport.size("orders.dlq")) + + def test_delete_and_dead_lettering_keep_their_settle_for_a_gated_delivery(self) -> None: + transport = _NativeTransport() + app = self._app(transport, on_unknown_urn=UnknownUrnStrategy.DELETE) + app.dispatch(ReceivedMessage(body=self.RAW, queue="orders", route_urn=self.GATE)) + self.assertEqual((0, 1), (len(transport.natives), transport.acks)) + + for kwargs in ({"dead_letter": True}, {"on_unknown_urn": UnknownUrnStrategy.DEAD_LETTER}): + with self.subTest(**kwargs): + transport = _NativeTransport() + app = self._app(transport, **kwargs) + app.dispatch(ReceivedMessage(body=self.RAW, queue="orders", route_urn=self.GATE)) + self.assertEqual((0, 1), (len(transport.natives), transport.acks)) + self.assertEqual(self.RAW, transport._queues["orders.dlq"][0]) # noqa: SLF001 + + def test_a_transport_without_the_capability_still_acks_and_drops_a_gated_delivery(self) -> None: + transport = _AckRecordingTransport() + app = self._app(transport) + with self.assertLogs("babelqueue.app", level="WARNING"): + app.dispatch(ReceivedMessage(body=self.RAW, queue="orders", route_urn=self.GATE)) + self.assertEqual(1, transport.acks) + + def test_a_poison_body_is_settled_natively_on_the_first_delivery(self) -> None: + for body in ("not json", '{"foo":1}', '{"job":" ","attempts":0}', "[1]"): + for strategy in (UnknownUrnStrategy.FAIL, UnknownUrnStrategy.RELEASE): + with self.subTest(body=body, strategy=strategy): + transport = _NativeTransport() + app = self._app(transport, on_unknown_urn=strategy) + with self.assertLogs("babelqueue.app", level="WARNING"): + app.dispatch(ReceivedMessage(body=body, queue="orders")) + self.assertEqual(["poison"], transport.natives) + self.assertEqual(0, transport.acks) + self.assertEqual(0, transport.size("orders")) # nothing re-published + self.assertEqual(0, transport.size("orders.dlq")) + + def test_delete_and_dead_lettering_keep_their_settle_for_a_poison_body(self) -> None: + transport = _NativeTransport() + app = self._app(transport, on_unknown_urn=UnknownUrnStrategy.DELETE) + with self.assertLogs("babelqueue.app", level="WARNING"): + app.dispatch(ReceivedMessage(body="not json", queue="orders")) + self.assertEqual((0, 1), (len(transport.natives), transport.acks)) + + for kwargs in ({"dead_letter": True}, {"on_unknown_urn": UnknownUrnStrategy.DEAD_LETTER}): + with self.subTest(**kwargs): + transport = _NativeTransport() + app = self._app(transport, **kwargs) + app.dispatch(ReceivedMessage(body="not json", queue="orders")) + self.assertEqual((0, 1), (len(transport.natives), transport.acks)) + self.assertEqual(1, transport.size("orders.dlq")) + self.assertEqual("poison", json.loads(transport._queues["orders.dlq"][0])["dead_letter"]["reason"]) # noqa: SLF001 + + def test_a_transport_without_the_capability_keeps_the_bounded_poison_retry(self) -> None: + transport = _AckRecordingTransport() + app = self._app(transport) + app.dispatch(ReceivedMessage(body="not json", queue="orders")) + self.assertEqual(1, transport.acks) # the original is acknowledged ... + self.assertEqual(1, transport.size("orders")) # ... and an empty envelope is re-published + + def test_a_failing_native_settle_neither_acks_nor_retries(self) -> None: + transport = _NativeTransport(fail=True) + app = self._app(transport) + with self.assertLogs("babelqueue.app", level="ERROR"): + app.dispatch(ReceivedMessage(body="not json", queue="orders")) + self.assertEqual((["poison"], 0, 0), (transport.natives, transport.acks, transport.size("orders"))) diff --git a/tests/test_asb_conformance.py b/tests/test_asb_conformance.py index abd5ffe..f6990d8 100644 --- a/tests/test_asb_conformance.py +++ b/tests/test_asb_conformance.py @@ -6,10 +6,15 @@ from __future__ import annotations import json +import logging import os import unittest +from typing import Any, List, Mapping, Tuple -from babelqueue import EnvelopeCodec +from test_asb_transport import FakeClient, FakeReceived, FakeReceiver +from test_behaviour_conformance import MANIFEST, _ensure, _run_cases, iter_cases + +from babelqueue import BabelQueue, EnvelopeCodec from babelqueue.asb_transport import AsbTransport CONFORMANCE = os.path.join(os.path.dirname(__file__), "conformance") @@ -51,6 +56,11 @@ def test_attempts_reconciliation(self) -> None: ) + def test_version_gate_settle(self) -> None: + """``version_gate_asb`` (ADR-0037): every row is asserted (Python has the strategy option).""" + self.assertEqual(17, len(iter_cases(MANIFEST, "version_gate_asb"))) + _run_cases(self, "version_gate_asb", _eval_version_gate_asb) + def _gate_carrier(self, case: dict): if "absent" in case: return {} @@ -80,5 +90,88 @@ def test_schema_version_gate(self) -> None: self.assertEqual(rejected, str(case["value"]), "schema_version_gate case %r" % (case,)) +class _Capture(logging.Handler): + def __init__(self, records: List[logging.LogRecord]) -> None: + super().__init__() + self._records = records + + def emit(self, record: logging.LogRecord) -> None: + self._records.append(record) + + +def _eval_version_gate_asb(case: Mapping[str, Any]) -> None: + """``version_gate_asb`` (ADR-0037 decision 1; broker-bindings §5.7 / §4.5): a version-gated + delivery goes through the real ``AsbTransport`` and the runtime. With the DLQ off (and not the + ``dead_letter`` strategy) the outcome is ``native``: no ``Complete``, no write, no retry; this + transport dead-letters it natively (explicit ``DeadLetter`` into ``$DeadLetterQueue``). An + ``unsettled`` delivery is ``Abandon``ed (never ``Complete``d or ``DeadLetter``ed) and the failure is + reported (an error is logged). + Python always has a writable DLQ target, so ``outcome_no_writer`` never applies.""" + expect = case["expect"] + for incoming in MANIFEST["version_gate_asb"]["incoming_bodies"]: + label = f"{case['name']}/{incoming['name']}" + delivery_count = 1 + if "bq_attempts" in case: + # The outcome must not depend on the attempts or the native delivery count. + delivery_count = 8 + properties = {"bq-schema-version": "2"} + if "bq_attempts" in case: + properties["bq-attempts"] = case["bq_attempts"] + receiver = FakeReceiver([FakeReceived(incoming["body"], delivery_count, properties)]) + transport = AsbTransport(client=FakeClient(receiver=receiver)) + written: List[Tuple[str, str]] = [] + + def publish(queue: str, body: str) -> None: + if case.get("dlq_write") == "fails": + raise RuntimeError("dlq unavailable") + written.append((queue, body)) + + transport.publish = publish # type: ignore[method-assign] # the real one needs azure-servicebus + calls: list = [] + app = BabelQueue( + transport=transport, + queue="orders", + max_attempts=3, + dead_letter=case["dead_letter"] == "on", + on_unknown_urn=case.get("on_unknown_urn", "fail"), + ) + app.register("urn:babelqueue:order.created", lambda data, meta: calls.append("called")) + app._rewind_pause = 0 # noqa: SLF001 - no pause between the abandon and the next poll + records: List[logging.LogRecord] = [] + capture = _Capture(records) + logging.getLogger("babelqueue.app").addHandler(capture) + try: + app.consume("orders", max_messages=1, timeout=0, handle_signals=False) + finally: + logging.getLogger("babelqueue.app").removeHandler(capture) + + _ensure(calls == [], f"{label}: a handler ran for a version-gated delivery") + _ensure( + all(queue == "orders.dlq" for queue, _ in written), + f"{label}: only the DLQ may be written, got {written!r}", + ) + outcome = expect["outcome"] + if outcome == "ack": + _ensure(len(receiver.completed) == 1 and written == [] and receiver.dead_lettered == [], + f"{label}: want one Complete and nothing else") + elif outcome == "dlq_raw": + _ensure(len(written) == 1 and len(receiver.completed) == 1 and receiver.dead_lettered == [], + f"{label}: want one DLQ write then one Complete") + if expect.get("record") == "incoming_body": + _ensure(written[0][1] == incoming["body"], f"{label}: record is not the incoming body") + elif outcome == "unsettled": + _ensure(receiver.completed == [] and receiver.dead_lettered == [] and written == [], + f"{label}: want neither a Complete, a DeadLetter nor a write") + _ensure(len(receiver.abandoned) == 1, f"{label}: want exactly one Abandon, got {len(receiver.abandoned)}") + _ensure(any(r.levelno >= logging.ERROR for r in records), f"{label}: the failure must be reported") + elif outcome == "native": + _ensure(receiver.completed == [] and written == [] and len(receiver.dead_lettered) == 1, + f"{label}: native means no Complete, no write and one native DeadLetter") + _ensure(receiver.dead_lettered[0][1] == "unsupported schema version", + f"{label}: DeadLetterReason {receiver.dead_lettered[0][1]!r}") + else: + raise AssertionError(f"{label}: unhandled expected outcome {outcome!r}") + + if __name__ == "__main__": # pragma: no cover unittest.main() diff --git a/tests/test_asb_transport.py b/tests/test_asb_transport.py index 9f463cb..9f98683 100644 --- a/tests/test_asb_transport.py +++ b/tests/test_asb_transport.py @@ -13,7 +13,7 @@ from babelqueue import BabelQueue, EnvelopeCodec from babelqueue.asb_transport import AsbTransport -from babelqueue.transport import ReceivedMessage, make_transport +from babelqueue.transport import Abandoner, ReceivedMessage, Rewinder, make_transport try: import azure.servicebus # noqa: F401 @@ -72,6 +72,14 @@ class FakeReceiver: def __init__(self, messages=()) -> None: self._messages = list(messages) self.completed: list = [] + self.dead_lettered: list = [] # (message, reason, error_description) + self.abandoned: list = [] + + def dead_letter_message(self, message, reason=None, error_description=None) -> None: + self.dead_lettered.append((message, reason, error_description)) + + def abandon_message(self, message) -> None: + self.abandoned.append(message) def receive_messages(self, max_message_count: int = 1, max_wait_time=None): batch = self._messages[:max_message_count] @@ -273,6 +281,145 @@ def test_gated_message_is_completed_through_its_handle(self): tr.ack(got) self.assertEqual(receiver.completed, [got.handle]) + def test_settle_native_dead_letters_with_reason_and_description(self): + receiver = FakeReceiver([FakeReceived("{not json", 1, {"bq-schema-version": "2"})]) + tr = AsbTransport(client=FakeClient(receiver=receiver)) + got = tr.pop("orders") + tr.settle_native(got, "unsupported schema version", "declares version 2") + self.assertEqual( + receiver.dead_lettered, [(got.handle, "unsupported schema version", "declares version 2")] + ) + self.assertEqual(receiver.completed, []) + self.assertEqual(receiver.abandoned, []) + + def test_settle_native_abandons_when_the_receiver_cannot_dead_letter(self): + class AbandonOnly: + def __init__(self, inner): + self.inner = inner + + def receive_messages(self, **kwargs): + return self.inner.receive_messages(**kwargs) + + def abandon_message(self, message): + self.inner.abandon_message(message) + + inner = FakeReceiver([FakeReceived("{not json", 1, {"bq-schema-version": "2"})]) + tr = AsbTransport(client=FakeClient(receiver=AbandonOnly(inner))) + got = tr.pop("orders") + tr.settle_native(got, "poison", "x") + self.assertEqual(inner.abandoned, [got.handle]) + self.assertEqual(inner.completed, []) + + def test_abandon_releases_the_lock_and_never_completes(self): + receiver = FakeReceiver([FakeReceived(body(), 1, {"bq-schema-version": "2"})]) + tr = AsbTransport(client=FakeClient(receiver=receiver)) + got = tr.pop("orders") + self.assertIsInstance(tr, Abandoner) + tr.abandon(got) + self.assertEqual(receiver.abandoned, [got.handle]) + self.assertEqual(receiver.completed, []) + self.assertEqual(receiver.dead_lettered, []) + + def test_abandon_is_a_noop_without_a_handle(self): + receiver = FakeReceiver() + tr = AsbTransport(client=FakeClient(receiver=receiver)) + tr.abandon(ReceivedMessage(body="x", queue="orders", handle=None)) + self.assertEqual(receiver.abandoned, []) + + def test_a_failed_dlq_write_for_a_gated_message_abandons_it(self): + receiver = FakeReceiver([FakeReceived(body(), 1, {"bq-schema-version": "2"})]) + app = BabelQueue( + transport=AsbTransport(client=FakeClient(receiver=receiver)), max_attempts=5, dead_letter=True + ) + app._rewind_pause = 0 # noqa: SLF001 + with mock.patch.object(AsbTransport, "publish", side_effect=RuntimeError("dlq unavailable")): + with self.assertLogs("babelqueue.app", level="ERROR"): + app.consume(queue="orders", max_messages=1) + self.assertEqual(len(receiver.abandoned), 1) + self.assertEqual(receiver.completed, []) + + def test_a_failed_dlq_write_for_a_poison_body_is_not_abandoned(self): + # Only the version-gate unsettled branch abandons (owner decision 2026-10-25). + receiver = FakeReceiver([FakeReceived("not json", 1)]) + app = BabelQueue( + transport=AsbTransport(client=FakeClient(receiver=receiver)), max_attempts=5, dead_letter=True + ) + app._rewind_pause = 0 # noqa: SLF001 + with mock.patch.object(AsbTransport, "publish", side_effect=RuntimeError("dlq unavailable")): + with self.assertLogs("babelqueue.app", level="ERROR"): + app.consume(queue="orders", max_messages=1) + self.assertEqual((receiver.abandoned, receiver.completed), ([], [])) + + def test_asb_is_not_a_rewinder_so_shutdown_and_retry_paths_keep_their_old_handling(self): + self.assertNotIsInstance(AsbTransport(client=FakeClient()), Rewinder) + + def test_graceful_shutdown_release_republishes_a_copy_and_completes(self): + # A message whose handler never completed at shutdown: copy re-publish + Complete, no Abandon. + raw = body() + received = FakeReceived(raw, 1) + receiver = FakeReceiver([received]) + tr = AsbTransport(client=FakeClient(receiver=receiver)) + app = BabelQueue(transport=tr, max_attempts=5) + msg = tr.pop("orders") + with mock.patch.object(AsbTransport, "publish") as publish: + app._release_unprocessed(msg) # noqa: SLF001 + publish.assert_called_once_with("orders", raw) + self.assertEqual(receiver.completed, [received]) + self.assertEqual((receiver.abandoned, receiver.dead_lettered), ([], [])) + + def test_a_failed_retry_republish_is_left_unsettled_without_abandon(self): + received = FakeReceived(body(), 1) + receiver = FakeReceiver([received]) + app = BabelQueue(transport=AsbTransport(client=FakeClient(receiver=receiver)), max_attempts=5) + app._rewind_pause = 0 # noqa: SLF001 + + @app.handler(URN) + def _boom(data, meta): + raise RuntimeError("handler failed") + + with mock.patch.object(AsbTransport, "publish", side_effect=RuntimeError("broker down")): + with self.assertLogs("babelqueue.app", level="ERROR"): + app.consume(queue="orders", max_messages=1) + self.assertEqual((receiver.abandoned, receiver.completed, receiver.dead_lettered), ([], [], [])) + + def test_settle_native_is_a_noop_without_a_handle(self): + receiver = FakeReceiver() + tr = AsbTransport(client=FakeClient(receiver=receiver)) + tr.settle_native(ReceivedMessage(body="x", queue="orders", handle=None), "poison", "x") + self.assertEqual(receiver.dead_lettered, []) + + def test_app_without_dead_letter_dead_letters_natively_not_completes(self): + # ADR-0037 decision 1 rule 3: DLQ off + default fail -> explicit native DeadLetter. + for kwargs in ({}, {"on_unknown_urn": "release"}): + with self.subTest(**kwargs): + receiver = FakeReceiver([FakeReceived(body(), 1, {"bq-schema-version": "2"})]) + app = BabelQueue( + transport=AsbTransport(client=FakeClient(receiver=receiver)), max_attempts=5, **kwargs + ) + seen: list = [] + + @app.handler(URN) + def _on(data, meta): + seen.append(data) + + with self.assertLogs("babelqueue.app", level="WARNING"): + app.consume(queue="orders", max_messages=1) + + self.assertEqual(seen, []) + self.assertEqual(receiver.completed, []) + self.assertEqual([r for _, r, _ in receiver.dead_lettered], ["unsupported schema version"]) + + def test_delete_strategy_completes_a_gated_message_without_dead_letter(self): + receiver = FakeReceiver([FakeReceived(body(), 1, {"bq-schema-version": "2"})]) + app = BabelQueue( + transport=AsbTransport(client=FakeClient(receiver=receiver)), + max_attempts=5, + on_unknown_urn="delete", + ) + app.consume(queue="orders", max_messages=1) + self.assertEqual(len(receiver.completed), 1) + self.assertEqual(receiver.dead_lettered, []) + def test_app_dead_letters_unsupported_version_without_reaching_a_handler(self): # max_attempts > 1: a gated message is dead-lettered directly, never retried for value in ("2", "x", 2, "2&x=%"): diff --git a/tests/test_behaviour_conformance.py b/tests/test_behaviour_conformance.py index db80a39..ae6eaaa 100644 --- a/tests/test_behaviour_conformance.py +++ b/tests/test_behaviour_conformance.py @@ -46,7 +46,7 @@ from babelqueue.redis_transport import RedisTransport, _LegacyHandle from babelqueue.schema import validate_schema from babelqueue.sqs_transport import SqsTransport -from babelqueue.transport import NON_NUMERIC_ATTEMPTS_ROUTE, PoisonRoute, ReceivedMessage +from babelqueue.transport import NON_NUMERIC_ATTEMPTS_ROUTE, PoisonRoute, ReceivedMessage, Transport SUITE = Path(__file__).parent / "conformance" MANIFEST = json.loads((SUITE / "manifest.json").read_text(encoding="utf-8")) @@ -1438,6 +1438,95 @@ def fail(data: Any, meta: Any) -> None: ) +_GATE_ROUTE = "babelqueue:unsupported-schema-version:2" + + +class VersionGateRecorder(Transport): + """A transport that hands the runtime ONE version-gated delivery (the reserved route a gating + transport sets) and records every publish and ack. It has no ``NativeSettler`` capability, so it + models Pulsar and Artemis (which keep "acknowledge and log"). ``fail_dlq`` makes every write to a + ``.dlq`` queue fail.""" + + def __init__(self, body: str, native_attempts: int = 0, fail_dlq: bool = False) -> None: + self._pending: List[ReceivedMessage] = [ + ReceivedMessage(body=body, queue="orders", route_urn=_GATE_ROUTE, native_attempts=native_attempts) + ] + self.fail_dlq = fail_dlq + self.published: List[Tuple[str, str]] = [] + self.acks = 0 + + def publish(self, queue: str, body: str) -> None: + if self.fail_dlq and queue.endswith(".dlq"): + raise RuntimeError("dlq unavailable") + self.published.append((queue, body)) + + def pop(self, queue: str, timeout: float = 1.0) -> Optional[ReceivedMessage]: + return self._pending.pop(0) if self._pending else None + + def ack(self, message: ReceivedMessage) -> None: + self.acks += 1 + + +def version_gate_app(case: Mapping[str, Any], transport: Any) -> Tuple[BabelQueue, List[str]]: + """The runtime configured by a ``version_gate_*`` row: ``dead_letter`` on/off and the + ``on_unknown_urn`` strategy (absent = ``fail``). The returned list collects handler calls.""" + calls: List[str] = [] + app = BabelQueue( + transport=transport, + queue="orders", + max_attempts=3, + dead_letter=case["dead_letter"] == "on", + on_unknown_urn=case.get("on_unknown_urn", "fail"), + ) + app.register("urn:babelqueue:order.created", lambda data, meta: calls.append("called")) + return app, calls + + +def _version_gate_bodies(section: str) -> List[Mapping[str, Any]]: + return MANIFEST[section]["incoming_bodies"] + + +def _eval_version_gate_app(section: str) -> Callable[[Mapping[str, Any]], None]: + """``version_gate_pulsar`` / ``version_gate_artemis`` (ADR-0037 decision 1; broker-bindings + §5.7): the gating transport hands over the reserved route, so the rows run through the runtime + over a recording transport without the ``NativeSettler`` capability (those two bindings keep + "acknowledge and log"). Python has the ``on_unknown_urn`` option: every row runs.""" + + def evaluate(case: Mapping[str, Any]) -> None: + expect = case["expect"] + for incoming in _version_gate_bodies(section): + transport = VersionGateRecorder( + incoming["body"], + native_attempts=7 if "bq_attempts" in case else 0, + fail_dlq=case.get("dlq_write") == "fails", + ) + app, calls = version_gate_app(case, transport) + app.consume("orders", max_messages=1, timeout=0.01, handle_signals=False) + label = f"{case['name']}/{incoming['name']}" + _ensure(calls == [], f"{label}: a handler ran for a version-gated delivery") + _ensure( + all(queue.endswith(".dlq") for queue, _ in transport.published), + f"{label}: published to {transport.published!r}: a gated delivery is never retried", + ) + outcome = expect["outcome"] + if outcome == "ack": + _ensure(transport.acks == 1 and transport.published == [], f"{label}: want one ack and no write") + elif outcome == "dlq_raw": + _ensure(transport.acks == 1 and len(transport.published) == 1, f"{label}: want one write then one ack") + if expect.get("record") == "incoming_body": + written = transport.published[0][1] + _ensure(written == incoming["body"], f"{label}: record {written!r} != incoming body") + elif outcome == "unsettled": + _ensure( + transport.acks == 0 and transport.published == [], + f"{label}: want no ack and no write, got acks={transport.acks} published={transport.published!r}", + ) + else: + raise AssertionError(f"{label}: unhandled expected outcome {outcome!r}") + + return evaluate + + 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]] = [] @@ -1563,8 +1652,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.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} + #: ``section -> number of case ids`` of the vendored conformance 1.8.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, "version_gate_pulsar": 17, "version_gate_artemis": 17, "version_gate_sqs": 17, "version_gate_asb": 17} def test_case_id_counts_match_the_manifest(self) -> None: for section, count in self.EXPECTED_COUNTS.items(): @@ -1593,7 +1682,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.7.0 case is implemented (r1 Faz 08; conformance 1.7.0, post-R1 Faz 06): the gap list stays empty, and a + # Every conformance 1.8.0 case is implemented (r1 Faz 08; conformance 1.8.0, post-R1 Faz 06, poison-shape-parity Faz 07): the gap list stays empty, and a # new entry is a deliberate, reviewed change to this test as well. self.assertEqual({}, KNOWN_GAPS) @@ -1655,6 +1744,21 @@ def test_the_strategy_rows_are_not_skipped(self) -> None: self.assertEqual(10, len(rows)) +class VersionGatePulsarConformanceTest(unittest.TestCase): + def test_version_gate_pulsar_cases(self) -> None: + _run_cases(self, "version_gate_pulsar", _eval_version_gate_app("version_gate_pulsar")) + + 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, "version_gate_pulsar") if "on_unknown_urn" in c] + self.assertEqual(14, len(rows)) + + +class VersionGateArtemisConformanceTest(unittest.TestCase): + def test_version_gate_artemis_cases(self) -> None: + _run_cases(self, "version_gate_artemis", _eval_version_gate_app("version_gate_artemis")) + + 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 598f86e..da9053f 100644 --- a/tests/test_conformance.py +++ b/tests/test_conformance.py @@ -103,6 +103,10 @@ def test_cases(self) -> None: "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")], + "version_gate_pulsar": [("test_behaviour_conformance", "VersionGatePulsarConformanceTest", "test_version_gate_pulsar_cases")], + "version_gate_artemis": [("test_behaviour_conformance", "VersionGateArtemisConformanceTest", "test_version_gate_artemis_cases")], + "version_gate_sqs": [("test_sqs_conformance", "SqsConformanceTest", "test_version_gate_settle")], + "version_gate_asb": [("test_asb_conformance", "AsbConformanceTest", "test_version_gate_settle")], # 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_poison_routing.py b/tests/test_poison_routing.py index e715cbd..6b9ad03 100644 --- a/tests/test_poison_routing.py +++ b/tests/test_poison_routing.py @@ -219,21 +219,46 @@ def test_a_poison_message_is_dead_lettered_with_the_original_body_and_deleted(se 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 + def test_a_poison_message_is_left_to_sqs_when_the_dlq_is_off(self) -> None: + # ADR-0037 decision 2 (Go and Python): with the DLQ off and a strategy other than delete, + # a poison body (invalid JSON, identity-less object) is settled natively on the FIRST + # delivery: no DeleteMessage, no empty-envelope re-send, no ChangeMessageVisibility retry. + # Visibility expiry redelivers it and the queue's RedrivePolicy takes over. + for body in ("not json", '{"foo":1}'): + for strategy in (UnknownUrnStrategy.FAIL, UnknownUrnStrategy.RELEASE): + with self.subTest(body=body, strategy=strategy): + fake = FakeSQS() + transport = SqsTransport("sqs://", client=fake, queue_url_prefix="http://fake") + fake.seed(self.URL, body, 1) + app = BabelQueue(transport=transport, queue="orders", max_attempts=3, + on_unknown_urn=strategy) + app.register(URN, lambda data, meta: None) + with self.assertLogs("babelqueue.app", level="WARNING") as logs: + app.consume("orders", max_messages=1, timeout=0, handle_signals=False) + self.assertEqual([], fake.deleted) + self.assertEqual([], fake.sent) + self.assertEqual([], fake.visibility_changes) + self.assertTrue(any("native mechanism" in line for line in logs.output)) + + def test_delete_still_deletes_and_the_dlq_on_record_is_unchanged(self) -> None: + fake = FakeSQS() + transport = SqsTransport("sqs://", client=fake, queue_url_prefix="http://fake") + fake.seed(self.URL, "not json", 1) + app = BabelQueue(transport=transport, queue="orders", max_attempts=3, + on_unknown_urn=UnknownUrnStrategy.DELETE) 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) + app.consume("orders", max_messages=1, timeout=0, handle_signals=False) + self.assertEqual(1, len(fake.deleted)) + self.assertEqual([], fake.sent) + + fake = FakeSQS() + transport = SqsTransport("sqs://", client=fake, queue_url_prefix="http://fake") + fake.seed(self.URL, "not json", 1) + app = BabelQueue(transport=transport, queue="orders", max_attempts=3, + on_unknown_urn=UnknownUrnStrategy.DEAD_LETTER) + app.consume("orders", max_messages=1, timeout=0, handle_signals=False) + self.assertEqual(1, len(fake.deleted)) + assert_poison_record(self, fake.sent[-1]["MessageBody"], "not json") def test_the_delivery_carries_the_native_attempts(self) -> None: self.fake.seed(self.URL, "not json", 4) @@ -413,6 +438,51 @@ def test_a_poison_message_is_dead_lettered_and_completed(self) -> None: self.assertEqual([received], receiver.completed) + def test_a_poison_message_is_dead_lettered_natively_when_the_dlq_is_off(self) -> None: + # ADR-0037 decision 2 (Go and Python): no Complete, no re-send, no retry; the body is + # unchanged and Azure Service Bus moves it to $DeadLetterQueue (reason "poison"). + for body in ("not json", '{"foo":1}'): + for strategy in (UnknownUrnStrategy.FAIL, UnknownUrnStrategy.RELEASE): + with self.subTest(body=body, strategy=strategy): + received = AsbReceived(body) + receiver = AsbReceiver([received]) + transport = AsbTransport(client=AsbClient(receiver=receiver)) + capture = _PublishCapture(transport) + app = BabelQueue(transport=transport, queue="orders", max_attempts=3, + on_unknown_urn=strategy) + app.register(URN, lambda data, meta: None) + with self.assertLogs("babelqueue.app", level="WARNING"): + app.consume("orders", max_messages=1, timeout=0, handle_signals=False) + self.assertEqual([], receiver.completed) + self.assertEqual([], capture.written) + self.assertEqual([(received, "poison")], [(m, r) for m, r, _ in receiver.dead_lettered]) + + def test_delete_completes_and_the_dlq_on_record_is_unchanged(self) -> None: + received = AsbReceived("not json") + receiver = AsbReceiver([received]) + transport = AsbTransport(client=AsbClient(receiver=receiver)) + app = BabelQueue(transport=transport, queue="orders", max_attempts=3, + on_unknown_urn=UnknownUrnStrategy.DELETE) + with self.assertLogs("babelqueue.app", level="WARNING"): + app.consume("orders", max_messages=1, timeout=0, handle_signals=False) + self.assertEqual([received], receiver.completed) + self.assertEqual([], receiver.dead_lettered) + + received = AsbReceived("not json") + receiver = AsbReceiver([received]) + transport = AsbTransport(client=AsbClient(receiver=receiver)) + capture = _PublishCapture(transport) + app = BabelQueue(transport=transport, queue="orders", max_attempts=3, + on_unknown_urn=UnknownUrnStrategy.DEAD_LETTER) + 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([received], receiver.completed) + self.assertEqual([], receiver.dead_lettered) + + class ArtemisPoisonTest(unittest.TestCase): def test_a_poison_message_is_dead_lettered_and_accepted(self) -> None: for body in ("not json", '{"foo":1}'): diff --git a/tests/test_sqs_conformance.py b/tests/test_sqs_conformance.py index a0de48d..c5e66a8 100644 --- a/tests/test_sqs_conformance.py +++ b/tests/test_sqs_conformance.py @@ -7,8 +7,12 @@ import json import os import unittest +from typing import Any, Mapping -from babelqueue import EnvelopeCodec +from test_behaviour_conformance import MANIFEST, _ensure, _run_cases, iter_cases +from test_sqs_transport import FakeSQS + +from babelqueue import BabelQueue, EnvelopeCodec from babelqueue.sqs_transport import SqsTransport CONFORMANCE = os.path.join(os.path.dirname(__file__), "conformance") @@ -50,6 +54,11 @@ def test_attempts_reconciliation(self) -> None: ) + def test_version_gate_settle(self) -> None: + """``version_gate_sqs`` (ADR-0037): every row is asserted (Python has the strategy option).""" + self.assertEqual(17, len(iter_cases(MANIFEST, "version_gate_sqs"))) + _run_cases(self, "version_gate_sqs", _eval_version_gate_sqs) + def _gate_carrier(self, case: dict): if "absent" in case: return {} @@ -76,5 +85,71 @@ def test_schema_version_gate(self) -> None: self.assertEqual(rejected, str(case["value"]), "schema_version_gate case %r" % (case,)) +class _DlqFailsSQS(FakeSQS): + """FakeSQS whose ``SendMessage`` fails (the raw record cannot be written to ``.dlq``).""" + + def send_message(self, **kw): + raise RuntimeError("dlq unavailable") + + +def _eval_version_gate_sqs(case: Mapping[str, Any]) -> None: + """``version_gate_sqs`` (ADR-0037 decision 1; broker-bindings §5.7 / §3.5): a version-gated + delivery goes through the real ``SqsTransport`` and the runtime. With the DLQ off (and not the + ``dead_letter`` strategy) the outcome is ``native``: no ``DeleteMessage``, no write, no + visibility change. Python always has a writable DLQ target, so ``outcome_no_writer`` never + applies.""" + expect = case["expect"] + for incoming in MANIFEST["version_gate_sqs"]["incoming_bodies"]: + label = f"{case['name']}/{incoming['name']}" + fake = _DlqFailsSQS() if case.get("dlq_write") == "fails" else FakeSQS() + attributes = {"bq-schema-version": {"DataType": "Number", "StringValue": "2"}} + receive_count = "1" + if "bq_attempts" in case: + # The outcome must not depend on the attempts or the native receive count. + attributes["bq-attempts"] = {"DataType": "Number", "StringValue": case["bq_attempts"]} + receive_count = "8" + fake.visible["http://fake/orders"] = [ + { + "Body": incoming["body"], + "ReceiptHandle": "rh-gate", + "MessageAttributes": attributes, + "Attributes": {"ApproximateReceiveCount": receive_count}, + } + ] + calls: list = [] + app = BabelQueue( + transport=SqsTransport("sqs://", client=fake, queue_url_prefix="http://fake"), + queue="orders", + max_attempts=3, + dead_letter=case["dead_letter"] == "on", + on_unknown_urn=case.get("on_unknown_urn", "fail"), + ) + app.register("urn:babelqueue:order.created", lambda data, meta: calls.append("called")) + app.consume("orders", max_messages=1, timeout=0, handle_signals=False) + + _ensure(calls == [], f"{label}: a handler ran for a version-gated delivery") + _ensure(fake.visibility_changes == [], f"{label}: a gated delivery is never released") + _ensure( + all(sent["QueueUrl"] == "http://fake/orders.dlq" for sent in fake.sent), + f"{label}: only the DLQ may be written", + ) + outcome = expect["outcome"] + if outcome == "ack": + _ensure(fake.deleted == ["rh-gate"] and fake.sent == [], f"{label}: want one DeleteMessage and no write") + elif outcome == "dlq_raw": + _ensure(len(fake.sent) == 1 and fake.deleted == ["rh-gate"], f"{label}: want one write then one delete") + if expect.get("record") == "incoming_body": + _ensure(fake.sent[0]["MessageBody"] == incoming["body"], f"{label}: record is not the incoming body") + elif outcome == "unsettled": + _ensure(fake.deleted == [], f"{label}: a failed write must not delete the message") + elif outcome == "native": + _ensure( + fake.deleted == [] and fake.sent == [], + f"{label}: native means no DeleteMessage and no write, got deleted={fake.deleted!r} sent={fake.sent!r}", + ) + else: + raise AssertionError(f"{label}: unhandled expected outcome {outcome!r}") + + if __name__ == "__main__": # pragma: no cover unittest.main() diff --git a/tests/test_sqs_transport.py b/tests/test_sqs_transport.py index 466a2b4..7db5347 100644 --- a/tests/test_sqs_transport.py +++ b/tests/test_sqs_transport.py @@ -660,22 +660,49 @@ def _on(data, meta): self.assertEqual(fake.sent[0]["MessageBody"], incoming) self.assertNotIn("dead_letter", fake.sent[0]["MessageBody"]) - def test_app_without_dead_letter_drops_unsupported_version_without_retry(self): - tr, fake = self._tr() - app = BabelQueue(transport=tr, queue="orders", max_attempts=3, retry_backoff=30) - seen: list = [] + def test_app_without_dead_letter_leaves_unsupported_version_to_sqs_without_retry(self): + # ADR-0037 decision 1 rule 3: DLQ off + the default fail strategy. The message is neither + # deleted nor re-sent nor released: it is left to visibility expiry and the queue's + # RedrivePolicy (the gate rejects it again on every redelivery). + for strategy in (UnknownUrnStrategy.FAIL, UnknownUrnStrategy.RELEASE): + with self.subTest(strategy=strategy): + tr, fake = self._tr() + app = BabelQueue(transport=tr, queue="orders", max_attempts=3, retry_backoff=30, + on_unknown_urn=strategy) + seen: list = [] - @app.handler(self.URN) - def _on(data, meta): - seen.append(data) + @app.handler(self.URN) + def _on(data, meta): + seen.append(data) + + self._enqueue(fake, self._body(), "2") + with self.assertLogs("babelqueue.app", level="WARNING") as logs: + app.consume("orders", max_messages=1, timeout=0) + + self.assertEqual(seen, []) + self.assertEqual(fake.sent, []) + self.assertEqual(fake.visibility_changes, []) # no ChangeMessageVisibility retry loop + self.assertEqual(fake.deleted, []) # no DeleteMessage + self.assertTrue(any("native mechanism" in line for line in logs.output)) + def test_delete_strategy_still_deletes_an_unsupported_version_without_dead_letter(self): + tr, fake = self._tr() + app = BabelQueue(transport=tr, queue="orders", max_attempts=3, + on_unknown_urn=UnknownUrnStrategy.DELETE) self._enqueue(fake, self._body(), "2") app.consume("orders", max_messages=1, timeout=0) + self.assertEqual(fake.deleted, ["rh-gate"]) + self.assertEqual(fake.sent, []) - self.assertEqual(seen, []) + def test_settle_native_does_nothing_on_the_wire(self): + tr, fake = self._tr() + self._enqueue(fake, self._body(), "2", receive_count=None) + got = tr.pop("orders", timeout=0) + tr.settle_native(got, "unsupported schema version", "x") + self.assertEqual(fake.deleted, []) self.assertEqual(fake.sent, []) - self.assertEqual(fake.visibility_changes, []) # no ChangeMessageVisibility retry loop - self.assertEqual(fake.deleted, ["rh-gate"]) + self.assertEqual(fake.visibility_changes, []) + self.assertNotIn("rh-gate", tr._uncounted_handles) # the handle is not leaked def test_gated_message_without_receive_count_is_still_acknowledged_by_its_handle(self): tr, fake = self._tr() From f0ec8979f9fa40d33e607798475c29296a40b6ba Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Muhammet=20=C5=9Eafak?= Date: Mon, 26 Oct 2026 07:23:41 +0300 Subject: [PATCH 2/3] Default the SQS dead-letter queue to -dlq (ADR-0037) SQS queue names cannot contain a dot, so the old .dlq default could never exist on AWS. The SQS transport now defaults to -dlq and places any suffix before .fifo; an explicit dead-letter queue still wins. Other transports keep .dlq. --- CHANGELOG.md | 15 ++++++++ README.md | 4 +- src/babelqueue/__init__.py | 2 + src/babelqueue/app.py | 16 ++++++-- src/babelqueue/sqs_transport.py | 13 +++++++ src/babelqueue/transport.py | 15 ++++++++ tests/test_poison_routing.py | 2 +- tests/test_sqs_conformance.py | 4 +- tests/test_sqs_transport.py | 65 +++++++++++++++++++++++++++++++-- 9 files changed, 125 insertions(+), 11 deletions(-) diff --git a/CHANGELOG.md b/CHANGELOG.md index 974e506..40eabc9 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -38,6 +38,21 @@ version number is set at release time. `RedrivePolicy` (SQS) or accepts them in `$DeadLetterQueue` (Azure Service Bus), or sets `dead_letter=True` / `on_unknown_urn=DELETE`. +### Fixed +- **SQS: the default dead-letter queue name is now `-dlq` (FIFO `-dlq.fifo`) (ADR-0037 + decision 7).** The old default `.dlq` was never a valid Amazon SQS queue name (a name may hold only + alphanumerics, hyphens and underscores, plus one `.fifo` suffix), so a DLQ write by default could not find a + queue on real SQS. `dead_letter_suffix` now defaults to `None` ("not set"); unless `dead_letter_queue` is set + the runtime asks the transport's new optional `DeadLetterNamer` capability + (`dead_letter_queue_name(queue, suffix)`, exported from `babelqueue`), which `SqsTransport` implements. The + runtime also hands it an explicit `dead_letter_suffix`; on a FIFO queue the `.fifo` suffix always stays last, + an explicit suffix included: `orders.fifo` with `dead_letter_suffix="-x"` is `orders-x.fifo`, never + `orders.fifo-x` (an SQS FIFO name must end in `.fifo`; before, that explicit suffix produced the uncreatable + `orders.fifo-x`). An explicit queue name behaves as before; every other transport keeps `.dlq` (the + attribute `App.dead_letter_suffix` is `None` instead of `".dlq"` until set). Deployments + that relied on the SQS default must name the DLQ `-dlq`, or pass `dead_letter_suffix=".dlq"` / + `dead_letter_queue=` to keep the old name. An explicitly empty `dead_letter_suffix=""` now counts as "no suffix" on SQS (`-dlq`); before, it made the DLQ the source queue itself. + ### Added - **`babelqueue.transport.NativeSettler`**, an optional `Transport` capability (`settle_native(message, reason, description)`). With the DLQ off the runtime calls it instead of diff --git a/README.md b/README.md index 059656f..f78f254 100644 --- a/README.md +++ b/README.md @@ -205,7 +205,7 @@ standard queue; a larger value, or any per-message delay on a FIFO queue, raises `BabelQueueError` instead of being truncated or dropped). **Poison bodies.** A body that does not decode, or has no identity, is never retried by the SDK -(ADR-0037): `dead_letter=True` writes the ADR-0035 record to `.dlq` and then deletes it, +(ADR-0037): `dead_letter=True` writes the ADR-0035 record to the DLQ (`-dlq` on SQS, `.dlq` elsewhere) and then deletes it, `on_unknown_urn=DELETE` deletes it, and with the DLQ off it is **not deleted and not re-sent**: the transport's `settle_native` capability leaves it to SQS on the first delivery (visibility expiry, then the queue's `RedrivePolicy`; the event is logged). Azure Service Bus does the same with a native @@ -214,7 +214,7 @@ then the queue's `RedrivePolicy`; the event is logged). Azure Service Bus does t **Poison-loop risk:** a message that always fails is redelivered at once until `max_attempts` is reached. If the receive count is unavailable the SDK cannot advance attempts at all. **Configure a native `RedrivePolicy`** -(`maxReceiveCount` ≥ `max_attempts`, pointing at `.dlq`) on every SQS +(`maxReceiveCount` ≥ `max_attempts`, pointing at `-dlq`, the SQS default DLQ name: FIFO `-dlq.fifo`, because an SQS queue name cannot contain a dot; `dead_letter_queue` overrides it, and `dead_letter_suffix` replaces `-dlq`, placed before `.fifo` on a FIFO queue) on every SQS queue as the broker-side backstop, and set a non-zero `retry_backoff` if the handler's dependencies need time to recover. diff --git a/src/babelqueue/__init__.py b/src/babelqueue/__init__.py index b162189..5670ff1 100644 --- a/src/babelqueue/__init__.py +++ b/src/babelqueue/__init__.py @@ -32,6 +32,7 @@ from .transport import ( Abandoner, BatchPopper, + DeadLetterNamer, DeadLetterWriter, HeaderPublisher, InMemoryTransport, @@ -59,6 +60,7 @@ "ReceivedMessage", "Abandoner", "BatchPopper", + "DeadLetterNamer", "DeadLetterWriter", "HeaderPublisher", "NativeSettler", diff --git a/src/babelqueue/app.py b/src/babelqueue/app.py index cf255fa..1678b86 100644 --- a/src/babelqueue/app.py +++ b/src/babelqueue/app.py @@ -44,6 +44,7 @@ def on_order_created(data, meta): from .transport import ( NON_NUMERIC_ATTEMPTS_ROUTE, Abandoner, + DeadLetterNamer, DeadLetterWriter, HeaderPublisher, NativeSettler, @@ -111,7 +112,7 @@ def __init__( max_attempts: int = 3, dead_letter: bool = False, dead_letter_queue: Optional[str] = None, - dead_letter_suffix: str = ".dlq", + dead_letter_suffix: Optional[str] = None, retry_backoff: float = 0.0, retry_backoff_multiplier: float = 1.0, retry_backoff_max: float = 0.0, @@ -145,6 +146,8 @@ def __init__( self.max_attempts = max_attempts self.dead_letter_enabled = bool(dead_letter) self.dead_letter_queue = dead_letter_queue + #: ``None`` = not set: the transport's :class:`~babelqueue.transport.DeadLetterNamer` default + #: (SQS: ``-dlq``) or else ``.dlq``. An explicit value is handed to the namer. self.dead_letter_suffix = dead_letter_suffix self.retry_backoff = float(retry_backoff) if retry_jitter not in RETRY_JITTERS: @@ -837,8 +840,15 @@ def _settle_transport_poison( ) def _dead_letter_target(self, received: ReceivedMessage) -> str: - """The queue a dead-lettered delivery is written to (configured DLQ or ````).""" - return self.dead_letter_queue or (received.queue + self.dead_letter_suffix) + """The queue a dead-lettered delivery is written to: the configured ``dead_letter_queue``, else + the transport's :class:`~babelqueue.transport.DeadLetterNamer` name (SQS: ``-dlq``, FIFO + ``-dlq.fifo``, an explicit ``dead_letter_suffix`` placed before ``.fifo``; ADR-0037 + decision 7), else ```` (default ``.dlq``).""" + if self.dead_letter_queue: + return self.dead_letter_queue + if isinstance(self.transport, DeadLetterNamer): + return self.transport.dead_letter_queue_name(received.queue, self.dead_letter_suffix) + return received.queue + (".dlq" if self.dead_letter_suffix is None else self.dead_letter_suffix) def _settle_version_gated(self, received: ReceivedMessage, always: bool = False) -> None: """Dead-letter a version-gated delivery directly (no retry). With no DLQ set it is left to the diff --git a/src/babelqueue/sqs_transport.py b/src/babelqueue/sqs_transport.py index aec9227..fa234f2 100644 --- a/src/babelqueue/sqs_transport.py +++ b/src/babelqueue/sqs_transport.py @@ -369,6 +369,19 @@ def ack(self, message: ReceivedMessage) -> None: QueueUrl=self._resolve_url(message.queue), ReceiptHandle=message.handle ) + def dead_letter_queue_name(self, queue: str, suffix: Optional[str] = None) -> str: + """:class:`~babelqueue.transport.DeadLetterNamer`: the BabelQueue DLQ of an SQS queue is + ``-dlq`` by default and, for a FIFO queue, ``-dlq.fifo``. An explicit ``suffix`` + (``dead_letter_suffix``; ``None``/empty = none) replaces ``-dlq`` and is placed before + ``.fifo`` as well: an SQS FIFO queue name must end in ``.fifo``, so ``orders.fifo`` + ``-x`` is + ``orders-x.fifo``, never ``orders.fifo-x``. An SQS queue name may hold only alphanumerics, + hyphens and underscores (plus that one ``.fifo`` suffix), so the runtime-wide ``.dlq`` + cannot be created on SQS (ADR-0037 decision 7). An explicit ``dead_letter_queue`` overrides it.""" + suffix = suffix or "-dlq" + if queue.endswith(".fifo"): + return queue[: -len(".fifo")] + suffix + ".fifo" + return queue + suffix + def settle_native(self, message: ReceivedMessage, reason: str, description: str) -> None: """Leave ``message`` to SQS itself (:class:`~babelqueue.transport.NativeSettler`, ADR-0037 decisions 1 and 2): a version-gated delivery or a poison body that the runtime may neither diff --git a/src/babelqueue/transport.py b/src/babelqueue/transport.py index c3c0a34..0eae6cb 100644 --- a/src/babelqueue/transport.py +++ b/src/babelqueue/transport.py @@ -229,6 +229,21 @@ def rewind(self, message: ReceivedMessage) -> None: ... +@runtime_checkable +class DeadLetterNamer(Protocol): + """Optional :class:`Transport` capability: the name of the BabelQueue dead-letter queue for a + source queue, for a broker whose queue-name rules the runtime-wide ```` shape cannot + satisfy. The runtime asks it whenever ``dead_letter_queue`` is not set, passing ``dead_letter_suffix`` + (``None`` when not set). Amazon SQS implements it (a queue name cannot contain a dot and a FIFO name + must end in ``.fifo``): the default is ``-dlq``, FIFO ``-dlq.fifo``, and an explicit + suffix goes before ``.fifo`` too (``orders.fifo`` + ``-x`` -> ``orders-x.fifo``), never after it + (ADR-0037 decision 7). A transport without it keeps ```` with the default ``.dlq``.""" + + def dead_letter_queue_name(self, queue: str, suffix: Optional[str] = None) -> str: + """The dead-letter queue name for ``queue`` (``suffix`` is the explicit one, or ``None``).""" + ... + + @runtime_checkable class Abandoner(Protocol): """Optional :class:`Transport` capability: hand an unsettled *version-gated* delivery back to the diff --git a/tests/test_poison_routing.py b/tests/test_poison_routing.py index 6b9ad03..8d4d06e 100644 --- a/tests/test_poison_routing.py +++ b/tests/test_poison_routing.py @@ -212,7 +212,7 @@ def test_a_poison_message_is_dead_lettered_with_the_original_body_and_deleted(se 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"]) + 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 diff --git a/tests/test_sqs_conformance.py b/tests/test_sqs_conformance.py index c5e66a8..9af77e7 100644 --- a/tests/test_sqs_conformance.py +++ b/tests/test_sqs_conformance.py @@ -86,7 +86,7 @@ def test_schema_version_gate(self) -> None: class _DlqFailsSQS(FakeSQS): - """FakeSQS whose ``SendMessage`` fails (the raw record cannot be written to ``.dlq``).""" + """FakeSQS whose ``SendMessage`` fails (the raw record cannot be written to ``-dlq``).""" def send_message(self, **kw): raise RuntimeError("dlq unavailable") @@ -130,7 +130,7 @@ def _eval_version_gate_sqs(case: Mapping[str, Any]) -> None: _ensure(calls == [], f"{label}: a handler ran for a version-gated delivery") _ensure(fake.visibility_changes == [], f"{label}: a gated delivery is never released") _ensure( - all(sent["QueueUrl"] == "http://fake/orders.dlq" for sent in fake.sent), + all(sent["QueueUrl"] == "http://fake/orders-dlq" for sent in fake.sent), f"{label}: only the DLQ may be written", ) outcome = expect["outcome"] diff --git a/tests/test_sqs_transport.py b/tests/test_sqs_transport.py index 7db5347..113dc06 100644 --- a/tests/test_sqs_transport.py +++ b/tests/test_sqs_transport.py @@ -17,7 +17,7 @@ from babelqueue import BabelQueue, EnvelopeCodec, UnknownUrnStrategy from babelqueue.sqs_transport import SqsTransport -from babelqueue.transport import HeaderPublisher, ReceivedMessage, Redeliverer, make_transport +from babelqueue.transport import DeadLetterNamer, HeaderPublisher, ReceivedMessage, Redeliverer, make_transport class FakeSQS: @@ -465,7 +465,7 @@ def _on(data, meta): app.publish("urn:babel:orders:created", {"order_id": 1}) with self.assertLogs("babelqueue.app", level="ERROR"): app.consume("orders", max_messages=1, timeout=0) - dlq = [m for m in fake.sent if m["QueueUrl"].endswith("orders.dlq")] + dlq = [m for m in fake.sent if m["QueueUrl"].endswith("orders-dlq")] self.assertEqual(len(dlq), 1) # dead-lettered exactly once self.assertEqual(fake.visibility_changes, []) # and not released on top of that @@ -654,7 +654,7 @@ def _on(data, meta): self.assertEqual(fake.deleted, ["rh-gate"]) self.assertEqual(fake.visibility_changes, []) # not released for a retry self.assertEqual(len(fake.sent), 1) - self.assertEqual(fake.sent[0]["QueueUrl"], "http://fake/orders.dlq") + self.assertEqual(fake.sent[0]["QueueUrl"], "http://fake/orders-dlq") # The gate rejects without decoding: the DLQ record is the incoming body, # byte-identical, with no dead_letter block. self.assertEqual(fake.sent[0]["MessageBody"], incoming) @@ -793,3 +793,62 @@ def test_traceparent_round_trips_on_message_attributes(self): if __name__ == "__main__": # pragma: no cover unittest.main() + + +class SqsDeadLetterNameTest(unittest.TestCase): + """ADR-0037 decision 7: the default BabelQueue DLQ of an SQS queue is ``-dlq``, FIFO + ``-dlq.fifo`` (an SQS queue name cannot contain a dot). An explicit ``dead_letter_queue`` or + ``dead_letter_suffix`` always wins; every other transport keeps ``.dlq``.""" + + def test_the_capability_names_the_default(self): + tr = SqsTransport("sqs://", client=FakeSQS(), queue_url_prefix="http://fake") + self.assertIsInstance(tr, DeadLetterNamer) + for queue, suffix, want in ( + ("orders", None, "orders-dlq"), + ("orders_v2", None, "orders_v2-dlq"), + ("orders.fifo", None, "orders-dlq.fifo"), + ("my-orders.fifo", None, "my-orders-dlq.fifo"), + ("fifo", None, "fifo-dlq"), + ("orders", "-x", "orders-x"), + ("orders", "_failed", "orders_failed"), + ("orders.fifo", "-x", "orders-x.fifo"), + ("orders.fifo", "-dlq", "orders-dlq.fifo"), + ("orders.fifo", "_failed", "orders_failed.fifo"), + ): + with self.subTest(queue=queue, suffix=suffix): + self.assertEqual(tr.dead_letter_queue_name(queue, suffix), want) + + def _dead_letter_target(self, queue: str, **kwargs) -> str: + fake = FakeSQS() + tr = SqsTransport("sqs://", client=fake, queue_url_prefix="http://fake", fifo=queue.endswith(".fifo")) + app = BabelQueue(transport=tr, queue=queue, max_attempts=3, dead_letter=True, **kwargs) + fake.seed("http://fake/" + queue, "not json at all", 1) + app.consume(queue, max_messages=1, timeout=0, handle_signals=False) + self.assertEqual(len(fake.sent), 1) + return fake.sent[0]["QueueUrl"] + + def test_default_and_fifo(self): + self.assertEqual(self._dead_letter_target("orders"), "http://fake/orders-dlq") + self.assertEqual(self._dead_letter_target("orders.fifo"), "http://fake/orders-dlq.fifo") + + def test_an_explicit_suffix_or_queue_wins(self): + self.assertEqual(self._dead_letter_target("orders", dead_letter_suffix="_failed"), "http://fake/orders_failed") + self.assertEqual(self._dead_letter_target("orders", dead_letter_suffix=".dlq"), "http://fake/orders.dlq") + # A FIFO name must end in ".fifo": an explicit suffix goes before it, never after it. + self.assertEqual( + self._dead_letter_target("orders.fifo", dead_letter_suffix="-x"), "http://fake/orders-x.fifo" + ) + self.assertEqual( + self._dead_letter_target("orders.fifo", dead_letter_suffix="-dlq"), "http://fake/orders-dlq.fifo" + ) + self.assertEqual(self._dead_letter_target("orders", dead_letter_queue="parking-lot"), "http://fake/parking-lot") + self.assertEqual( + self._dead_letter_target("orders", dead_letter_suffix="-x", dead_letter_queue="parking-lot"), + "http://fake/parking-lot", + ) + + def test_other_transports_keep_dot_dlq(self): + app = BabelQueue("memory://", queue="orders", dead_letter=True) + app.transport.publish("orders", "not json at all") + app.consume("orders", max_messages=1, timeout=0, handle_signals=False) + self.assertEqual(app.transport.size("orders.dlq"), 1) From 3897503cd0b13c805dd3ee6d53ff7f875283f091 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Muhammet=20=C5=9Eafak?= Date: Mon, 26 Oct 2026 08:30:29 +0300 Subject: [PATCH 3/3] Release v1.24.0 --- CHANGELOG.md | 5 +++-- pyproject.toml | 2 +- src/babelqueue/__init__.py | 2 +- 3 files changed, 5 insertions(+), 4 deletions(-) diff --git a/CHANGELOG.md b/CHANGELOG.md index 40eabc9..7d6f769 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -9,12 +9,13 @@ The envelope wire format is versioned separately by `meta.schema_version` ## [Unreleased] +## [1.24.0] - 2026-10-26 + Implements conformance **1.8.0** (vendored from the `v1.8.0` release asset, verified with `fetch.sh --check`; `BQ_CONFORMANCE_VERSION` is now `1.8.0` in `ci.yml` and `release.yml`). The envelope stays at `schema_version: 1`; nothing here touches the wire (ADR-0037). The new `version_gate_pulsar`, `version_gate_artemis`, `version_gate_sqs` and `version_gate_asb` sections run -and pass (17 cases each; Python has the `on_unknown_urn` option, so no row is skipped). A MINOR; the -version number is set at release time. +and pass (17 cases each; Python has the `on_unknown_urn` option, so no row is skipped). A MINOR. ### Changed (behaviour) - **With the DLQ off, SQS and Azure Service Bus no longer drop a version-gated message or retry a diff --git a/pyproject.toml b/pyproject.toml index 842b4f5..f9af87c 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -4,7 +4,7 @@ build-backend = "hatchling.build" [project] name = "babelqueue" -version = "1.23.0" +version = "1.24.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 5670ff1..8768097 100644 --- a/src/babelqueue/__init__.py +++ b/src/babelqueue/__init__.py @@ -45,7 +45,7 @@ Transport, ) -__version__ = "1.23.0" +__version__ = "1.24.0" __all__ = [ "BabelQueue",