fix(gate): forward the raw body to the DLQ on a schema-version rejection

A delivery rejected by the bq-schema-version gate is now dead-lettered as
its raw body, byte-identical to the incoming one, with no dead_letter block
and without ever being decoded. This matches the Java, .NET and Node
consumers and the reject-without-decoding rule of contract sections
3.7/4.7/5.7/6.7/7.7. A failed DLQ write still leaves the message
unacknowledged for redelivery.

Author: muhammetsafak

CHANGELOG.md

@@ -9,6 +9,16 @@ The envelope wire format is versioned separately by `meta.schema_version`
 
 ## [Unreleased]
 
+### Changed
+- **Schema-version gate rejection now dead-letters the incoming body, byte-identical.** A message
+  whose `bq-schema-version` (`bq_schema_version` on Artemis) is unsupported is rejected *without
+  decoding it* (broker-bindings §3.7/§4.7/§5.7/§6.7/§7.7) and, when the DLQ is enabled, the DLQ
+  record is the raw body as received, with no `dead_letter` block. Previously the core decoded
+  the envelope and wrote an annotated `dead_letter{reason:"unknown_urn"}` record. This aligns the
+  Python SDK with Java, .NET and Node. A DLQ consumer that expected the annotation for gated
+  messages is affected; a failed DLQ write still leaves the message unsettled (rewound on Kafka),
+  and DLQ off / `UnknownUrnStrategy.DELETE` still drops it. No wire or API change.
+
 ### Fixed
 - **Kafka: a record whose settlement raised was silently lost on a consume-loop restart.** When a
   retry / release / dead-letter `publish` failed (or the loop was otherwise interrupted before

README.md

@@ -132,8 +132,8 @@ ASCII whitespace) follows the normal path; any other value must be exactly `"1"`
 untrimmed; Azure Service Bus and Artemis also accept the integer `1`). A message carrying
 anything else is not decoded: it reaches the runtime on the reserved route
 `babelqueue:unsupported-schema-version:<v>` (the version percent-escaped) and is **never
-retried** — it goes straight to the dead-letter queue (reason `unknown_urn`, the body kept
-verbatim) when `dead_letter` is enabled, and is dropped otherwise, whatever `max_attempts` is.
+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
 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

src/babelqueue/app.py

@@ -279,7 +279,11 @@ def dispatch(self, received: ReceivedMessage) -> None:
         with _headers_scope(received.headers), _replay_scope(
             bool(received.headers.get(HEADER_REPLAY_BYPASS))
         ):
-            envelope = EnvelopeCodec.decode(received.body)
+            # A version-gated delivery is rejected without decoding its body (broker-bindings
+            # §3.7/§4.7/§5.7/§6.7/§7.7): the raw body goes to the DLQ byte-identical, so there is
+            # no envelope to read here.
+            gated = received.route_urn.startswith(_UNSUPPORTED_SCHEMA_ROUTE_PREFIX)
+            envelope = {} if gated else EnvelopeCodec.decode(received.body)
             urn = str(envelope.get("job") or envelope.get("urn") or "")
             if received.route_urn and envelope:
                 # Transport-supplied routing override; body untouched. Only for a body that
@@ -479,7 +483,7 @@ def _route_unknown(self, urn: str, received: ReceivedMessage, envelope: Mapping[
         # gate would be bypassed or the message would loop. It goes straight to the DLQ (or is
         # dropped when no DLQ is configured).
         if received.route_urn.startswith(_UNSUPPORTED_SCHEMA_ROUTE_PREFIX):
-            self._settle_version_gated(received, envelope)
+            self._settle_version_gated(received)
             return
         # An undecodable body or a body naming no URN can never gain a handler either, so
         # releasing it unchanged would redeliver it forever. It takes the bounded FAIL path
@@ -495,11 +499,21 @@ def _route_unknown(self, urn: str, received: ReceivedMessage, envelope: Mapping[
         # (never kills the loop).
         raise UnknownUrnError(f"No handler mapped for URN [{urn or '(empty)'}].")
 
-    def _settle_version_gated(self, received: ReceivedMessage, envelope: Mapping[str, Any]) -> None:
-        """Dead-letter a version-gated delivery directly (no retry); drop it when no DLQ is set."""
+    def _dead_letter_target(self, received: ReceivedMessage) -> str:
+        """The queue a dead-lettered delivery is written to (configured DLQ or ``<queue><suffix>``)."""
+        return self.dead_letter_queue or (received.queue + self.dead_letter_suffix)
+
+    def _settle_version_gated(self, received: ReceivedMessage) -> None:
+        """Dead-letter a version-gated delivery directly (no retry); drop it when no DLQ is set.
+
+        The gate rejects the message *without decoding it*, so the DLQ record is the incoming raw
+        body, byte-identical, with no ``dead_letter`` block (matching the Java, .NET and Node
+        SDKs). A failed DLQ write propagates: the delivery is left unacknowledged and handed back
+        (see :meth:`dispatch`), never retried.
+        """
         if self.dead_letter_enabled or self.on_unknown_urn == UnknownUrnStrategy.DEAD_LETTER:
-            error = UnknownUrnError(f"No handler mapped for URN [{received.route_urn}].")
-            self._dead_letter(received, dict(envelope), "unknown_urn", error)
+            self._publish_for(received, self._dead_letter_target(received), received.body)
+            self._ack_settled(received)
             return
         self._ack_settled(received)
 
@@ -537,8 +551,9 @@ def _dead_letter(
             error=(str(exc) if exc is not None else None),
             exception=(type(exc).__name__ if exc is not None else None),
         )
-        target = self.dead_letter_queue or (received.queue + self.dead_letter_suffix)
-        self._publish_for(received, target, EnvelopeCodec.encode(annotated))
+        self._publish_for(
+            received, self._dead_letter_target(received), EnvelopeCodec.encode(annotated)
+        )
         self._ack_settled(received)
 
 

src/babelqueue/artemis_transport.py

@@ -47,8 +47,9 @@
 #: U+0085, U+00A0, U+2028, U+3000 ...
 _BLANK_ASCII_PATTERN = re.compile(r"[ \t\n\x0b\x0c\r]*")
 #: Prefix of the reserved ``route_urn`` a version-gated message is delivered under. It is not a
-#: ``urn:babel:`` URN, so no handler can legitimately own it; the runtime reports it as an
-#: unknown URN, which names the rejected ``bq_schema_version``.
+#: ``urn:babel:`` URN, so no handler can legitimately own it; the runtime dead-letters it
+#: directly (never retried, never decoded; the DLQ record is the incoming raw body). The
+#: rejected ``bq_schema_version`` is named in the logged error.
 UNSUPPORTED_SCHEMA_ROUTE_PREFIX = "babelqueue:unsupported-schema-version:"
 
 JMS_TYPE_ANNOTATION = "x-opt-jms-type"

src/babelqueue/asb_transport.py

@@ -38,8 +38,9 @@
 #: U+0085, U+00A0, U+2028, U+3000 ...
 _BLANK_ASCII_PATTERN = re.compile(r"[ \t\n\x0b\x0c\r]*")
 #: Prefix of the reserved ``route_urn`` a version-gated message is delivered under. It is not a
-#: ``urn:babel:`` URN, so no handler can legitimately own it; the runtime reports it as an
-#: unknown URN, which names the rejected ``bq-schema-version``.
+#: ``urn:babel:`` URN, so no handler can legitimately own it; the runtime dead-letters it
+#: directly (never retried, never decoded; the DLQ record is the incoming raw body). The
+#: rejected ``bq-schema-version`` is named in the logged error.
 UNSUPPORTED_SCHEMA_ROUTE_PREFIX = "babelqueue:unsupported-schema-version:"
 
 

src/babelqueue/kafka_transport.py

@@ -40,8 +40,9 @@
 #: U+0085, U+00A0, U+2028, U+3000 ...
 _BLANK_ASCII_PATTERN = re.compile(r"[ \t\n\x0b\x0c\r]*")
 #: Prefix of the reserved ``route_urn`` a version-gated message is delivered under. It is not a
-#: ``urn:babel:`` URN, so no handler can legitimately own it; the runtime reports it as an
-#: unknown URN, which names the rejected ``bq-schema-version``.
+#: ``urn:babel:`` URN, so no handler can legitimately own it; the runtime dead-letters it
+#: directly (never retried, never decoded; the DLQ record is the incoming raw body). The
+#: rejected ``bq-schema-version`` is named in the logged error.
 UNSUPPORTED_SCHEMA_ROUTE_PREFIX = "babelqueue:unsupported-schema-version:"
 
 Headers = List[Tuple[str, bytes]]

src/babelqueue/pulsar_transport.py

@@ -59,8 +59,9 @@
 #: ... that fits a signed 32-bit int, matching the Java and .NET consumers (GR-5).
 _MAX_ATTEMPTS = 2**31 - 1
 #: Prefix of the reserved ``route_urn`` a version-gated message is delivered under. It is not a
-#: ``urn:babel:`` URN, so no handler can legitimately own it; the runtime reports it as an
-#: unknown URN, which names the rejected ``bq-schema-version``.
+#: ``urn:babel:`` URN, so no handler can legitimately own it; the runtime dead-letters it
+#: directly (never retried, never decoded; the DLQ record is the incoming raw body). The
+#: rejected ``bq-schema-version`` is named in the logged error.
 UNSUPPORTED_SCHEMA_ROUTE_PREFIX = "babelqueue:unsupported-schema-version:"
 #: Separates the rejected version from the delivery's original ``bq-job`` in a gate route.
 _GATE_JOB_PARAM = "?bq-job="

src/babelqueue/sqs_transport.py

@@ -54,8 +54,9 @@
 #: U+0085, U+00A0, U+2028, U+3000 ...
 _BLANK_ASCII_PATTERN = re.compile(r"[ \t\n\x0b\x0c\r]*")
 #: Prefix of the reserved ``route_urn`` a version-gated message is delivered under. It is not a
-#: ``urn:babel:`` URN, so no handler can legitimately own it; the runtime reports it as an
-#: unknown URN, which names the rejected ``bq-schema-version``.
+#: ``urn:babel:`` URN, so no handler can legitimately own it; the runtime dead-letters it
+#: directly (never retried, never decoded; the DLQ record is the incoming raw body). The
+#: rejected ``bq-schema-version`` is named in the logged error.
 UNSUPPORTED_SCHEMA_ROUTE_PREFIX = "babelqueue:unsupported-schema-version:"
 
 

tests/test_app.py

@@ -265,9 +265,11 @@ def test_version_gated_message_is_dead_lettered_directly_without_retry(self) ->
                     self.assertEqual(seen, [])  # never reaches the handler
                     self.assertEqual(app.transport.size("orders"), 0)
                     self.assertEqual(app.transport.size("orders.dlq"), 1)
-                    dead = EnvelopeCodec.decode(app.transport._queues["orders.dlq"][0])  # noqa: SLF001
-                    self.assertEqual(dead["dead_letter"]["reason"], "unknown_urn")
-                    self.assertIn(route, dead["dead_letter"]["error"])
+                    # The gate rejects without decoding: the DLQ record is the incoming raw body,
+                    # byte-identical, with no dead_letter block (even for an undecodable body).
+                    dead = app.transport._queues["orders.dlq"][0]  # noqa: SLF001
+                    self.assertEqual(dead, body)
+                    self.assertNotIn("dead_letter", dead)
 
     def test_version_gated_message_whose_dead_letter_write_fails_is_not_retried(self) -> None:
         from babelqueue.app import _UNSUPPORTED_SCHEMA_ROUTE_PREFIX

tests/test_artemis_transport.py

@@ -301,7 +301,8 @@ 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&x=%"):
             with self.subTest(value=value):
-                transport, receiver = self._transport(body(), {"bq_schema_version": value})
+                incoming = body()
+                transport, receiver = self._transport(incoming, {"bq_schema_version": value})
                 app = BabelQueue(transport=transport, max_attempts=5, dead_letter=True)
                 seen: list = []
 
@@ -318,10 +319,9 @@ def _on(data, meta):
                 publish.assert_called_once()
                 queue, dead_body = publish.call_args.args
                 self.assertEqual("orders.dlq", queue)
-                dead = EnvelopeCodec.decode(dead_body)
-                self.assertEqual(URN, dead["job"])  # the producer's body, annotated
-                self.assertEqual("unknown_urn", dead["dead_letter"]["reason"])
-                self.assertIn(self.PREFIX + quote(value, safe=""), dead["dead_letter"]["error"])
+                # rejected without decoding: the DLQ record is the incoming body, byte-identical
+                self.assertEqual(incoming, dead_body)
+                self.assertNotIn("dead_letter", dead_body)
 
 
 class ArtemisUrlTest(unittest.TestCase):