Skip to content

Commit 3e737c5

Browse files
committed
fix: gate Pulsar messages on bq-schema-version before decoding
PulsarTransport.pop rejects a delivery whose bq-schema-version property is present, not blank and not the supported version, before the body is decoded (broker-bindings §5.7 step 3). Under on_unknown_urn=RELEASE, gated, undecodable and URN-less messages now take the bounded FAIL path instead of being released forever.
1 parent 7c9ebd3 commit 3e737c5

5 files changed

Lines changed: 351 additions & 7 deletions

File tree

‎CHANGELOG.md‎

Lines changed: 23 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -28,6 +28,29 @@ The envelope wire format is versioned separately by `meta.schema_version`
2828
stays on the poison / unknown-URN path instead of being routed to a handler.
2929

3030
### Fixed
31+
- **`on_unknown_urn=RELEASE` no longer loops a version-gated or poison message forever.**
32+
`BabelQueue.dispatch()` re-published such a message unchanged on every delivery (`attempts` never
33+
grew, so it never reached the DLQ). A delivery a transport version-gated (reserved
34+
`unsupported-schema-version` `route_urn`, e.g. Pulsar `bq-schema-version`), an undecodable body
35+
and a body naming no URN now take the bounded `FAIL` path under `RELEASE` (retry up to
36+
`max_attempts`, then dead-letter or drop), matching the Go core (GR-5). A genuinely unknown URN
37+
still releases verbatim. No public API or wire change. Note: as on the existing `FAIL` path, a
38+
body that is not valid JSON is not carried verbatim through the retry / dead-letter re-publish
39+
(the runtime re-encodes the envelope it could build, so the producer's original bytes are lost);
40+
previously `RELEASE` re-published those bytes unchanged.
41+
- **Pulsar consumer version-gates on `bq-schema-version` (broker-bindings §5.7 step 3).**
42+
`PulsarTransport.pop()` now reads the `bq-schema-version` message property before decoding the
43+
body. A missing or blank value changes nothing; "Blank" is the same set in every SDK: the empty string or only ASCII whitespace (space, `\t`, `\n`, `\v`, `\f`, `\r`); anything else (NBSP, U+001C-U+001F, U+0085, U+FEFF, U+3000, ...) is not blank and is rejected unless it is exactly the supported version. (`str.strip()` is not used: it also strips U+001C-U+001F, U+0085, NBSP, ...) a value other than the supported
44+
`SCHEMA_VERSION` (`"2"`, `"x"`, `"01"`, ...) is not decoded: the delivery carries the reserved
45+
`route_urn` `babelqueue:unsupported-schema-version:<value>`, which no handler owns, so the
46+
runtime quarantines it through the existing unknown-URN retry / dead-letter path (the error
47+
names the rejected version). The body is passed through untouched. No wire change. The route
48+
is an in-process marker only: it also carries the delivery's original `bq-job`, and the retry /
49+
dead-letter re-publish (`publish_routed`) writes that original `bq-job` (or, when the delivery
50+
had none, the body's job) and the original raw `bq-schema-version` back, never the sentinel, so
51+
a redrive meets the same gate. The gate also logs a `babelqueue.pulsar` warning naming the
52+
rejected version and queue, including for a body that is not valid JSON (which the runtime
53+
quarantines as a poison message whose error names the reserved route).
3154
- **Pulsar consumer now follows broker-bindings §5.2 / §5.7.** `PulsarTransport.pop()` routes on
3255
the `bq-job` message property (set on `ReceivedMessage.route_urn`; the body's URN is used only
3356
when the property is absent or blank) and takes `attempts` from the authoritative

‎src/babelqueue/app.py‎

Lines changed: 17 additions & 3 deletions
Original file line numberDiff line numberDiff line change
@@ -440,15 +440,22 @@ def _route_unknown(self, urn: str, received: ReceivedMessage, envelope: Mapping[
440440
if strategy == UnknownUrnStrategy.DELETE:
441441
self._ack_settled(received)
442442
return
443-
if strategy == UnknownUrnStrategy.RELEASE:
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:
444450
self._release(received, received.body, self.unknown_urn_release_delay)
445451
return
446452
if strategy == UnknownUrnStrategy.DEAD_LETTER:
447453
self._dead_letter(received, dict(envelope), "unknown_urn", None)
448454
return
449-
# FAIL — surfaced through the retry/dead-letter path (never kills the loop).
455+
# 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).
450457
raise UnknownUrnError(
451-
f"No handler mapped for URN [{urn or '(empty)'}]."
458+
f"No handler mapped for URN [{urn or (received.route_urn if gated else '') or '(empty)'}]."
452459
)
453460

454461
def _retry_or_dead_letter(
@@ -490,6 +497,13 @@ def _dead_letter(
490497
self._ack_settled(received)
491498

492499

500+
#: Reserved ``ReceivedMessage.route_urn`` prefix a transport uses to hand over a delivery it
501+
#: version-gated (broker-bindings §5.7 step 3, e.g. the pulsar transport's ``bq-schema-version``
502+
#: 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.
504+
_UNSUPPORTED_SCHEMA_ROUTE_PREFIX = "babelqueue:unsupported-schema-version:"
505+
506+
493507
def _without_forbidden_keys(body: str) -> str:
494508
"""``body`` unchanged, unless it carries a forbidden envelope key (message-envelope.md §10,
495509
e.g. from a legacy producer) — then it is re-encoded without it (warned by the codec), so a

‎src/babelqueue/pulsar_transport.py‎

Lines changed: 94 additions & 3 deletions
Original file line numberDiff line numberDiff line change
@@ -19,6 +19,18 @@
1919
the property wins and the optional ``on_routing_mismatch(property_urn, body_urn, envelope)``
2020
hook is called (for an accepted envelope, after attempts are reconciled). A message left un-acked is redelivered (at-least-once).
2121
22+
Before any of that, a ``bq-schema-version`` property that is present but not the core's supported
23+
``SCHEMA_VERSION`` (``"2"``, ``"x"``, ``"01"``, ...) is version-gated (§5.7 step 3): the body is
24+
**not decoded** and the message is handed to the runtime under a reserved ``route_urn`` no
25+
handler owns, so it is quarantined through the existing unknown-URN retry / dead-letter path (the
26+
error names the rejected version, and a ``babelqueue.pulsar`` warning is logged with the version
27+
and queue, also for a body that is not valid JSON). A missing or blank property (empty or only
28+
ASCII whitespace: space, tab, LF, VT, FF, CR - nothing else counts as blank) decodes as usual.
29+
The reserved route is an in-process marker only: it carries the delivery's original ``bq-job``,
30+
and a retry / dead-letter re-publish writes that ``bq-job`` (or, when the delivery had none, the
31+
body's job) and the original raw ``bq-schema-version`` back — never the sentinel — so a redrive
32+
meets the same gate.
33+
2234
This implements §5 of the broker-bindings contract. The envelope is unchanged
2335
(``schema_version`` stays 1); Apache Pulsar is purely additive.
2436
@@ -30,16 +42,51 @@
3042

3143
from __future__ import annotations
3244

45+
import logging
3346
import re
47+
from urllib.parse import quote, unquote
3448
from typing import Any, Callable, Dict, Optional, Tuple
3549

36-
from .codec import EnvelopeCodec, parse_envelope
50+
from .codec import SCHEMA_VERSION, EnvelopeCodec, parse_envelope
3751
from .transport import ReceivedMessage, Transport
3852

3953
#: ``bq-attempts`` is valid only as a run of ASCII digits (no sign, no whitespace) ...
4054
_ATTEMPTS_PATTERN = re.compile(r"[0-9]+")
55+
#: A blank ``bq-schema-version``: empty or only ASCII whitespace (space, \t, \n, \v, \f, \r) — the
56+
#: same set in every SDK (GR-5). Deliberately not ``str.strip()``, which also strips U+001C-U+001F,
57+
#: U+0085, U+00A0, U+2028, U+3000 ...
58+
_BLANK_ASCII_PATTERN = re.compile(r"[ \t\n\x0b\x0c\r]*")
4159
#: ... that fits a signed 32-bit int, matching the Java and .NET consumers (GR-5).
4260
_MAX_ATTEMPTS = 2**31 - 1
61+
#: Prefix of the reserved ``route_urn`` a version-gated message is delivered under. It is not a
62+
#: ``urn:babel:`` URN, so no handler can legitimately own it; the runtime reports it as an
63+
#: unknown URN, which names the rejected ``bq-schema-version``.
64+
UNSUPPORTED_SCHEMA_ROUTE_PREFIX = "babelqueue:unsupported-schema-version:"
65+
#: Separates the rejected version from the delivery's original ``bq-job`` in a gate route.
66+
_GATE_JOB_PARAM = "?bq-job="
67+
68+
_logger = logging.getLogger("babelqueue.pulsar")
69+
70+
71+
def _gate_route(version: str, properties: Dict[str, Any]) -> str:
72+
"""The reserved route of a version-gated delivery: the prefix, the rejected version and, when
73+
the delivery carried one, its original ``bq-job`` (both percent-escaped). The runtime hands
74+
it back through :meth:`PulsarTransport.publish_routed` on every retry / dead-letter write."""
75+
route = UNSUPPORTED_SCHEMA_ROUTE_PREFIX + quote(version, safe="")
76+
job = properties.get("bq-job")
77+
if job is not None:
78+
route += _GATE_JOB_PARAM + quote(str(job), safe="")
79+
return route
80+
81+
82+
def _parse_gate_route(route: str) -> Optional[Tuple[str, Optional[str]]]:
83+
"""Inverse of :func:`_gate_route`: ``(version, bq_job_or_None)``, or ``None`` for any other
84+
route."""
85+
if not route.startswith(UNSUPPORTED_SCHEMA_ROUTE_PREFIX):
86+
return None
87+
rest = route[len(UNSUPPORTED_SCHEMA_ROUTE_PREFIX):]
88+
raw_version, sep, raw_job = rest.partition(_GATE_JOB_PARAM)
89+
return unquote(raw_version), (unquote(raw_job) if sep else None)
4390

4491

4592
class PulsarTransport(Transport):
@@ -124,8 +171,19 @@ def _projection(body: str, route_urn: str = "") -> Dict[str, str]:
124171
the body: bq-job/bq-trace-id/bq-message-id + bq-schema-version/lang/attempts. §5.2.
125172
126173
A non-empty ``route_urn`` pins ``bq-job`` (a retry / release re-publish keeps the original
127-
delivery's routing URN); blank derives it from the body."""
174+
delivery's routing URN); blank derives it from the body.
175+
176+
A ``route_urn`` that is a version-gate route is never written: the original delivery's
177+
``bq-job`` (if it had one) and raw ``bq-schema-version`` are restored instead, so the
178+
republished message is gated the same way when it is consumed again."""
179+
gate = _parse_gate_route(route_urn)
128180
env = parse_envelope(body)
181+
if gate is not None:
182+
props = PulsarTransport._projection(body) if env else {}
183+
props["bq-schema-version"] = gate[0]
184+
if gate[1] is not None:
185+
props["bq-job"] = gate[1]
186+
return props
129187
if not env:
130188
return {}
131189
meta = env.get("meta") or {}
@@ -146,6 +204,23 @@ def _projection(body: str, route_urn: str = "") -> Dict[str, str]:
146204
props["bq-job"] = route_urn
147205
return props
148206

207+
@staticmethod
208+
def _unsupported_schema_version(properties: Dict[str, Any]) -> Optional[str]:
209+
"""The §5.7 step-3 gate on ``bq-schema-version``: the offending raw value, or ``None``.
210+
211+
A missing or blank property is not gated (the usual decode path runs); blank means the
212+
empty string or only ASCII whitespace (space, ``\\t``, ``\\n``, ``\\v``, ``\\f``, ``\\r``) — the
213+
same set in every SDK (GR-5), not ``str.strip()``, which also strips U+001C-U+001F,
214+
U+0085, U+00A0, U+2028, U+3000 ... A present property is accepted only when it is exactly
215+
the core's supported ``SCHEMA_VERSION`` as a plain decimal string, so ``"2"``, ``"x"``,
216+
``"01"`` and ``" 1"`` are all rejected."""
217+
raw = properties.get("bq-schema-version")
218+
if raw is None or (isinstance(raw, str) and _BLANK_ASCII_PATTERN.fullmatch(raw) is not None):
219+
return None
220+
if isinstance(raw, str) and raw == str(SCHEMA_VERSION):
221+
return None
222+
return str(raw)
223+
149224
@staticmethod
150225
def _parse_attempts(raw: Any) -> Optional[int]:
151226
"""The ``bq-attempts`` value as an int, or ``None`` when absent or invalid (§5.5)."""
@@ -267,10 +342,26 @@ def pop(self, queue: str, timeout: float = 1.0) -> Optional[ReceivedMessage]:
267342
raise
268343
if message is None: # pragma: no cover - defensive (real client raises instead)
269344
return None
345+
properties = self._properties(message)
346+
rejected = self._unsupported_schema_version(properties)
347+
if rejected is not None:
348+
# Version gate (§5.7 step 3): quarantine before decoding. The body is handed over
349+
# untouched; the reserved route keeps the runtime from dispatching it to a handler.
350+
_logger.warning(
351+
"Pulsar message on %r rejected by the bq-schema-version gate: unsupported "
352+
"version %r (supported: %s); quarantined without decoding the body",
353+
queue, rejected, SCHEMA_VERSION,
354+
)
355+
return ReceivedMessage(
356+
body=self._payload(message),
357+
queue=queue,
358+
handle=message,
359+
route_urn=_gate_route(rejected, properties),
360+
)
270361
body, route_urn = self._reconcile(
271362
self._payload(message),
272363
self._redelivery_count(message),
273-
self._properties(message),
364+
properties,
274365
self._on_routing_mismatch,
275366
)
276367
return ReceivedMessage(body=body, queue=queue, handle=message, route_urn=route_urn)

‎tests/test_app.py‎

Lines changed: 74 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -218,6 +218,80 @@ def handle(data, meta): # noqa: ANN001
218218
self.assertEqual(app.transport.size("orders"), 0)
219219
self.assertEqual(app.transport.size("orders.dlq"), 0)
220220

221+
def _redeliver_until_settled(self, app, queue, route_urn, first_body, limit=6):
222+
"""Feed ``first_body`` through ``dispatch`` and keep redelivering whatever the runtime
223+
re-publishes to ``queue`` (as a broker would); returns the number of deliveries."""
224+
tr = app.transport
225+
body = first_body
226+
deliveries = 0
227+
while body is not None and deliveries < limit:
228+
deliveries += 1
229+
app.dispatch(ReceivedMessage(body=body, queue=queue, route_urn=route_urn))
230+
body = tr._queues[queue].popleft() if tr._queues.get(queue) else None # noqa: SLF001
231+
return deliveries
232+
233+
def test_version_gated_message_is_bounded_under_release(self) -> None:
234+
from babelqueue.app import _UNSUPPORTED_SCHEMA_ROUTE_PREFIX
235+
from babelqueue.pulsar_transport import UNSUPPORTED_SCHEMA_ROUTE_PREFIX
236+
237+
self.assertEqual(_UNSUPPORTED_SCHEMA_ROUTE_PREFIX, UNSUPPORTED_SCHEMA_ROUTE_PREFIX)
238+
route = _UNSUPPORTED_SCHEMA_ROUTE_PREFIX + "2"
239+
decodable = EnvelopeCodec.encode(EnvelopeCodec.make("urn:babel:orders:created", {}, queue="orders"))
240+
for name, body in (("decodable", decodable), ("undecodable", "{not json")):
241+
with self.subTest(body=name):
242+
app = BabelQueue(
243+
"memory://", queue="orders", max_attempts=3, dead_letter=True,
244+
on_unknown_urn=UnknownUrnStrategy.RELEASE,
245+
)
246+
seen = []
247+
app.handler("urn:babel:orders:created")(lambda d, m: seen.append(d))
248+
249+
deliveries = self._redeliver_until_settled(app, "orders", route, body)
250+
251+
self.assertEqual(deliveries, 3) # max_attempts, not an endless release loop
252+
self.assertEqual(seen, [])
253+
self.assertEqual(app.transport.size("orders"), 0)
254+
self.assertEqual(app.transport.size("orders.dlq"), 1)
255+
dead = EnvelopeCodec.decode(app.transport._queues["orders.dlq"][0]) # noqa: SLF001
256+
self.assertEqual(dead["dead_letter"]["reason"], "unknown_urn")
257+
self.assertIn(route, dead["dead_letter"]["error"])
258+
259+
def test_version_gated_message_without_dead_letter_is_dropped_under_release(self) -> None:
260+
route = "babelqueue:unsupported-schema-version:x"
261+
body = EnvelopeCodec.encode(EnvelopeCodec.make("urn:babel:orders:created", {}, queue="orders"))
262+
app = BabelQueue(
263+
"memory://", queue="orders", max_attempts=2, on_unknown_urn=UnknownUrnStrategy.RELEASE
264+
)
265+
self.assertEqual(self._redeliver_until_settled(app, "orders", route, body), 2)
266+
self.assertEqual(app.transport.size("orders"), 0)
267+
268+
def test_unknown_urn_release_still_releases_verbatim(self) -> None:
269+
body = EnvelopeCodec.encode(EnvelopeCodec.make("urn:babel:orders:nope", {}, queue="orders"))
270+
app = BabelQueue(
271+
"memory://", queue="orders", max_attempts=2, dead_letter=True,
272+
on_unknown_urn=UnknownUrnStrategy.RELEASE,
273+
)
274+
for route_urn in ("", "urn:babel:orders:nope"):
275+
with self.subTest(route_urn=route_urn):
276+
app.dispatch(ReceivedMessage(body=body, queue="orders", route_urn=route_urn))
277+
self.assertEqual(app.transport.size("orders.dlq"), 0)
278+
self.assertEqual(app.transport._queues["orders"].popleft(), body) # noqa: SLF001
279+
280+
def test_ungated_poison_is_bounded_under_release(self) -> None:
281+
# Not version-gated: an undecodable body and a body naming no URN can never gain a handler
282+
# either, so RELEASE takes the bounded FAIL path for them too (Go routePoison parity).
283+
no_urn = EnvelopeCodec.encode(EnvelopeCodec.make("urn:babel:orders:created", {}, queue="orders"))
284+
no_urn = no_urn.replace('"urn:babel:orders:created"', '""')
285+
for name, body in (("undecodable", "{not json"), ("no-urn", no_urn)):
286+
with self.subTest(body=name):
287+
app = BabelQueue(
288+
"memory://", queue="orders", max_attempts=3, dead_letter=True,
289+
on_unknown_urn=UnknownUrnStrategy.RELEASE,
290+
)
291+
self.assertEqual(self._redeliver_until_settled(app, "orders", "", body), 3)
292+
self.assertEqual(app.transport.size("orders"), 0)
293+
self.assertEqual(app.transport.size("orders.dlq"), 1)
294+
221295
def test_unsupported_broker_scheme_raises(self) -> None:
222296
from babelqueue import BabelQueueError
223297

0 commit comments

Comments
 (0)