Skip to content

Commit 69925bd

Browse files
committed
fix(runtime): gate bq-schema-version before decode on SQS/ASB/Kafka/Artemis
Check the transport schema-version property (SQS/ASB bq-schema-version, Kafka header, Artemis bq_schema_version) before the body is decoded, per broker-bindings §3.7/§4.7/§6.7/§7.7. A raw "1" decodes, a missing or ASCII-blank value keeps the existing path, anything else is rejected undecoded and dead-lettered directly without retries. ASB accepts integer 1 or the string "1" and rejects floats; rejected values are echoed to at most 64 characters. When the dead-letter write fails the message is never acked or handed to a handler: Kafka seeks back to the failed offset via the optional Rewinder protocol so a later cumulative commit cannot skip it.
1 parent d5e1e95 commit 69925bd

20 files changed

Lines changed: 1363 additions & 53 deletions

‎CHANGELOG.md‎

Lines changed: 29 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -9,6 +9,35 @@ The envelope wire format is versioned separately by `meta.schema_version`
99

1010
## [Unreleased]
1111

12+
### Fixed
13+
- **`bq-schema-version` pre-decode gate on the SQS, Azure Service Bus, Kafka and Artemis
14+
transports** (`pop`). The broker's schema-version property is read **before** the body is
15+
decoded (broker-bindings §3.7 / §4.7 / §6.7 / §7.7), mirroring the existing Pulsar gate:
16+
a missing or blank value (empty or only ASCII whitespace) takes the normal path; anything
17+
else is compared untrimmed to `"1"` and a mismatch is returned with the raw body and the
18+
reserved route `babelqueue:unsupported-schema-version:<v>` (version percent-escaped) without
19+
decoding it. SQS/ASB/Kafka read `bq-schema-version`, Artemis `bq_schema_version`
20+
(ADR-0017); ASB and Artemis also accept an integer `1` (a `bool` is not `1`). The conformance suite now reads each binding's
21+
`schema_version_gate` table (including Pulsar's). No wire change.
22+
- **A version-gated delivery is dead-lettered directly, never retried.** `BabelQueue` now
23+
sends a message under the reserved gate route straight to the DLQ (reason `unknown_urn`,
24+
body untouched) when `dead_letter` is enabled and drops it otherwise, regardless of
25+
`max_attempts` / `on_unknown_urn`. A retry would re-publish through paths that rebuild the
26+
version header from the (valid) body (ASB/Kafka/Artemis) or never advance the attempt
27+
counter (SQS), letting the gate be bypassed or the message loop. Applies to Pulsar too
28+
(previously it retried up to `max_attempts` first).
29+
If the dead-letter write itself fails, the message is still never retried: it is left
30+
unacknowledged (an error is logged) so the broker redelivers it and the gate rejects it again
31+
(SQS at the visibility timeout, Azure Service Bus at lock expiry, Artemis when the link or
32+
connection closes). Kafka commits offsets cumulatively, so merely not acknowledging the
33+
record would lose it once a later record is committed: `KafkaTransport` now `seek`s the
34+
partition back to the failed record's offset so it is polled again and the dead-letter write
35+
is retried (no commit moves past it).
36+
- New optional `Rewinder` protocol (`rewind(message)`), implemented by `KafkaTransport` and
37+
exported from `babelqueue`; the runtime calls it after a failed version-gate dead-letter write
38+
and then pauses briefly (1 s) before the next poll.
39+
The rejected version is capped at 64 characters plus `...` in the log and the route.
40+
1241
## [1.15.0] - 2026-10-10
1342

1443
### Added

‎README.md‎

Lines changed: 37 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -124,6 +124,43 @@ app.run() # consume forever (Ctrl-C to stop)
124124
the message unchanged once `shutdown_timeout` expires. Delivery is
125125
**at-least-once** — keep handlers idempotent (see the idempotency helper).
126126

127+
### Unsupported `schema_version` (SQS, Azure Service Bus, Kafka, Artemis, Pulsar)
128+
129+
These transports read the binding's schema-version property (`bq-schema-version`; Artemis
130+
`bq_schema_version`) **before** decoding the body. A missing or blank value (empty or only
131+
ASCII whitespace) follows the normal path; any other value must be exactly `"1"` (compared
132+
untrimmed; Azure Service Bus and Artemis also accept the integer `1`). A message carrying
133+
anything else is not decoded: it reaches the runtime on the reserved route
134+
`babelqueue:unsupported-schema-version:<v>` (the version percent-escaped) and is **never
135+
retried** — it goes straight to the dead-letter queue (reason `unknown_urn`, the body kept
136+
verbatim) when `dead_letter` is enabled, and is dropped otherwise, whatever `max_attempts` is.
137+
It does not depend on a broker `RedrivePolicy`. If the dead-letter write fails the message is
138+
left unacknowledged (an error is logged), is never retried and never reaches a handler; the
139+
dead-letter write is attempted again when the broker brings it back, which depends on the
140+
broker:
141+
142+
- **SQS** — redelivered when the visibility timeout expires.
143+
- **Azure Service Bus** — redelivered when the message lock expires.
144+
- **Artemis** — an unsettled AMQP 1.0 delivery is returned only when the link or connection
145+
closes, so it stays unsettled until then.
146+
- **Kafka** — offsets are committed cumulatively per partition, so merely skipping the commit
147+
would lose the record once a later one is committed. The transport therefore implements
148+
the optional `Rewinder` protocol: it `seek`s the partition back to the failed record's
149+
offset and pauses `_rewind_pause` (1 s by default, so an unavailable DLQ is not hammered)
150+
before the record is polled again; the gate rejects it again and the dead-letter write
151+
is retried. While the dead-letter write keeps failing (for example the `.dlq` topic is not
152+
writable) the partition **stalls on that record** (head-of-line blocking, consumer lag
153+
grows) and the worker waits the pause on every attempt, which also delays its other
154+
partitions and queues. The commit does not move past the record as long as the seek
155+
succeeds; if the seek itself fails, an error naming the topic, partition and offset is
156+
logged and a later commit on that partition may skip the record, so recover it by hand from
157+
that offset. If the partition was lost to a rebalance, the new owner resumes from the last
158+
committed offset and the record is not lost.
159+
- **Pulsar** — acknowledgement is per message; the unacknowledged message is redelivered by
160+
the broker.
161+
162+
A warning naming the version (cut at 64 characters plus `...`) and queue is logged.
163+
127164
### SQS release and poison messages
128165

129166
On a handler failure, unknown-URN `release` or shutdown release, the SQS

‎src/babelqueue/__init__.py‎

Lines changed: 2 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -34,6 +34,7 @@
3434
InMemoryTransport,
3535
ReceivedMessage,
3636
Redeliverer,
37+
Rewinder,
3738
RoutedPublisher,
3839
Transport,
3940
)
@@ -53,6 +54,7 @@
5354
"ReceivedMessage",
5455
"HeaderPublisher",
5556
"Redeliverer",
57+
"Rewinder",
5658
"RoutedPublisher",
5759
"BabelQueueError",
5860
"UnknownUrnError",

‎src/babelqueue/app.py‎

Lines changed: 58 additions & 12 deletions
Original file line numberDiff line numberDiff line change
@@ -43,6 +43,7 @@ def on_order_created(data, meta):
4343
HeaderPublisher,
4444
ReceivedMessage,
4545
Redeliverer,
46+
Rewinder,
4647
Transport,
4748
make_transport,
4849
publish_keeping_route,
@@ -111,6 +112,7 @@ def __init__(
111112
self.shutdown_timeout = float(shutdown_timeout)
112113
self._handlers: Dict[str, Handler] = {}
113114
self._stop = threading.Event()
115+
self._rewind_pause = 1.0 # seconds to wait before a rewound gated message is re-polled
114116
self._phase = _IDLE
115117
self._last_phase = _IDLE
116118
self._signalled = False
@@ -295,6 +297,20 @@ def dispatch(self, received: ReceivedMessage) -> None:
295297
self._phase = _SETTLING
296298
except Exception as exc: # noqa: BLE001 - one bad message must not kill the loop
297299
self._phase = _SETTLING
300+
if received.route_urn.startswith(_UNSUPPORTED_SCHEMA_ROUTE_PREFIX):
301+
# A version-gated delivery never takes the generic retry path, even when its
302+
# dead-letter write fails: a retry re-publish may rebuild the version header
303+
# from the (valid) body and walk the message past the gate into a handler.
304+
# It is left unacked; the broker (or a rewind, for cumulative-offset brokers
305+
# such as Kafka) brings it back and the gate rejects it again.
306+
logger.error(
307+
"Dead-lettering a version-gated message on queue %r failed; it was NOT "
308+
"retried or acknowledged; it will come back and the dead-letter write will be retried.",
309+
received.queue,
310+
exc_info=exc,
311+
)
312+
self._rewind_unsettled(received)
313+
return
298314
self._retry_or_dead_letter(received, envelope, exc)
299315
else:
300316
# Outside the handler's try: a failed ack is not a handler failure, so it must
@@ -417,6 +433,23 @@ def _ack_settled(self, received: ReceivedMessage) -> None:
417433
exc_info=True,
418434
)
419435

436+
def _rewind_unsettled(self, received: ReceivedMessage) -> None:
437+
"""Make a deliberately unsettled message come back (cumulative-offset brokers such as Kafka
438+
would otherwise lose it when a later record on the partition is committed), after a short
439+
pause so an unavailable DLQ is not hammered in a hot loop."""
440+
if not isinstance(self.transport, Rewinder):
441+
return
442+
try:
443+
self.transport.rewind(received)
444+
except Exception: # noqa: BLE001 - reported, never kills the loop
445+
logger.error(
446+
"Failed to rewind an unsettled version-gated message on queue %r.",
447+
received.queue,
448+
exc_info=True,
449+
)
450+
return
451+
self._stop.wait(self._rewind_pause)
452+
420453
def _release_unprocessed(self, received: ReceivedMessage) -> None:
421454
"""Release a message whose handler never completed (shutdown): unchanged body, no
422455
backoff, and the runtime adds no ``attempts`` increment. A broker that counts
@@ -440,23 +473,35 @@ def _route_unknown(self, urn: str, received: ReceivedMessage, envelope: Mapping[
440473
if strategy == UnknownUrnStrategy.DELETE:
441474
self._ack_settled(received)
442475
return
443-
# A transport's version-gated delivery, an undecodable body or a body naming no URN can
444-
# never gain a handler, so releasing it unchanged would redeliver it forever. It takes the
445-
# bounded FAIL path instead (retry up to max_attempts, then dead-letter or drop); only a
446-
# genuinely unknown URN keeps the verbatim release.
447-
gated = received.route_urn.startswith(_UNSUPPORTED_SCHEMA_ROUTE_PREFIX)
448-
poison = gated or not urn
449-
if strategy == UnknownUrnStrategy.RELEASE and not poison:
476+
# A transport's version-gated delivery (unsupported ``bq-schema-version``) can never gain
477+
# a handler and must not be retried: a retry re-publishes through paths that may rebuild
478+
# the version header from the (valid) body, or never advance the attempt counter, so the
479+
# gate would be bypassed or the message would loop. It goes straight to the DLQ (or is
480+
# dropped when no DLQ is configured).
481+
if received.route_urn.startswith(_UNSUPPORTED_SCHEMA_ROUTE_PREFIX):
482+
self._settle_version_gated(received, envelope)
483+
return
484+
# An undecodable body or a body naming no URN can never gain a handler either, so
485+
# releasing it unchanged would redeliver it forever. It takes the bounded FAIL path
486+
# instead (retry up to max_attempts, then dead-letter or drop); only a genuinely unknown
487+
# URN keeps the verbatim release.
488+
if strategy == UnknownUrnStrategy.RELEASE and urn:
450489
self._release(received, received.body, self.unknown_urn_release_delay)
451490
return
452491
if strategy == UnknownUrnStrategy.DEAD_LETTER:
453492
self._dead_letter(received, dict(envelope), "unknown_urn", None)
454493
return
455494
# FAIL (and RELEASE of a poison message) — surfaced through the retry/dead-letter path
456-
# (never kills the loop). A gated delivery names its reserved route (the rejected version).
457-
raise UnknownUrnError(
458-
f"No handler mapped for URN [{urn or (received.route_urn if gated else '') or '(empty)'}]."
459-
)
495+
# (never kills the loop).
496+
raise UnknownUrnError(f"No handler mapped for URN [{urn or '(empty)'}].")
497+
498+
def _settle_version_gated(self, received: ReceivedMessage, envelope: Mapping[str, Any]) -> None:
499+
"""Dead-letter a version-gated delivery directly (no retry); drop it when no DLQ is set."""
500+
if self.dead_letter_enabled or self.on_unknown_urn == UnknownUrnStrategy.DEAD_LETTER:
501+
error = UnknownUrnError(f"No handler mapped for URN [{received.route_urn}].")
502+
self._dead_letter(received, dict(envelope), "unknown_urn", error)
503+
return
504+
self._ack_settled(received)
460505

461506
def _retry_or_dead_letter(
462507
self, received: ReceivedMessage, envelope: Dict[str, Any], exc: BaseException
@@ -500,7 +545,8 @@ def _dead_letter(
500545
#: Reserved ``ReceivedMessage.route_urn`` prefix a transport uses to hand over a delivery it
501546
#: version-gated (broker-bindings §5.7 step 3, e.g. the pulsar transport's ``bq-schema-version``
502547
#: check). Not a ``urn:babel:`` URN, so no handler owns it. Must equal
503-
#: ``pulsar_transport.UNSUPPORTED_SCHEMA_ROUTE_PREFIX`` (locked by a test); never reaches the wire.
548+
#: ``UNSUPPORTED_SCHEMA_ROUTE_PREFIX`` of the pulsar, sqs, asb, kafka and artemis transports (locked
549+
#: by a test); never reaches the wire. A delivery under it is dead-lettered directly, never retried.
504550
_UNSUPPORTED_SCHEMA_ROUTE_PREFIX = "babelqueue:unsupported-schema-version:"
505551

506552

‎src/babelqueue/artemis_transport.py‎

Lines changed: 63 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -32,10 +32,24 @@
3232

3333
from __future__ import annotations
3434

35+
import logging
36+
import re
3537
from typing import Any, Dict, Optional
38+
from urllib.parse import quote
3639

37-
from .codec import EnvelopeCodec, parse_envelope
38-
from .transport import ReceivedMessage, Transport
40+
from .codec import SCHEMA_VERSION, EnvelopeCodec, parse_envelope
41+
from .transport import ReceivedMessage, Transport, truncate_declared_version
42+
43+
_logger = logging.getLogger("babelqueue.artemis")
44+
45+
#: A blank ``bq_schema_version``: empty or only ASCII whitespace (space, \t, \n, \v, \f, \r) — the
46+
#: same set in every SDK (GR-5). Deliberately not ``str.strip()``, which also strips U+001C-U+001F,
47+
#: U+0085, U+00A0, U+2028, U+3000 ...
48+
_BLANK_ASCII_PATTERN = re.compile(r"[ \t\n\x0b\x0c\r]*")
49+
#: Prefix of the reserved ``route_urn`` a version-gated message is delivered under. It is not a
50+
#: ``urn:babel:`` URN, so no handler can legitimately own it; the runtime reports it as an
51+
#: unknown URN, which names the rejected ``bq_schema_version``.
52+
UNSUPPORTED_SCHEMA_ROUTE_PREFIX = "babelqueue:unsupported-schema-version:"
3953

4054
JMS_TYPE_ANNOTATION = "x-opt-jms-type"
4155
APP_ID = "babelqueue"
@@ -187,6 +201,33 @@ def _delivery_count(message: Any) -> int:
187201
except (TypeError, ValueError): # pragma: no cover - defensive
188202
return 0
189203

204+
@staticmethod
205+
def _unsupported_schema_version(properties: Any) -> Optional[str]:
206+
"""The §7.7 step-3 gate on the ``bq_schema_version`` application property (underscores —
207+
ADR-0017): the offending raw value, or ``None``.
208+
209+
A missing property, a null value or a blank value (the empty string or only ASCII
210+
whitespace — the same set in every SDK, GR-5, not ``str.strip()``) is not gated: the usual
211+
decode path runs. A ``bytes`` value is decoded as UTF-8. A present value is accepted only
212+
when it is exactly the core's supported ``SCHEMA_VERSION``: the string ``"1"`` or the
213+
integer ``1`` (a ``bool`` is not an integer here, though it is one in Python); ``"2"``,
214+
``"x"``, ``"01"``, ``" 1"``, ``0``, ``2``, ``1.5``, ``True`` and any other type are
215+
rejected."""
216+
if not isinstance(properties, dict):
217+
return None
218+
raw = properties.get("bq_schema_version")
219+
if raw is None:
220+
return None
221+
if isinstance(raw, (bytes, bytearray)):
222+
raw = bytes(raw).decode("utf-8", "replace")
223+
if isinstance(raw, str):
224+
if _BLANK_ASCII_PATTERN.fullmatch(raw) is not None or raw == str(SCHEMA_VERSION):
225+
return None
226+
return raw
227+
if isinstance(raw, int) and not isinstance(raw, bool) and raw == SCHEMA_VERSION:
228+
return None
229+
return str(raw)
230+
190231
@staticmethod
191232
def _payload(message: Any) -> str:
192233
body = getattr(message, "body", None)
@@ -212,6 +253,26 @@ def pop(self, queue: str, timeout: float = 1.0) -> Optional[ReceivedMessage]:
212253
raise
213254
if message is None: # pragma: no cover - defensive (real client raises instead)
214255
return None
256+
# §7.7 step 3: gate on the ``bq_schema_version`` property BEFORE any body decode (the
257+
# attempts reconcile below decodes the body). A rejected message keeps its raw body and the
258+
# reserved route; the runtime then dead-letters it directly (no retry).
259+
unsupported = truncate_declared_version(
260+
self._unsupported_schema_version(getattr(message, "properties", None))
261+
)
262+
if unsupported is not None:
263+
_logger.warning(
264+
"Artemis message on %r carries unsupported bq_schema_version %r (supported: %s); "
265+
"not decoding it.",
266+
queue,
267+
unsupported,
268+
SCHEMA_VERSION,
269+
)
270+
return ReceivedMessage(
271+
body=self._payload(message),
272+
queue=queue,
273+
handle=receiver,
274+
route_urn=UNSUPPORTED_SCHEMA_ROUTE_PREFIX + quote(unsupported, safe=""),
275+
)
215276
body = self._reconcile(self._payload(message), self._delivery_count(message))
216277
return ReceivedMessage(body=body, queue=queue, handle=receiver)
217278

0 commit comments

Comments
 (0)