From b3dcfac31592d1368e426bb409c1243c7cea83bc Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Muhammet=20=C5=9Eafak?= Date: Sun, 25 Oct 2026 17:10:04 +0300 Subject: [PATCH 1/3] Write the ADR-0035 poison record to .dlq on SQS (ADR-0037) Poison and version-gated bodies are dead-lettered as ADR-0035 records when dead-lettering is on and left to the native RedrivePolicy when it is off. Both routes read only the explicitly configured on_unknown_urn strategy. Pin conformance v1.8.0 and run version_gate_sqs. --- .github/workflows/ci.yml | 2 +- CHANGELOG.md | 34 + README.md | 14 +- .../java/com/babelqueue/sqs/SqsConsumer.java | 231 ++- .../com/babelqueue/sqs/FakeSqsClient.java | 15 + .../babelqueue/sqs/PoisonRecordChecks.java | 131 ++ .../babelqueue/sqs/SqsConformanceTest.java | 114 +- .../com/babelqueue/sqs/SqsConsumerTest.java | 277 +++ .../resources/conformance/CONFORMANCE_VERSION | 2 +- src/test/resources/conformance/manifest.json | 1523 ++++++++++++++++- .../conformance/schema/results.example.json | 2 +- 11 files changed, 2317 insertions(+), 28 deletions(-) create mode 100644 src/test/java/com/babelqueue/sqs/PoisonRecordChecks.java diff --git a/.github/workflows/ci.yml b/.github/workflows/ci.yml index 31a870b..ba13797 100644 --- a/.github/workflows/ci.yml +++ b/.github/workflows/ci.yml @@ -34,7 +34,7 @@ jobs: name: Conformance suite in sync runs-on: ubuntu-latest env: - BQ_CONFORMANCE_VERSION: 1.6.0 + BQ_CONFORMANCE_VERSION: 1.8.0 # Integrity pin for fetch.sh (sha256 of conformance/fetch.sh at the pinned tag). BQ_FETCH_SH_SHA256: 8bb6ba567b288fb0ac71a70fd17041b2503df119e01998d76fdad372570bf849 steps: diff --git a/CHANGELOG.md b/CHANGELOG.md index 3e5689b..0361d5b 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -9,6 +9,40 @@ The envelope wire format is versioned separately by `meta.schema_version` ## [Unreleased] +MINOR (ADR-0037): the consumer gains an opt-in BabelQueue DLQ and settles poison bodies and version-gated +deliveries on their first delivery. The envelope is unchanged (`schema_version` stays `1`). + +### Added + +- `SqsConsumer.Builder.deadLetterQueueUrl(url)` (opt-in, like the .NET `DeadLetterQueueUrl` and the Node + `deadLetterQueueUrl`): the `.dlq` queue (a FIFO one is `.dlq.fifo`; the consumer sends it + the queue name as message group and the SQS message id as deduplication id). "Dead-lettering on" means this + URL is set. The queue must exist. +- With a DLQ queue set, the `dead_letter` unknown-URN strategy now writes the envelope, annotated with a + `dead_letter` block (reason `unknown_urn`), to it and deletes the message only after the send succeeded. + Without one it still degrades to `delete`. + +### Changed (behaviour) + +- **Poison body** (not JSON, not an envelope, no valid identity, or an unsupported `meta.schema_version`): + never handed to a handler and never retried. With a DLQ queue it is sent as the ADR-0035 record + (`PoisonRecord`, `reason: "poison"`, `original_queue` = this queue) and deleted **only after** the send + succeeded; a failed send leaves it unsettled and is reported. Under an explicit `delete` strategy it is + deleted. With no DLQ queue nothing changes: it is reported and left to the queue's `RedrivePolicy`. +- **Version-gated delivery** (`bq-schema-version` present, not blank, not `1`): settled on the first delivery, + body never decoded, no handler or hook, no retry. With a DLQ queue the **raw body, byte for byte** (no + `dead_letter` block) is sent to it and the message deleted only after the send succeeded (a failed send + leaves it unsettled, so the write is retried on redelivery); under an explicit `delete` strategy it is + deleted (**dropped**); with no DLQ queue nothing changes (reported, not deleted, native redrive). +- The poison and version-gate rules read only the **explicitly configured** `unknownUrnStrategy` + (`fail` when none is set). An `onUnknownUrn` hook, and the `delete` it implies for an unknown URN, no + longer reach them: a hook-only consumer never drops a poison or version-gated delivery silently (owner + decision, 2026-10-25: the poison route reads only the explicitly configured strategy, like the version + gate). The unknown-URN path is unchanged. +- Built against `babelqueue-core` 1.13.0 (`PoisonRecord`), unchanged. +- Vendored the cross-SDK conformance suite v1.8.0 (`version_gate_sqs`, `poison_dlq` runners added) and moved + the CI pin to it. + ## [1.3.0] - 2026-10-24 ### Added diff --git a/README.md b/README.md index 90602b8..83cdfe7 100644 --- a/README.md +++ b/README.md @@ -70,8 +70,17 @@ delete — observe via `onError` / `onUnknownUrn`. The envelope is unchanged **Schema-version gate (§3.7).** Before decoding the body the consumer reads the `bq-schema-version` attribute. Missing or blank (empty / ASCII whitespace only) → normal decode. Any other value that is not exactly `1` (no trimming — `"2"`, `"x"`, `" 1"`) → the body is **not** -decoded, the handler is not called, `onError` is notified with a `null` envelope, and the message is -left undeleted so SQS redrives it to the DLQ. +decoded, the handler is not called, no hook runs, `onError` is notified with a `null` envelope, and the +message is settled on its **first** delivery (ADR-0037): with a `deadLetterQueueUrl` the body is sent to +`.dlq` **exactly as received** (no `dead_letter` block) and deleted after the send succeeded (a failed +send leaves it for redelivery); under an explicit `delete` strategy it is deleted; without a DLQ queue it is +left undeleted so SQS redrives it (`RedrivePolicy`). Only an explicitly set `unknownUrnStrategy` is read — an +`onUnknownUrn` hook never drops a gated message. + +**Poison bodies (ADR-0037).** A body that is not a valid envelope (not JSON, no identity, unsupported +`meta.schema_version`) never reaches a handler and is never retried. With a `deadLetterQueueUrl` it is written +to the DLQ as the ADR-0035 record (`reason: "poison"`; the original body kept, spliced or under `raw`) and +deleted after the send succeeded; without one it is reported and left to the queue's `RedrivePolicy`. ```java SqsConsumer.builder(sqs, url) @@ -81,6 +90,7 @@ SqsConsumer.builder(sqs, url) //.retryBackoff(5).retryBackoffMultiplier(2).retryBackoffMax(300).retryJitter("full") .unknownUrnStrategy(UnknownUrnStrategy.RELEASE) // fail | delete | release | dead_letter .unknownUrnReleaseDelaySeconds(60) // default 0 + .deadLetterQueueUrl(dlqUrl) // opt-in BabelQueue DLQ (.dlq / .dlq.fifo) .build(); ``` diff --git a/src/main/java/com/babelqueue/sqs/SqsConsumer.java b/src/main/java/com/babelqueue/sqs/SqsConsumer.java index 95edf18..26e7fa5 100644 --- a/src/main/java/com/babelqueue/sqs/SqsConsumer.java +++ b/src/main/java/com/babelqueue/sqs/SqsConsumer.java @@ -1,11 +1,19 @@ package com.babelqueue.sqs; import com.babelqueue.BabelQueueException; +import com.babelqueue.DeadLetters; import com.babelqueue.Envelope; import com.babelqueue.EnvelopeCodec; +import com.babelqueue.JsonValues; +import com.babelqueue.PoisonRecord; import com.babelqueue.UnknownUrnException; import com.babelqueue.UnknownUrnStrategy; +import java.nio.charset.StandardCharsets; +import java.security.MessageDigest; +import java.security.NoSuchAlgorithmException; import java.util.HashMap; +import java.util.HexFormat; +import java.util.LinkedHashMap; import java.util.List; import java.util.Map; import java.util.Objects; @@ -18,6 +26,7 @@ import software.amazon.awssdk.services.sqs.model.Message; import software.amazon.awssdk.services.sqs.model.MessageAttributeValue; import software.amazon.awssdk.services.sqs.model.ReceiveMessageRequest; +import software.amazon.awssdk.services.sqs.model.SendMessageRequest; /** * Polls an SQS queue, decodes and validates each message, routes it to the handler @@ -31,21 +40,36 @@ *

A URN with no handler follows the {@link UnknownUrnStrategy unknown-URN strategy}: * {@code fail} reports {@link UnknownUrnException} and leaves the message (visibility expiry * redelivers it); {@code delete} deletes it; {@code release} releases it with - * {@code unknownUrnReleaseDelaySeconds} (default {@code 0}); {@code dead_letter} degrades to {@code delete} - * because this transport has no DLQ publisher (the contract's "DLQ disabled" rule). + * {@code unknownUrnReleaseDelaySeconds} (default {@code 0}); {@code dead_letter} writes the envelope, annotated with + * a {@code dead_letter} block (reason {@code unknown_urn}), to the {@link Builder#deadLetterQueueUrl DLQ queue} and + * then deletes it, or degrades to {@code delete} when no DLQ queue is configured (the contract's "DLQ disabled" rule). * The poll loop never stops on a bad message, a failed release or a failed delete — observe via - * {@code onError}/{@code onUnknownUrn}. Configure a queue {@code RedrivePolicy} so a poison - * message, redelivered immediately under the default {@code 0} delay, ends in a native DLQ. + * {@code onError}/{@code onUnknownUrn}. Configure a queue {@code RedrivePolicy} so a message left to the broker + * (see below), redelivered immediately under the default {@code 0} delay, ends in a native DLQ. * - *

Per §3.7 the {@code bq-schema-version} message attribute is checked before the body is - * decoded: when it is present and is not exactly the schema version the core supports - * ({@link EnvelopeCodec#SCHEMA_VERSION}, compared exactly as sent, with no trimming — so {@code "2"}, - * {@code "x"}, {@code "01"} and {@code " 1"} are all unknown), the body is never decoded or routed; - * {@code onError} is notified (with a {@code null} envelope) and the message is neither deleted nor - * released — the same settle path as a non-conformant envelope, so the queue's redrive policy moves it - * to the native DLQ. A missing or blank (empty or ASCII-whitespace-only: space, tab, LF, VT, FF, CR) - * attribute changes nothing, and a {@code "1"} attribute still goes through the post-decode - * {@link EnvelopeCodec#accepts} check. + *

Poison and version gate (ADR-0037). "Dead-lettering on" means a {@link Builder#deadLetterQueueUrl DLQ + * queue URL} is configured (opt-in; a FIFO DLQ is {@code .dlq.fifo}). Only the explicitly configured + * {@link Builder#unknownUrnStrategy strategy} is read by these two rules (none configured means {@code fail}); an + * {@code onUnknownUrn} hook, and the {@code delete} it implies for an unknown URN, never affect them. + *

    + *
  • Poison body — not JSON, not an envelope, no valid identity or an unsupported + * {@code meta.schema_version}: never handed to a handler and never retried. {@code delete} deletes it; with a + * DLQ queue the ADR-0035 record ({@link PoisonRecord}, {@code reason: "poison"}) is sent to it and the message + * deleted only after the send succeeded (a failed send leaves it unsettled and is reported); with no + * DLQ queue it is reported and left to the broker's native mechanism (no {@code DeleteMessage}: visibility + * expiry redelivers it and the queue's {@code RedrivePolicy} moves it after {@code maxReceiveCount}).
    • + *
  • Version gate (§3.7) — the {@code bq-schema-version} message attribute is checked before the + * body is decoded: when it is present and is not exactly the schema version the core supports + * ({@link EnvelopeCodec#SCHEMA_VERSION}, compared exactly as sent, with no trimming — so {@code "2"}, + * {@code "x"}, {@code "01"} and {@code " 1"} are all unknown), the body is never decoded or routed, no handler + * or hook runs, and nothing is retried, on the first delivery, whatever {@code ApproximateReceiveCount} says. + * {@code onError} is notified (with a {@code null} envelope). {@code delete} deletes it; with a DLQ queue the + * body is sent to it exactly as received (no {@code dead_letter} block) and the message deleted only + * after the send succeeded (a failed send leaves it unsettled); with no DLQ queue it is left to the broker's + * native mechanism, as above. A missing or blank (empty or ASCII-whitespace-only: space, tab, LF, VT, FF, CR) + * attribute changes nothing, and a {@code "1"} attribute still goes through the post-decode + * {@link EnvelopeCodec#accepts} check (a body with an unsupported {@code meta.schema_version} is then poison).
    • + *
*/ public final class SqsConsumer { @@ -81,6 +105,9 @@ public interface UnknownUrnHandler { private final ErrorHandler onError; private final UnknownUrnHandler onUnknownUrn; private final String unknownUrnStrategy; + /** The strategy the caller set explicitly, or {@code null}: the only one the poison and version-gate rules read. */ + private final String configuredStrategy; + private final String deadLetterQueueUrl; private final double retryBackoffSeconds; private final double retryBackoffMultiplier; private final double retryBackoffMaxSeconds; @@ -103,6 +130,8 @@ private SqsConsumer(Builder builder) { this.unknownUrnStrategy = builder.unknownUrnStrategy != null ? builder.unknownUrnStrategy : (builder.onUnknownUrn != null ? UnknownUrnStrategy.DELETE : UnknownUrnStrategy.FAIL); + this.configuredStrategy = builder.unknownUrnStrategy; + this.deadLetterQueueUrl = builder.deadLetterQueueUrl; this.retryBackoffSeconds = builder.retryBackoffSeconds; this.retryBackoffMultiplier = builder.retryBackoffMultiplier; this.retryBackoffMaxSeconds = builder.retryBackoffMaxSeconds; @@ -149,9 +178,7 @@ private void handle(Message message) { String declaredVersion = rawSchemaVersion(message); if (declaredVersion != null && !isBlankSchemaVersion(declaredVersion) && !declaredVersion.equals(Integer.toString(EnvelopeCodec.SCHEMA_VERSION))) { - report(new BabelQueueException("Rejected a BabelQueue message with unsupported " - + SCHEMA_VERSION_ATTRIBUTE + " attribute '" + declaredVersion + "' (supported: " - + EnvelopeCodec.SCHEMA_VERSION + "); body not decoded."), null, message); + settleVersionGated(message, declaredVersion); return; } @@ -161,8 +188,7 @@ private void handle(Message message) { message.attributesAsStrings().get("ApproximateReceiveCount")); if (!EnvelopeCodec.accepts(envelope)) { - report(new BabelQueueException("Rejected a non-conformant BabelQueue envelope from SQS."), - envelope, message); + settlePoison(message, body, envelope); return; } @@ -187,6 +213,127 @@ private void handle(Message message) { delete(message, envelope); } + /** + * Settles a version-gated delivery on its first delivery (ADR-0037 decision 1, broker-bindings §3.7): never + * decoded, no handler or hook, no retry. Reads only the explicitly configured strategy. + */ + private void settleVersionGated(Message message, String declaredVersion) { + report(new BabelQueueException("Rejected a BabelQueue message with unsupported " + + SCHEMA_VERSION_ATTRIBUTE + " attribute '" + declaredVersion + "' (supported: " + + EnvelopeCodec.SCHEMA_VERSION + "); body not decoded."), null, message); + if (UnknownUrnStrategy.DELETE.equals(configuredStrategy)) { + delete(message, null); + return; + } + if (deadLetterQueueUrl == null) { + // Rule 3: nothing to write to. Neither deleted nor released: the broker's native mechanism + // (visibility expiry, then the queue's RedrivePolicy) takes it. The SDK does not retry it. + return; + } + // Rule 4: the raw body, byte for byte, then delete. Rule 5: a failed send leaves it unsettled. + String body = message.body() == null ? "" : message.body(); + if (sendToDeadLetterQueue(message, body, message.hasMessageAttributes() + ? message.messageAttributes() : Map.of(), null)) { + delete(message, null); + } + } + + /** + * Settles a poison body (ADR-0037 decision 2, error-handling.md §7): never handed to a handler, never retried. + * Reads only the explicitly configured strategy. + */ + private void settlePoison(Message message, String body, Envelope partial) { + BabelQueueException cause = new BabelQueueException("Rejected a non-conformant BabelQueue envelope from SQS."); + report(cause, partial, message); + if (UnknownUrnStrategy.DELETE.equals(configuredStrategy)) { + delete(message, partial); + return; + } + if (deadLetterQueueUrl == null) { + // Native: neither deleted nor released (visibility expiry, then the queue's RedrivePolicy). + return; + } + String record; + try { + record = poisonRecord(body.getBytes(StandardCharsets.UTF_8), SqsQueues.nameFromUrl(queueUrl), + receiveAttempts(message), cause); + } catch (RuntimeException error) { + report(error, partial, message); + return; + } + if (sendToDeadLetterQueue(message, record, message.hasMessageAttributes() + ? message.messageAttributes() : Map.of(), partial)) { + delete(message, partial); + } + } + + /** + * The ADR-0035 DLQ record of a poison {@code body}: the one builder this consumer writes with (conformance + * {@code poison_dlq} drives it directly for bodies SQS itself cannot carry, such as invalid UTF-8). + */ + static String poisonRecord(byte[] body, String originalQueue, int attempts, Throwable cause) { + Map block = new LinkedHashMap<>(); + block.put("reason", "poison"); + block.put("error", cause.getMessage()); + block.put("exception", cause.getClass().getName()); + block.put("failed_at", System.currentTimeMillis()); + block.put("original_queue", originalQueue); + block.put("attempts", attempts); + block.put("lang", EnvelopeCodec.SOURCE_LANG); + return PoisonRecord.build(body, JsonValues.encode(block)); + } + + /** {@code ApproximateReceiveCount - 1} (informational {@code dead_letter.attempts}), never negative. */ + private static int receiveAttempts(Message message) { + String receiveCount = message.attributesAsStrings().get("ApproximateReceiveCount"); + if (receiveCount == null) { + return 0; + } + try { + return Math.max(Integer.parseInt(receiveCount.strip()) - 1, 0); + } catch (NumberFormatException ex) { + return 0; + } + } + + /** + * Sends {@code body} to the DLQ queue; returns {@code false} (after reporting the failure) when the send failed, + * so the caller leaves the message unsettled and SQS redelivers it once the visibility timeout expires. A FIFO + * DLQ ({@code .fifo}) gets the consumer's queue name as the message group and the SQS message id (else a + * digest of the body) as the deduplication id, so a send repeated after a lost delete is not duplicated. + */ + private boolean sendToDeadLetterQueue( + Message message, String body, Map attributes, Envelope envelope) { + try { + SendMessageRequest.Builder request = SendMessageRequest.builder() + .queueUrl(deadLetterQueueUrl) + .messageBody(body); + if (!attributes.isEmpty()) { + request.messageAttributes(attributes); + } + if (deadLetterQueueUrl.endsWith(".fifo")) { + request.messageGroupId(SqsQueues.nameFromUrl(queueUrl)); + request.messageDeduplicationId(message.messageId() != null && !message.messageId().isEmpty() + ? message.messageId() + : sha256Hex(body)); + } + client.sendMessage(request.build()); + return true; + } catch (RuntimeException sendError) { + report(sendError, envelope, message); + return false; + } + } + + private static String sha256Hex(String value) { + try { + return HexFormat.of().formatHex( + MessageDigest.getInstance("SHA-256").digest(value.getBytes(StandardCharsets.UTF_8))); + } catch (NoSuchAlgorithmException impossible) { + throw new IllegalStateException(impossible); + } + } + /** The {@code bq-schema-version} attribute's string value exactly as sent (no trimming), or {@code null} when absent. */ private static String rawSchemaVersion(Message message) { if (!message.hasMessageAttributes()) { @@ -216,7 +363,8 @@ private void handleUnknownUrn(String urn, Envelope envelope, Message message) { onUnknownUrn.onUnknownUrn(envelope, message); } switch (unknownUrnStrategy) { - case UnknownUrnStrategy.DELETE, UnknownUrnStrategy.DEAD_LETTER -> delete(message, envelope); + case UnknownUrnStrategy.DELETE -> delete(message, envelope); + case UnknownUrnStrategy.DEAD_LETTER -> deadLetterUnknownUrn(message, envelope); case UnknownUrnStrategy.RELEASE -> release(message, unknownUrnReleaseDelaySeconds, envelope); default -> { // fail: leave the message; visibility expiry redelivers it, native redrive quarantines it. @@ -227,6 +375,29 @@ private void handleUnknownUrn(String urn, Envelope envelope, Message message) { } } + /** + * The {@code dead_letter} unknown-URN strategy: with a DLQ queue the envelope, annotated with a {@code dead_letter} + * block (reason {@code unknown_urn}), is sent to it and the message deleted only after the send succeeded (a + * failed send leaves it unsettled and is reported); with none it degrades to {@code delete}. + */ + private void deadLetterUnknownUrn(Message message, Envelope envelope) { + if (deadLetterQueueUrl == null) { + delete(message, envelope); + return; + } + Envelope dead; + String body; + try { + dead = DeadLetters.annotate(envelope, "unknown_urn", SqsQueues.nameFromUrl(queueUrl)); + body = EnvelopeCodec.encode(dead); + if (sendToDeadLetterQueue(message, body, SqsAttributes.project(dead), envelope)) { + delete(message, envelope); + } + } catch (RuntimeException error) { + report(error, envelope, message); + } + } + /** * Set {@code attempts} to max(current, ApproximateReceiveCount - 1): a first delivery * reads 0, a natively-redelivered message reflects its true count, and a @@ -342,6 +513,7 @@ public static final class Builder { private ErrorHandler onError; private UnknownUrnHandler onUnknownUrn; private String unknownUrnStrategy; + private String deadLetterQueueUrl; private double retryBackoffSeconds; private double retryBackoffMultiplier = 1.0; private double retryBackoffMaxSeconds; @@ -462,8 +634,9 @@ Builder randomSource(DoubleSupplier source) { /** * Unknown-URN strategy: one of {@link UnknownUrnStrategy#FAIL}, {@link UnknownUrnStrategy#DELETE}, * {@link UnknownUrnStrategy#RELEASE} or {@link UnknownUrnStrategy#DEAD_LETTER} (degrades to - * {@code delete}: no DLQ publisher in this transport). Default: {@code fail}, or {@code delete} - * when only {@link #onUnknownUrn} is set (pre-1.2.0 behaviour). + * {@code delete} when no {@link #deadLetterQueueUrl DLQ queue} is configured). Default: {@code fail}, or + * {@code delete} for an unknown URN when only {@link #onUnknownUrn} is set (pre-1.2.0 behaviour; that implied + * {@code delete} never reaches the poison and version-gate rules). */ public Builder unknownUrnStrategy(String strategy) { Objects.requireNonNull(strategy, "strategy"); @@ -475,6 +648,22 @@ public Builder unknownUrnStrategy(String strategy) { return this; } + /** + * Turns BabelQueue dead-lettering on (opt-in, ADR-0037 decision 2): the URL of the {@code .dlq} queue + * (a FIFO one is {@code .dlq.fifo}). A poison body is sent to it as the ADR-0035 record, a + * version-gated delivery as its raw body, and an unknown URN under the {@code dead_letter} strategy as an + * annotated envelope; the source message is deleted only after the send succeeded. Unset (the default): a + * poison or version-gated delivery is left to the queue's native {@code RedrivePolicy}. The queue must + * exist; this consumer never creates it. + */ + public Builder deadLetterQueueUrl(String url) { + if (url != null && url.isBlank()) { + throw new IllegalArgumentException("deadLetterQueueUrl must not be blank"); + } + this.deadLetterQueueUrl = url; + return this; + } + /** * Backoff (seconds) for the {@code release} unknown-URN strategy. Default {@code 0} * (redeliver now, matching the Python and Node transports); out-of-range values are diff --git a/src/test/java/com/babelqueue/sqs/FakeSqsClient.java b/src/test/java/com/babelqueue/sqs/FakeSqsClient.java index e33cdb8..ec4f795 100644 --- a/src/test/java/com/babelqueue/sqs/FakeSqsClient.java +++ b/src/test/java/com/babelqueue/sqs/FakeSqsClient.java @@ -12,6 +12,7 @@ import software.amazon.awssdk.services.sqs.model.DeleteMessageRequest; import software.amazon.awssdk.services.sqs.model.DeleteMessageResponse; import software.amazon.awssdk.services.sqs.model.Message; +import software.amazon.awssdk.services.sqs.model.MessageAttributeValue; import software.amazon.awssdk.services.sqs.model.ReceiveMessageRequest; import software.amazon.awssdk.services.sqs.model.ReceiveMessageResponse; import software.amazon.awssdk.services.sqs.model.SendMessageRequest; @@ -28,6 +29,8 @@ final class FakeSqsClient implements SqsClient { RuntimeException visibilityError; /** When set, only {@code deleteMessage} throws it (a failed acknowledgement). */ RuntimeException deleteError; + /** When set, only {@code sendMessage} throws it (a failed publish or dead-letter write). */ + RuntimeException sendError; private final Map> queues = new HashMap<>(); private final RuntimeException error; @@ -56,6 +59,9 @@ public SendMessageResponse sendMessage(SendMessageRequest request) { if (error != null) { throw error; } + if (sendError != null) { + throw sendError; + } sent.add(request); counter++; String handle = "rh-" + counter; @@ -122,6 +128,15 @@ void seed(String queueUrl, String body, String receiveCount) { push(queueUrl, message.build()); } + /** Seed a message with message attributes and a chosen ApproximateReceiveCount. */ + void seed(String queueUrl, String body, int receiveCount, Map attributes) { + counter++; + push(queueUrl, Message.builder().body(body).receiptHandle("seed-" + counter) + .messageAttributes(attributes) + .attributesWithStrings(Map.of("ApproximateReceiveCount", String.valueOf(receiveCount))) + .build()); + } + private void push(String queueUrl, Message message) { queues.computeIfAbsent(queueUrl, key -> new ArrayDeque<>()).addLast(message); } diff --git a/src/test/java/com/babelqueue/sqs/PoisonRecordChecks.java b/src/test/java/com/babelqueue/sqs/PoisonRecordChecks.java new file mode 100644 index 0000000..3ca4387 --- /dev/null +++ b/src/test/java/com/babelqueue/sqs/PoisonRecordChecks.java @@ -0,0 +1,131 @@ +package com.babelqueue.sqs; + +import static org.junit.jupiter.api.Assertions.assertEquals; +import static org.junit.jupiter.api.Assertions.assertFalse; +import static org.junit.jupiter.api.Assertions.assertTrue; + +import java.nio.charset.StandardCharsets; +import java.util.Base64; +import java.util.LinkedHashSet; +import java.util.Set; +import org.json.JSONArray; +import org.json.JSONObject; + +/** + * The assertions of the core-scope {@code poison_dlq} conformance section (ADR-0035), applied to the DLQ record a + * consumer actually wrote: the {@code dead_letter} block says {@code poison} / {@code orders}, and the rest of the + * record is the wrapper of the case or a byte-level splice of the body. + */ +final class PoisonRecordChecks { + + private PoisonRecordChecks() { + } + + /** The exact bytes of the case's body ({@code body} text or {@code body_base64}). */ + static byte[] bodyOf(JSONObject testCase) { + return testCase.has("body_base64") + ? Base64.getDecoder().decode(testCase.getString("body_base64")) + : testCase.getString("body").getBytes(StandardCharsets.UTF_8); + } + + static void check(JSONObject testCase, String record) { + String name = testCase.getString("name"); + JSONObject expect = testCase.getJSONObject("expect"); + String shape = expect.getString("shape"); + JSONObject want = expect.getJSONObject("record"); + assertFalse(record.indexOf('�') >= 0, name + ": the record holds U+FFFD"); + Set keys = topLevelKeys(record); + boolean wrapper = want.has("raw") && keys.contains("raw"); + + switch (shape) { + case "wrapper" -> assertTrue(wrapper, name + ": expected the wrapper, got " + record); + case "splice" -> { + assertFalse(keys.contains("raw_encoding"), name + ": expected a splice, got " + record); + assertTrue(keys.contains("dead_letter"), name + ": the splice lost dead_letter: " + record); + } + case "splice_or_wrapper" -> { } + default -> throw new AssertionError(name + ": unknown shape " + shape); + } + + if (wrapper) { + JSONObject parsed = new JSONObject(record); + JSONObject deadLetter = (JSONObject) parsed.remove("dead_letter"); + assertEquals("poison", deadLetter.getString("reason"), name); + assertEquals("orders", deadLetter.getString("original_queue"), name); + assertTrue(want.similar(parsed), name + ": " + record); + return; + } + + assertFalse(keys.contains("raw") && keys.contains("raw_encoding"), name + ": a splice must not wrap"); + assertFalse(keys.contains("raw_encoding"), name + ": a splice must not carry raw_encoding"); + if (expect.has("splice_keys")) { + Set expectedKeys = new LinkedHashSet<>(); + JSONArray list = expect.getJSONArray("splice_keys"); + for (int i = 0; i < list.length(); i++) { + expectedKeys.add(list.getString(i)); + } + Set actualKeys = new LinkedHashSet<>(keys); + actualKeys.remove("dead_letter"); + assertEquals(expectedKeys, actualKeys, name + ": top-level member names of " + record); + } + if (expect.has("splice_literals")) { + JSONArray literals = expect.getJSONArray("splice_literals"); + for (int i = 0; i < literals.length(); i++) { + assertTrue(record.contains(literals.getString(i)), name + ": missing literal " + literals.getString(i)); + } + assertEquals("poison", deadLetterReason(record), name); + } else { + JSONObject parsed = new JSONObject(record); + JSONObject deadLetter = (JSONObject) parsed.remove("dead_letter"); + assertEquals("poison", deadLetter.getString("reason"), name); + assertEquals("orders", deadLetter.getString("original_queue"), name); + assertTrue(want.similar(parsed), name + ": " + record); + } + } + + /** The {@code reason} of the final {@code dead_letter} member, read from the record tail the splice appends. */ + private static String deadLetterReason(String record) { + int at = record.lastIndexOf("\"dead_letter\":"); + JSONObject block = new JSONObject(new org.json.JSONTokener(record.substring(at + "\"dead_letter\":".length()))); + assertEquals("orders", block.getString("original_queue")); + return block.getString("reason"); + } + + /** + * The distinct top-level member names of a JSON object text, read with a tolerant scanner (string-aware, depth + * tracking) because a conforming splice may hold values the test's own parser cannot represent. + */ + static Set topLevelKeys(String json) { + Set keys = new LinkedHashSet<>(); + int depth = 0; + int i = 0; + int n = json.length(); + while (i < n) { + char ch = json.charAt(i); + if (ch == '"') { + int start = i++; + while (i < n && json.charAt(i) != '"') { + i += json.charAt(i) == '\\' ? 2 : 1; + } + int end = i++; + if (depth == 1) { + int j = i; + while (j < n && " \t\r\n".indexOf(json.charAt(j)) >= 0) { + j++; + } + if (j < n && json.charAt(j) == ':') { + keys.add(new JSONArray("[" + json.substring(start, end + 1) + "]").getString(0)); + } + } + continue; + } + if (ch == '{' || ch == '[') { + depth++; + } else if (ch == '}' || ch == ']') { + depth--; + } + i++; + } + return keys; + } +} diff --git a/src/test/java/com/babelqueue/sqs/SqsConformanceTest.java b/src/test/java/com/babelqueue/sqs/SqsConformanceTest.java index c2bea84..1fc1e48 100644 --- a/src/test/java/com/babelqueue/sqs/SqsConformanceTest.java +++ b/src/test/java/com/babelqueue/sqs/SqsConformanceTest.java @@ -19,7 +19,8 @@ /** * Amazon SQS binding conformance against the vendored canonical suite's {@code sqs} * block: the §3 attribute projection and the {@code attempts = ApproximateReceiveCount - 1} - * reconciliation. No AWS, no network. + * reconciliation, plus the ADR-0037 {@code version_gate_sqs} settle rule and the ADR-0035 + * {@code poison_dlq} record shape driven through this consumer. No AWS, no network. */ class SqsConformanceTest { @@ -109,4 +110,115 @@ void schemaVersionGateMatchesGolden() throws Exception { assertEquals(decode ? 1 : 0, sqs.deleted.size(), label); } } + // ---- poison_dlq (ADR-0035 shape, ADR-0037 decision 2) ----------------------------------------- + + private static final String DLQ_URL = "https://sqs.eu-central-1.amazonaws.com/123456789012/orders.dlq"; + + @Test + void poisonDlqRecordShapeMatchesGolden() throws Exception { + JSONArray cases = new JSONObject(resource("manifest.json")).getJSONObject("poison_dlq").getJSONArray("cases"); + assertTrue(cases.length() > 0); + for (int i = 0; i < cases.length(); i++) { + JSONObject testCase = cases.getJSONObject(i); + String record; + if (testCase.has("body_base64")) { + // SQS delivers text, so a body that is not valid UTF-8 never reaches the consumer: drive the builder it + // writes with directly. + record = SqsConsumer.poisonRecord(PoisonRecordChecks.bodyOf(testCase), "orders", 0, + new com.babelqueue.BabelQueueException("poison")); + } else { + FakeSqsClient sqs = new FakeSqsClient(); + sqs.seed(URL, testCase.getString("body"), 1); + int[] handled = {0}; + SqsConsumer.builder(sqs, URL) + .handler("urn:babel:orders:created", (env, message) -> handled[0]++) + .deadLetterQueueUrl(DLQ_URL) + .build() + .poll(); + assertEquals(0, handled[0], testCase.getString("name")); + assertEquals(1, sqs.sent.size(), testCase.getString("name")); + assertEquals(1, sqs.deleted.size(), testCase.getString("name")); + record = sqs.sent.get(0).messageBody(); + } + PoisonRecordChecks.check(testCase, record); + } + } + + // ---- version_gate_sqs (ADR-0037 decision 1) --------------------------------------------------- + + @Test + void versionGateSettleRuleMatchesGolden() throws Exception { + JSONObject section = new JSONObject(resource("manifest.json")).getJSONObject("version_gate_sqs"); + JSONArray cases = section.getJSONArray("cases"); + JSONArray bodies = section.getJSONArray("incoming_bodies"); + assertEquals(17, cases.length()); + for (int i = 0; i < cases.length(); i++) { + JSONObject testCase = cases.getJSONObject(i); + for (int b = 0; b < bodies.length(); b++) { + runVersionGateCase(testCase, bodies.getJSONObject(b).getString("body")); + } + } + } + + private static void runVersionGateCase(JSONObject testCase, String body) { + String label = testCase.getString("name"); + boolean deadLetter = "on".equals(testCase.getString("dead_letter")); + JSONObject expect = testCase.getJSONObject("expect"); + + FakeSqsClient sqs = new FakeSqsClient(); + Map attributes = new java.util.HashMap<>(); + attributes.put("bq-schema-version", MessageAttributeValue.builder().dataType("Number").stringValue("2").build()); + if (testCase.has("bq_attempts")) { + attributes.put("bq-attempts", MessageAttributeValue.builder().dataType("Number") + .stringValue(testCase.getString("bq_attempts")).build()); + } + // A high receive count: the rule applies on any delivery without retry, whatever SQS counts. + sqs.seed(URL, body, 9, attributes); + if ("fails".equals(testCase.optString("dlq_write", "ok"))) { + sqs.sendError = new IllegalStateException("dlq write failed"); + } + int[] handled = {0}; + java.util.List errors = new ArrayList<>(); + SqsConsumer.Builder builder = SqsConsumer.builder(sqs, URL) + .handler("urn:babel:orders:created", (env, message) -> handled[0]++) + .onError((err, env, msg) -> errors.add(err)); + if (testCase.has("on_unknown_urn")) { + builder.unknownUrnStrategy(testCase.getString("on_unknown_urn")); + } + if (deadLetter) { + builder.deadLetterQueueUrl(DLQ_URL); + } + builder.build().poll(); + + assertEquals(0, handled[0], label); + // "off" means no DLQ writer exists in this consumer (no queue URL): outcome_no_writer applies. + String outcome = !deadLetter && expect.has("outcome_no_writer") + ? expect.getString("outcome_no_writer") : expect.getString("outcome"); + assertTrue(sqs.visibilityChanges.isEmpty(), label + ": a version-gated delivery is never released"); + switch (outcome) { + case "dlq_raw" -> { + assertEquals(1, sqs.sent.size(), label); + assertEquals(DLQ_URL, sqs.sent.get(0).queueUrl(), label); + assertEquals(body, sqs.sent.get(0).messageBody(), label); + assertTrue(!sqs.sent.get(0).messageBody().contains("\"dead_letter\""), label); + assertEquals(1, sqs.deleted.size(), label); + } + case "ack" -> { + assertTrue(sqs.sent.isEmpty(), label); + assertEquals(1, sqs.deleted.size(), label); + } + case "unsettled" -> { + assertTrue(sqs.sent.isEmpty(), label); + assertTrue(sqs.deleted.isEmpty(), label); + // The write failure is reported on top of the gate rejection itself. + assertTrue(errors.size() >= 2, label + ": the failed write must be reported"); + } + case "native" -> { + assertTrue(sqs.sent.isEmpty(), label); + assertTrue(sqs.deleted.isEmpty(), label); + assertEquals(1, errors.size(), label + ": reported, nothing written"); + } + default -> throw new AssertionError("unknown outcome in " + label + ": " + outcome); + } + } } diff --git a/src/test/java/com/babelqueue/sqs/SqsConsumerTest.java b/src/test/java/com/babelqueue/sqs/SqsConsumerTest.java index e930945..27f51f2 100644 --- a/src/test/java/com/babelqueue/sqs/SqsConsumerTest.java +++ b/src/test/java/com/babelqueue/sqs/SqsConsumerTest.java @@ -450,4 +450,281 @@ void schemaVersionGateRejectsWithoutDecodingOrHandlerAndLeavesMessage() { assertTrue(sqs.visibilityChanges.isEmpty(), "version=" + version); } } + // ---- ADR-0037: poison, version gate and the DLQ queue --------------------------------------- + + private static final String DLQ_URL = "https://sqs.eu-central-1.amazonaws.com/123456789012/orders.dlq"; + private static final String FIFO_DLQ_URL = "https://sqs.eu-central-1.amazonaws.com/123456789012/orders.dlq.fifo"; + + private static SqsConsumer.Builder consumerWithDlq(FakeSqsClient sqs, List errors, int[] handled) { + return SqsConsumer.builder(sqs, URL) + .handler("urn:babel:orders:created", (e, m) -> handled[0]++) + .onError((err, env, msg) -> errors.add(err)) + .deadLetterQueueUrl(DLQ_URL); + } + + private static org.json.JSONObject sentJson(FakeSqsClient sqs, int index) { + return new org.json.JSONObject(sqs.sent.get(index).messageBody()); + } + + @Test + void poisonBodyWithDlqWritesTheAdr0035RecordThenDeletes() { + FakeSqsClient sqs = new FakeSqsClient(); + sqs.seed(URL, "{\"not\":\"an envelope\",\"big\":12345678901234567890}", 4); + List errors = new ArrayList<>(); + int[] handled = {0}; + + consumerWithDlq(sqs, errors, handled).build().poll(); + + assertEquals(0, handled[0]); + assertEquals(1, sqs.sent.size()); + assertEquals(DLQ_URL, sqs.sent.get(0).queueUrl()); + // An object body is spliced: every original member stays as written, plus the dead_letter block. + assertTrue(sqs.sent.get(0).messageBody().contains("\"big\":12345678901234567890")); + org.json.JSONObject block = sentJson(sqs, 0).getJSONObject("dead_letter"); + assertEquals("poison", block.getString("reason")); + assertEquals("orders", block.getString("original_queue")); + assertEquals(3, block.getInt("attempts")); + assertEquals(1, sqs.deleted.size()); + assertTrue(sqs.visibilityChanges.isEmpty()); + assertEquals(1, errors.size()); + } + + @Test + void poisonNonJsonBodyIsWrappedUnderRaw() { + FakeSqsClient sqs = new FakeSqsClient(); + sqs.seed(URL, "plain text, not JSON", 1); + + consumerWithDlq(sqs, new ArrayList<>(), new int[1]).build().poll(); + + org.json.JSONObject record = sentJson(sqs, 0); + assertEquals("plain text, not JSON", record.getString("raw")); + assertEquals("poison", record.getJSONObject("dead_letter").getString("reason")); + assertEquals(1, sqs.deleted.size()); + } + + @Test + void poisonWithUnsupportedMetaSchemaVersionIsDeadLettered() { + FakeSqsClient sqs = new FakeSqsClient(); + String body = validBody().replace("\"schema_version\":1", "\"schema_version\":2"); + sqs.seed(URL, body, 1); + int[] handled = {0}; + + consumerWithDlq(sqs, new ArrayList<>(), handled).build().poll(); + + assertEquals(0, handled[0]); + assertEquals("poison", sentJson(sqs, 0).getJSONObject("dead_letter").getString("reason")); + assertEquals(1, sqs.deleted.size()); + } + + @Test + void poisonFailedDlqWriteLeavesItUnsettledAndReports() { + FakeSqsClient sqs = new FakeSqsClient(); + sqs.seed(URL, "{}", 1); + sqs.sendError = new IllegalStateException("dlq down"); + List errors = new ArrayList<>(); + + consumerWithDlq(sqs, errors, new int[1]).build().poll(); + + assertTrue(sqs.deleted.isEmpty()); + assertTrue(sqs.visibilityChanges.isEmpty()); + assertEquals(2, errors.size()); + assertEquals("dlq down", errors.get(1).getMessage()); + } + + @Test + void poisonWithoutDlqIsLeftToTheNativeMechanism() { + FakeSqsClient sqs = new FakeSqsClient(); + sqs.seed(URL, "{}", 1); + List errors = new ArrayList<>(); + + SqsConsumer.builder(sqs, URL).onError((err, env, msg) -> errors.add(err)).build().poll(); + + assertEquals(1, errors.size()); + assertTrue(sqs.sent.isEmpty()); + assertTrue(sqs.deleted.isEmpty()); + assertTrue(sqs.visibilityChanges.isEmpty()); + } + + @Test + void poisonWithDeleteStrategyIsDeletedWithoutARecord() { + FakeSqsClient sqs = new FakeSqsClient(); + sqs.seed(URL, "{}", 1); + + consumerWithDlq(sqs, new ArrayList<>(), new int[1]).unknownUrnStrategy(UnknownUrnStrategy.DELETE).build().poll(); + + assertTrue(sqs.sent.isEmpty()); + assertEquals(1, sqs.deleted.size()); + } + + @Test + void poisonWithDeadLetterStrategyButNoDlqQueueIsNative() { + FakeSqsClient sqs = new FakeSqsClient(); + sqs.seed(URL, "{}", 1); + + SqsConsumer.builder(sqs, URL).unknownUrnStrategy(UnknownUrnStrategy.DEAD_LETTER).build().poll(); + + assertTrue(sqs.sent.isEmpty()); + assertTrue(sqs.deleted.isEmpty()); + } + + @Test + void poisonToAFifoDlqCarriesGroupAndDeduplicationIds() { + FakeSqsClient sqs = new FakeSqsClient(); + sqs.seed(URL, "{}", 1); + + SqsConsumer.builder(sqs, URL).deadLetterQueueUrl(FIFO_DLQ_URL).build().poll(); + + SendMessageRequest sent = sqs.sent.get(0); + assertEquals(FIFO_DLQ_URL, sent.queueUrl()); + assertEquals("orders", sent.messageGroupId()); + assertNotNull(sent.messageDeduplicationId()); + assertTrue(sent.messageDeduplicationId().length() <= 128); + assertEquals(1, sqs.deleted.size()); + } + + @Test + void versionGatedWithDlqWritesTheRawBodyThenDeletes() { + FakeSqsClient sqs = new FakeSqsClient(); + // Deliberately non-canonical: a writer that decodes and re-encodes would change these bytes. + String raw = "{\"meta\": {\"schema_version\": 2}, \"job\": \"urn:babel:orders:created\", \"data\": {\"n\": 1.50}}\n"; + sendWithVersion(sqs, raw, "2"); + sqs.sent.clear(); + List errors = new ArrayList<>(); + int[] handled = {0}; + int[] unknown = {0}; + + consumerWithDlq(sqs, errors, handled).onUnknownUrn((e, m) -> unknown[0]++).build().poll(); + + assertEquals(0, handled[0]); + assertEquals(0, unknown[0]); + assertEquals(1, sqs.sent.size()); + assertEquals(DLQ_URL, sqs.sent.get(0).queueUrl()); + assertEquals(raw, sqs.sent.get(0).messageBody()); + assertTrue(!sqs.sent.get(0).messageBody().contains("dead_letter")); + assertEquals(1, sqs.deleted.size()); + assertEquals(1, errors.size()); + } + + @Test + void versionGatedIsSettledOnTheFirstDeliveryWhateverTheReceiveCountSays() { + FakeSqsClient sqs = new FakeSqsClient(); + sqs.seed(URL, "body", 9, Map.of("bq-schema-version", + MessageAttributeValue.builder().dataType("Number").stringValue("2").build())); + + consumerWithDlq(sqs, new ArrayList<>(), new int[1]).build().poll(); + + assertEquals(1, sqs.sent.size()); + assertEquals(1, sqs.deleted.size()); + } + + @Test + void versionGatedFailedDlqWriteLeavesItUnsettledAndReports() { + FakeSqsClient sqs = new FakeSqsClient(); + sendWithVersion(sqs, "not json", "2"); + sqs.sendError = new IllegalStateException("dlq down"); + List errors = new ArrayList<>(); + + consumerWithDlq(sqs, errors, new int[1]).build().poll(); + + assertTrue(sqs.deleted.isEmpty()); + assertTrue(sqs.visibilityChanges.isEmpty()); + assertEquals(2, errors.size()); + } + + @Test + void versionGatedWithDeleteStrategyIsDeletedWithoutARecord() { + FakeSqsClient sqs = new FakeSqsClient(); + sendWithVersion(sqs, "not json", "2"); + sqs.sent.clear(); + + consumerWithDlq(sqs, new ArrayList<>(), new int[1]).unknownUrnStrategy(UnknownUrnStrategy.DELETE).build().poll(); + + assertTrue(sqs.sent.isEmpty()); + assertEquals(1, sqs.deleted.size()); + } + + @Test + void versionGatedWithDeadLetterStrategyButNoDlqQueueIsNative() { + FakeSqsClient sqs = new FakeSqsClient(); + sendWithVersion(sqs, "not json", "2"); + sqs.sent.clear(); + + SqsConsumer.builder(sqs, URL).unknownUrnStrategy(UnknownUrnStrategy.DEAD_LETTER).build().poll(); + + assertTrue(sqs.sent.isEmpty()); + assertTrue(sqs.deleted.isEmpty()); + assertTrue(sqs.visibilityChanges.isEmpty()); + } + + @Test + void hookOnlyConsumerNeverDropsAVersionGatedDeliverySilently() { + // ADR-0037 "Configured strategy only": the delete that onUnknownUrn implies for an unknown URN must not + // reach the version gate. DLQ off -> native (not deleted); DLQ on -> raw record, and the hook never runs. + int[] unknown = {0}; + + FakeSqsClient off = new FakeSqsClient(); + sendWithVersion(off, validBody(), "2"); + off.sent.clear(); + SqsConsumer.builder(off, URL).onUnknownUrn((e, m) -> unknown[0]++).build().poll(); + assertTrue(off.deleted.isEmpty()); + assertTrue(off.sent.isEmpty()); + + FakeSqsClient on = new FakeSqsClient(); + String body = validBody(); + sendWithVersion(on, body, "2"); + on.sent.clear(); + SqsConsumer.builder(on, URL).onUnknownUrn((e, m) -> unknown[0]++).deadLetterQueueUrl(DLQ_URL).build().poll(); + assertEquals(1, on.sent.size()); + assertEquals(body, on.sent.get(0).messageBody()); + assertEquals(1, on.deleted.size()); + assertEquals(0, unknown[0]); + } + + @Test + void hookOnlyConsumerNeverDropsAPoisonBodySilently() { + FakeSqsClient off = new FakeSqsClient(); + off.seed(URL, "{}", 1); + SqsConsumer.builder(off, URL).onUnknownUrn((e, m) -> { }).build().poll(); + assertTrue(off.deleted.isEmpty()); + + FakeSqsClient on = new FakeSqsClient(); + on.seed(URL, "{}", 1); + SqsConsumer.builder(on, URL).onUnknownUrn((e, m) -> { }).deadLetterQueueUrl(DLQ_URL).build().poll(); + assertEquals("poison", sentJson(on, 0).getJSONObject("dead_letter").getString("reason")); + assertEquals(1, on.deleted.size()); + } + + @Test + void unknownUrnDeadLetterStrategyWritesTheAnnotatedEnvelopeThenDeletes() { + FakeSqsClient sqs = new FakeSqsClient(); + sqs.seed(URL, validBody(), 1); + + SqsConsumer.builder(sqs, URL).unknownUrnStrategy(UnknownUrnStrategy.DEAD_LETTER) + .deadLetterQueueUrl(DLQ_URL).build().poll(); + + assertEquals(1, sqs.sent.size()); + assertEquals("unknown_urn", sentJson(sqs, 0).getJSONObject("dead_letter").getString("reason")); + assertEquals("urn:babel:orders:created", sentJson(sqs, 0).getString("job")); + assertEquals(1, sqs.deleted.size()); + } + + @Test + void unknownUrnDeadLetterFailedWriteLeavesItUnsettled() { + FakeSqsClient sqs = new FakeSqsClient(); + sqs.seed(URL, validBody(), 1); + sqs.sendError = new IllegalStateException("dlq down"); + List errors = new ArrayList<>(); + + SqsConsumer.builder(sqs, URL).unknownUrnStrategy(UnknownUrnStrategy.DEAD_LETTER) + .deadLetterQueueUrl(DLQ_URL).onError((err, env, msg) -> errors.add(err)).build().poll(); + + assertTrue(sqs.deleted.isEmpty()); + assertEquals(1, errors.size()); + } + + @Test + void blankDeadLetterQueueUrlIsRejected() { + assertThrows(IllegalArgumentException.class, + () -> SqsConsumer.builder(new FakeSqsClient(), URL).deadLetterQueueUrl(" ")); + } } diff --git a/src/test/resources/conformance/CONFORMANCE_VERSION b/src/test/resources/conformance/CONFORMANCE_VERSION index dc1e644..27f9cd3 100644 --- a/src/test/resources/conformance/CONFORMANCE_VERSION +++ b/src/test/resources/conformance/CONFORMANCE_VERSION @@ -1 +1 @@ -1.6.0 +1.8.0 diff --git a/src/test/resources/conformance/manifest.json b/src/test/resources/conformance/manifest.json index 63f7ca1..4b2accb 100644 --- a/src/test/resources/conformance/manifest.json +++ b/src/test/resources/conformance/manifest.json @@ -1,6 +1,6 @@ { "schema_version": 1, - "conformance_version": "1.6.0", + "conformance_version": "1.8.0", "sections": { "cases": { "scope": "core" @@ -79,6 +79,24 @@ }, "poison_dlq": { "scope": "core" + }, + "identity_verdict": { + "scope": "core" + }, + "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": [ @@ -5798,5 +5816,1508 @@ } } ] + }, + "identity_verdict": { + "description": "Body classification outside Kafka (ADR-0036 decision 4; ADR-0032 section 6; error-handling.md section 3). Scope: `core`, because the decoder verdict is the same for every binding. For each case, feed `body` (a JSON string; it is the raw message body, so a case may hold invalid JSON, a BOM or duplicate keys) to the SDK core decoder and assert the verdict `expect`: `ok` - the body decodes into an envelope with a valid identity; `poison` - it does not (a decode error, or no valid identity). On a transport other than Kafka a `poison` body is never handed to a handler: it takes the poison route (broker-bindings.md section 5.5 and error-handling.md section 7; ADR-0035 record with `reason: poison`), not the unknown-URN strategies of error-handling.md section 3, which remain for a valid envelope whose URN has no handler. The case set starts with EVERY `kafka_poison` case (same names, bodies and order, the first 92 cases), with `no_identity` mapped to `poison` and `ok` / `poison` unchanged, so the two sections stay in step; the 5 cases after them are extra rows that pin the ADR-0036 decision 4 bodies `kafka_poison` does not hold (a blank or non-string `job` next to a valid `urn`, and a tab-only `job`) (a body with a valid identity but a wrongly typed known member is `poison` in both); a CI oracle derives the first 92 cases from `kafka_poison`, fails on any difference, and checks the extra rows with its own identity rule. The duplicate-key rule (the LAST value of a repeated top-level key wins) and the identity rule are those of `kafka_poison`: a `job` that is not a string is ignored; a blank (empty or whitespace-only) string `job` falls back to `urn`; a `urn` that is not a non-blank string is ignored; when neither is valid the body has no identity. Rows that pin the ADR-0036 decision 4 table (a `job` of `\"\"`, `\" \\t\"`, `null` or `{}` next to a valid `urn` is `ok`; see also the extra rows): a valid envelope (`full valid`) and a body whose only valid identity is `urn` (`urn alias string`), a blank `job` with a valid `urn` (`job blank with valid urn alias`), a non-string `job` with a valid `urn` (`job number with valid urn alias`) are `ok`; `{}` (`empty object`), a non-string `job` without `urn` (`job number`, `job null`), a non-string `urn` without `job` (`urn alias number`), non-object top-level values (`null literal`, `empty array`, `string`, `number`) and a non-JSON or empty body (`not json`, `empty`) are `poison`. The edge that a transport routes on an out-of-band identity property (Pulsar / Kafka `bq-job`) is outside this section.", + "cases": [ + { + "name": "empty", + "body": "", + "expect": "poison" + }, + { + "name": "whitespace only", + "body": " ", + "expect": "poison" + }, + { + "name": "not json", + "body": "not-json", + "expect": "poison" + }, + { + "name": "truncated", + "body": "{\"job\":\"urn:a\"", + "expect": "poison" + }, + { + "name": "empty array", + "body": "[]", + "expect": "poison" + }, + { + "name": "array", + "body": "[1]", + "expect": "poison" + }, + { + "name": "string", + "body": "\"str\"", + "expect": "poison" + }, + { + "name": "number", + "body": "123", + "expect": "poison" + }, + { + "name": "boolean", + "body": "true", + "expect": "poison" + }, + { + "name": "null literal", + "body": "null", + "expect": "poison" + }, + { + "name": "empty object", + "body": "{}", + "expect": "poison" + }, + { + "name": "full valid", + "body": "{\"job\":\"urn:babel:orders:created\",\"trace_id\":\"t\",\"data\":{},\"meta\":{\"id\":\"i\",\"queue\":\"q\",\"lang\":\"php\",\"schema_version\":1,\"created_at\":1700000000000},\"attempts\":0}", + "expect": "ok" + }, + { + "name": "job number", + "body": "{\"job\":5}", + "expect": "poison" + }, + { + "name": "job null", + "body": "{\"job\":null}", + "expect": "poison" + }, + { + "name": "urn alias number", + "body": "{\"urn\":5}", + "expect": "poison" + }, + { + "name": "urn alias string", + "body": "{\"urn\":\"urn:x\"}", + "expect": "ok" + }, + { + "name": "urn alias padded", + "body": "{\"urn\":\" urn:x \"}", + "expect": "ok" + }, + { + "name": "trace_id number", + "body": "{\"trace_id\":7}", + "expect": "poison" + }, + { + "name": "data array", + "body": "{\"data\":[]}", + "expect": "poison" + }, + { + "name": "data string", + "body": "{\"data\":\"x\"}", + "expect": "poison" + }, + { + "name": "data null", + "body": "{\"data\":null}", + "expect": "poison" + }, + { + "name": "meta string", + "body": "{\"meta\":\"x\"}", + "expect": "poison" + }, + { + "name": "meta array", + "body": "{\"meta\":[]}", + "expect": "poison" + }, + { + "name": "meta null", + "body": "{\"meta\":null}", + "expect": "poison" + }, + { + "name": "meta.id number", + "body": "{\"meta\":{\"id\":5}}", + "expect": "poison" + }, + { + "name": "meta.queue number", + "body": "{\"meta\":{\"queue\":5}}", + "expect": "poison" + }, + { + "name": "meta.lang boolean", + "body": "{\"meta\":{\"lang\":true}}", + "expect": "poison" + }, + { + "name": "meta.created_at string", + "body": "{\"meta\":{\"created_at\":\"2026\"}}", + "expect": "poison" + }, + { + "name": "meta.created_at fractional", + "body": "{\"meta\":{\"created_at\":1.5}}", + "expect": "poison" + }, + { + "name": "meta.created_at integer", + "body": "{\"meta\":{\"created_at\":1700000000000}}", + "expect": "poison" + }, + { + "name": "meta.created_at exponent", + "body": "{\"meta\":{\"created_at\":1e3}}", + "expect": "poison" + }, + { + "name": "meta.created_at decimal zero", + "body": "{\"meta\":{\"created_at\":1.0}}", + "expect": "poison" + }, + { + "name": "meta.created_at int64 overflow", + "body": "{\"meta\":{\"created_at\":9223372036854775808}}", + "expect": "poison" + }, + { + "name": "meta.created_at int64 max", + "body": "{\"meta\":{\"created_at\":9223372036854775807}}", + "expect": "poison" + }, + { + "name": "meta.created_at negative", + "body": "{\"meta\":{\"created_at\":-1}}", + "expect": "poison" + }, + { + "name": "meta.created_at negative zero", + "body": "{\"meta\":{\"created_at\":-0}}", + "expect": "poison" + }, + { + "name": "meta.schema_version string", + "body": "{\"meta\":{\"schema_version\":\"1\"}}", + "expect": "poison" + }, + { + "name": "meta.schema_version two", + "body": "{\"meta\":{\"schema_version\":2}}", + "expect": "poison" + }, + { + "name": "meta.schema_version decimal zero", + "body": "{\"meta\":{\"schema_version\":1.0}}", + "expect": "poison" + }, + { + "name": "meta.schema_version null", + "body": "{\"meta\":{\"schema_version\":null}}", + "expect": "poison" + }, + { + "name": "meta.schema_version boolean", + "body": "{\"meta\":{\"schema_version\":true}}", + "expect": "poison" + }, + { + "name": "attempts fractional", + "body": "{\"attempts\":1.5}", + "expect": "poison" + }, + { + "name": "attempts string", + "body": "{\"attempts\":\"3\"}", + "expect": "poison" + }, + { + "name": "attempts integer", + "body": "{\"attempts\":2}", + "expect": "poison" + }, + { + "name": "attempts exponent", + "body": "{\"attempts\":1e0}", + "expect": "poison" + }, + { + "name": "attempts upper exponent", + "body": "{\"attempts\":1E2}", + "expect": "poison" + }, + { + "name": "attempts negative zero", + "body": "{\"attempts\":-0}", + "expect": "poison" + }, + { + "name": "attempts overflow", + "body": "{\"attempts\":9223372036854775808}", + "expect": "poison" + }, + { + "name": "attempts null", + "body": "{\"attempts\":null}", + "expect": "poison" + }, + { + "name": "dead_letter string", + "body": "{\"dead_letter\":\"x\"}", + "expect": "poison" + }, + { + "name": "dead_letter null", + "body": "{\"dead_letter\":null}", + "expect": "poison" + }, + { + "name": "dead_letter empty object", + "body": "{\"dead_letter\":{}}", + "expect": "poison" + }, + { + "name": "dead_letter.attempts string", + "body": "{\"dead_letter\":{\"attempts\":\"3\"}}", + "expect": "poison" + }, + { + "name": "dead_letter.reason number", + "body": "{\"dead_letter\":{\"reason\":5}}", + "expect": "poison" + }, + { + "name": "dead_letter.error number", + "body": "{\"dead_letter\":{\"error\":5}}", + "expect": "poison" + }, + { + "name": "dead_letter nullable strings", + "body": "{\"dead_letter\":{\"error\":null,\"exception\":null}}", + "expect": "poison" + }, + { + "name": "dead_letter.failed_at string", + "body": "{\"dead_letter\":{\"failed_at\":\"x\"}}", + "expect": "poison" + }, + { + "name": "dead_letter.failed_at fractional", + "body": "{\"dead_letter\":{\"failed_at\":1.5}}", + "expect": "poison" + }, + { + "name": "dead_letter.original_queue array", + "body": "{\"dead_letter\":{\"original_queue\":[]}}", + "expect": "poison" + }, + { + "name": "dead_letter.lang object", + "body": "{\"dead_letter\":{\"lang\":{}}}", + "expect": "poison" + }, + { + "name": "dead_letter full", + "body": "{\"dead_letter\":{\"error\":\"e\",\"exception\":\"x\",\"failed_at\":1,\"original_queue\":\"q\",\"attempts\":1,\"lang\":\"go\",\"reason\":\"r\"}}", + "expect": "poison" + }, + { + "name": "upper-case job key with number is an unknown member (names are case-sensitive)", + "body": "{\"JOB\":5}", + "expect": "poison" + }, + { + "name": "mixed-case meta.id key with number is an unknown member (names are case-sensitive)", + "body": "{\"Meta\":{\"ID\":5}}", + "expect": "poison" + }, + { + "name": "upper-case job key with string", + "body": "{\"JOB\":\"urn:x\"}", + "expect": "poison" + }, + { + "name": "trailing garbage", + "body": "{\"job\":\"x\"} trailing", + "expect": "poison" + }, + { + "name": "two objects", + "body": "{\"job\":\"x\"}{\"job\":\"y\"}", + "expect": "poison" + }, + { + "name": "unknown keys", + "body": "{\"x\":1}", + "expect": "poison" + }, + { + "name": "forbidden keys with odd types", + "body": "{\"timestamp\":\"x\",\"meta\":{\"max_retries\":\"x\",\"attempts\":\"y\",\"source\":5,\"ts\":\"z\"}}", + "expect": "poison" + }, + { + "name": "nested lookalikes in data", + "body": "{\"data\":{\"a\":[1,2,{\"b\":null}],\"meta\":5,\"job\":5},\"extra\":{\"meta\":5}}", + "expect": "poison" + }, + { + "name": "duplicate job string then number", + "body": "{\"job\":\"urn:x\",\"job\":5}", + "expect": "poison" + }, + { + "name": "duplicate job number then string", + "body": "{\"job\":5,\"job\":\"urn:x\"}", + "expect": "ok" + }, + { + "name": "byte order mark", + "body": "\ufeff{\"job\":\"x\"}", + "expect": "poison" + }, + { + "name": "leading zero number", + "body": "{\"attempts\":01}", + "expect": "poison" + }, + { + "name": "NaN literal", + "body": "{\"attempts\":NaN}", + "expect": "poison" + }, + { + "name": "single quotes", + "body": "{'a':1}", + "expect": "poison" + }, + { + "name": "meta duplicate-case key", + "body": "{\"meta\":{\"id\":\"a\",\"queue\":\"b\",\"lang\":\"c\",\"schema_version\":1,\"created_at\":1,\"ID\":\"dup\"}}", + "expect": "poison" + }, + { + "name": "attempts huge exponent", + "body": "{\"attempts\":1e400}", + "expect": "poison" + }, + { + "name": "data with huge number", + "body": "{\"data\":{\"n\":1e400}}", + "expect": "poison" + }, + { + "name": "job number with valid urn alias", + "body": "{\"job\":5,\"urn\":\"urn:x\"}", + "expect": "ok" + }, + { + "name": "job blank with valid urn alias", + "body": "{\"job\":\" \",\"urn\":\"urn:x\"}", + "expect": "ok" + }, + { + "name": "job valid with urn number", + "body": "{\"job\":\"urn:x\",\"urn\":5}", + "expect": "ok" + }, + { + "name": "job blank without urn", + "body": "{\"job\":\" \"}", + "expect": "poison" + }, + { + "name": "job and urn both empty strings", + "body": "{\"job\":\"\",\"urn\":\"\"}", + "expect": "poison" + }, + { + "name": "job and urn both arrays", + "body": "{\"job\":[],\"urn\":[]}", + "expect": "poison" + }, + { + "name": "job object without urn", + "body": "{\"job\":{}}", + "expect": "poison" + }, + { + "name": "identity with null data", + "body": "{\"job\":\"urn:x\",\"data\":null}", + "expect": "ok" + }, + { + "name": "identity with null meta", + "body": "{\"job\":\"urn:x\",\"meta\":null}", + "expect": "ok" + }, + { + "name": "identity with null attempts", + "body": "{\"job\":\"urn:x\",\"attempts\":null}", + "expect": "ok" + }, + { + "name": "identity with null dead_letter", + "body": "{\"job\":\"urn:x\",\"dead_letter\":null}", + "expect": "ok" + }, + { + "name": "identity with integer attempts", + "body": "{\"job\":\"urn:x\",\"attempts\":2}", + "expect": "ok" + }, + { + "name": "identity with wrong typed attempts", + "body": "{\"job\":\"urn:x\",\"attempts\":\"3\"}", + "expect": "poison" + }, + { + "name": "identity with wrong typed data", + "body": "{\"job\":\"urn:x\",\"data\":[]}", + "expect": "poison" + }, + { + "name": "job empty string with valid urn", + "body": "{\"job\":\"\",\"urn\":\"urn:x\"}", + "expect": "ok" + }, + { + "name": "job space and tab with valid urn", + "body": "{\"job\":\" \\t\",\"urn\":\"urn:x\"}", + "expect": "ok" + }, + { + "name": "job null with valid urn", + "body": "{\"job\":null,\"urn\":\"urn:x\"}", + "expect": "ok" + }, + { + "name": "job object with valid urn", + "body": "{\"job\":{},\"urn\":\"urn:x\"}", + "expect": "ok" + }, + { + "name": "job space and tab without urn", + "body": "{\"job\":\" \\t\"}", + "expect": "poison" + } + ] + }, + "pulsar_exhaustion": { + "description": "Pulsar max-attempts settle decision (ADR-0036 decisions 1 and 2, amends ADR-0014; broker-bindings.md section 5.5 \"Max attempts and poison\"). Scope: `pulsar`. A pure function, no broker: for each case compute the outcome the consumer takes for a delivery and assert `expect`. Case `{name, kind, body_attempts?, bq_attempts?, redelivery_count, max_attempts?, dead_letter, on_unknown_urn?, expect}`. `kind`: `failed` (the handler throws), `unknown_urn` (an unmapped URN with no hook; strategy `fail`), `poison` (the body does not decode into an envelope, including an object with no valid identity, see `identity_verdict`). Inputs: `body_attempts` is the envelope's top-level `attempts` (absent for `poison`, which has no decodable body: its input is `bq_attempts` and `redelivery_count` only); `bq_attempts` is the `bq-attempts` message property, a string (absent = not set); `redelivery_count` is the native Pulsar RedeliveryCount (0-based); `max_attempts` absent = the default 3 (the total number of deliveries, the first included; a value below 1 is a SHOULD-level configuration error and is not a case); `dead_letter` is a boolean (dead-lettering on or off) and `off` has two meanings: in the Go and Python runtimes it is the dead-letter SETTING only, the DLQ target stays writable and the `dead_letter` strategy still writes a record; in the PHP `Consume\\Dispatcher` \"on\" means a `DeadLetterPublisher` is configured, so `off` means NO publisher and nothing can be written; `on_unknown_urn` absent = `fail`. Derivation: `reconciled` = `bq_attempts` when present and valid, else `body_attempts`, then `max(reconciled, redelivery_count)` (pulsar.attempts_reconciliation); `counted = reconciled + 1`. Decision. `failed` and `unknown_urn` (strategy `fail`): `counted < max_attempts` is `retry`; otherwise dead-lettering on is `dlq` (`reason` `failed` or `unknown_urn`, `dead_letter_attempts = counted`) and off is `drop` (acknowledge and log). `poison`: `on_unknown_urn: delete` is `drop`; `on_unknown_urn: dead_letter` is `dlq` with `reason: poison` whatever `dead_letter` says (the strategy writes the record where a DLQ writer exists; see `outcome_no_writer` below); dead-lettering on (strategy `fail` or `release`) is `dlq` with `reason: poison` and NO retry (the ADR-0035 record, `dead_letter_attempts` is not compared); dead-lettering off takes the bounded `fail` path (`release` is replaced by it) - `counted < max_attempts` is `retry`, otherwise `drop`. `unknown_urn` with `on_unknown_urn: delete` is `drop`; with `dead_letter` it is a direct `dlq` with `reason: unknown_urn` whatever `dead_letter` says (`dead_letter_attempts` is not compared; see `outcome_no_writer` below). `expect.outcome_no_writer` (optional, `drop`): present only on a row whose result differs for an SDK without a DLQ writer (the PHP `Dispatcher` without a publisher: `on_unknown_urn: dead_letter` is acknowledged, dropped and logged; owner decision 2026-10-25, ADR-0036 decision 3). A runner for such an SDK asserts `expect.outcome_no_writer` instead of `expect.outcome`; every other runner asserts `expect.outcome` and ignores the field; when the field is absent both kinds of SDK assert `expect.outcome`. On Kafka the PHP `Dispatcher` rethrows instead of dropping (outside this Pulsar-scoped section). Rows with any `on_unknown_urn` value (they are the rows with that key) apply only to an SDK that has the strategy option (the Go and Python runtimes and the PHP `Consume\\Dispatcher`, ADR-0036 decision 3); a runner for the Java, .NET or Node consumer, which have no such option, skips them and reports them as `skip`, not `pass`. `unknown_urn` with `release` is not a case: on Pulsar the Go and Python runtimes release it immediately without counting (the `unknown_urn_release_delay` known gap of broker-bindings.md section 5.5). Not covered here (no broker-less test): the DLQ write itself, the acknowledgement after a successful write, the negative acknowledgement after a failed one, and the PHP queue name of ADR-0036 decision 3; each SDK covers them in its own tests.", + "cases": [ + { + "name": "failed-reconciled-0-dlq-on-retries", + "kind": "failed", + "body_attempts": 0, + "redelivery_count": 0, + "dead_letter": true, + "expect": { + "outcome": "retry" + } + }, + { + "name": "failed-reconciled-1-dlq-on-retries", + "kind": "failed", + "body_attempts": 1, + "redelivery_count": 0, + "dead_letter": true, + "expect": { + "outcome": "retry" + } + }, + { + "name": "failed-reconciled-2-dlq-on-dead-letters", + "kind": "failed", + "body_attempts": 2, + "redelivery_count": 0, + "dead_letter": true, + "expect": { + "outcome": "dlq", + "reason": "failed", + "dead_letter_attempts": 3 + } + }, + { + "name": "failed-reconciled-2-dlq-off-drops", + "kind": "failed", + "body_attempts": 2, + "redelivery_count": 0, + "dead_letter": false, + "expect": { + "outcome": "drop" + } + }, + { + "name": "failed-reconciled-7-dlq-on-counts-past-the-cap", + "kind": "failed", + "body_attempts": 7, + "redelivery_count": 0, + "dead_letter": true, + "expect": { + "outcome": "dlq", + "reason": "failed", + "dead_letter_attempts": 8 + } + }, + { + "name": "failed-max-1-dlq-on-first-delivery-dead-letters", + "kind": "failed", + "body_attempts": 0, + "redelivery_count": 0, + "max_attempts": 1, + "dead_letter": true, + "expect": { + "outcome": "dlq", + "reason": "failed", + "dead_letter_attempts": 1 + } + }, + { + "name": "failed-max-1-dlq-off-first-delivery-drops", + "kind": "failed", + "body_attempts": 0, + "redelivery_count": 0, + "max_attempts": 1, + "dead_letter": false, + "expect": { + "outcome": "drop" + } + }, + { + "name": "failed-max-5-reconciled-3-dlq-off-retries", + "kind": "failed", + "body_attempts": 3, + "redelivery_count": 0, + "max_attempts": 5, + "dead_letter": false, + "expect": { + "outcome": "retry" + } + }, + { + "name": "failed-max-5-reconciled-4-dlq-off-drops", + "kind": "failed", + "body_attempts": 4, + "redelivery_count": 0, + "max_attempts": 5, + "dead_letter": false, + "expect": { + "outcome": "drop" + } + }, + { + "name": "unknown-urn-reconciled-1-dlq-on-retries", + "kind": "unknown_urn", + "body_attempts": 1, + "redelivery_count": 0, + "dead_letter": true, + "expect": { + "outcome": "retry" + } + }, + { + "name": "unknown-urn-reconciled-2-dlq-on-dead-letters", + "kind": "unknown_urn", + "body_attempts": 2, + "redelivery_count": 0, + "dead_letter": true, + "expect": { + "outcome": "dlq", + "reason": "unknown_urn", + "dead_letter_attempts": 3 + } + }, + { + "name": "unknown-urn-reconciled-2-dlq-off-drops", + "kind": "unknown_urn", + "body_attempts": 2, + "redelivery_count": 0, + "dead_letter": false, + "expect": { + "outcome": "drop" + } + }, + { + "name": "poison-reconciled-0-dlq-on-dead-letters-without-retry", + "kind": "poison", + "redelivery_count": 0, + "dead_letter": true, + "expect": { + "outcome": "dlq", + "reason": "poison" + } + }, + { + "name": "poison-reconciled-0-dlq-off-retries", + "kind": "poison", + "redelivery_count": 0, + "dead_letter": false, + "expect": { + "outcome": "retry" + } + }, + { + "name": "poison-reconciled-2-dlq-off-drops", + "kind": "poison", + "redelivery_count": 2, + "dead_letter": false, + "expect": { + "outcome": "drop" + } + }, + { + "name": "poison-delete-dlq-on-drops", + "kind": "poison", + "redelivery_count": 0, + "dead_letter": true, + "on_unknown_urn": "delete", + "expect": { + "outcome": "drop" + } + }, + { + "name": "poison-delete-dlq-off-drops", + "kind": "poison", + "redelivery_count": 0, + "dead_letter": false, + "on_unknown_urn": "delete", + "expect": { + "outcome": "drop" + } + }, + { + "name": "poison-release-dlq-on-dead-letters-without-retry", + "kind": "poison", + "redelivery_count": 0, + "dead_letter": true, + "on_unknown_urn": "release", + "expect": { + "outcome": "dlq", + "reason": "poison" + } + }, + { + "name": "poison-release-dlq-off-reconciled-0-takes-the-bounded-fail-path", + "kind": "poison", + "redelivery_count": 0, + "dead_letter": false, + "on_unknown_urn": "release", + "expect": { + "outcome": "retry" + } + }, + { + "name": "poison-release-dlq-off-reconciled-2-drops-not-released-forever", + "kind": "poison", + "redelivery_count": 2, + "dead_letter": false, + "on_unknown_urn": "release", + "expect": { + "outcome": "drop" + } + }, + { + "name": "poison-dead-letter-strategy-dlq-off-dead-letters", + "kind": "poison", + "redelivery_count": 0, + "dead_letter": false, + "on_unknown_urn": "dead_letter", + "expect": { + "outcome": "dlq", + "reason": "poison", + "outcome_no_writer": "drop" + } + }, + { + "name": "unknown-urn-delete-dlq-on-drops", + "kind": "unknown_urn", + "body_attempts": 0, + "redelivery_count": 0, + "dead_letter": true, + "on_unknown_urn": "delete", + "expect": { + "outcome": "drop" + } + }, + { + "name": "unknown-urn-dead-letter-strategy-dlq-off-dead-letters-directly", + "kind": "unknown_urn", + "body_attempts": 0, + "redelivery_count": 0, + "dead_letter": false, + "on_unknown_urn": "dead_letter", + "expect": { + "outcome": "dlq", + "reason": "unknown_urn", + "outcome_no_writer": "drop" + } + }, + { + "name": "poison-dead-letter-strategy-dlq-on-dead-letters", + "kind": "poison", + "redelivery_count": 0, + "dead_letter": true, + "on_unknown_urn": "dead_letter", + "expect": { + "outcome": "dlq", + "reason": "poison" + } + }, + { + "name": "unknown-urn-dead-letter-strategy-dlq-on-dead-letters-directly", + "kind": "unknown_urn", + "body_attempts": 0, + "redelivery_count": 0, + "dead_letter": true, + "on_unknown_urn": "dead_letter", + "expect": { + "outcome": "dlq", + "reason": "unknown_urn" + } + }, + { + "name": "reconciliation-bq-attempts-string-over-body", + "kind": "failed", + "body_attempts": 0, + "bq_attempts": "2", + "redelivery_count": 0, + "dead_letter": true, + "expect": { + "outcome": "dlq", + "reason": "failed", + "dead_letter_attempts": 3 + } + }, + { + "name": "reconciliation-redelivery-count-floors-body", + "kind": "failed", + "body_attempts": 1, + "redelivery_count": 2, + "dead_letter": true, + "expect": { + "outcome": "dlq", + "reason": "failed", + "dead_letter_attempts": 3 + } + }, + { + "name": "reconciliation-bq-attempts-below-cap-retries", + "kind": "unknown_urn", + "body_attempts": 0, + "bq_attempts": "1", + "redelivery_count": 0, + "dead_letter": true, + "expect": { + "outcome": "retry" + } + }, + { + "name": "reconciliation-poison-bq-attempts-dlq-off-drops", + "kind": "poison", + "bq_attempts": "2", + "redelivery_count": 0, + "dead_letter": false, + "expect": { + "outcome": "drop" + } + } + ] + }, + "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/src/test/resources/conformance/schema/results.example.json b/src/test/resources/conformance/schema/results.example.json index 9130d29..e2b7bb1 100644 --- a/src/test/resources/conformance/schema/results.example.json +++ b/src/test/resources/conformance/schema/results.example.json @@ -1,7 +1,7 @@ { "sdk": "example-sdk/redis", "version": "1.9.0", - "conformance_version": "1.6.0", + "conformance_version": "1.8.0", "generated_at": "2026-10-18T09:30:00Z", "bindings": [ "redis", From 847ed21b4958fa6c1ead643168db02b606e92ee5 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Muhammet=20=C5=9Eafak?= Date: Mon, 26 Oct 2026 07:23:42 +0300 Subject: [PATCH 2/3] Document -dlq as the SQS dead-letter queue name (ADR-0037) SQS queue names cannot contain a dot; the docs now name the DLQ -dlq (FIFO -dlq.fifo). --- CHANGELOG.md | 6 +++++- README.md | 4 ++-- src/main/java/com/babelqueue/sqs/SqsConsumer.java | 12 ++++++------ 3 files changed, 13 insertions(+), 9 deletions(-) diff --git a/CHANGELOG.md b/CHANGELOG.md index 0361d5b..c18b8ec 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -15,7 +15,7 @@ deliveries on their first delivery. The envelope is unchanged (`schema_version` ### Added - `SqsConsumer.Builder.deadLetterQueueUrl(url)` (opt-in, like the .NET `DeadLetterQueueUrl` and the Node - `deadLetterQueueUrl`): the `.dlq` queue (a FIFO one is `.dlq.fifo`; the consumer sends it + `deadLetterQueueUrl`): the `-dlq` queue (a FIFO one is `-dlq.fifo`; the consumer sends it the queue name as message group and the SQS message id as deduplication id). "Dead-lettering on" means this URL is set. The queue must exist. - With a DLQ queue set, the `dead_letter` unknown-URN strategy now writes the envelope, annotated with a @@ -43,6 +43,10 @@ deliveries on their first delivery. The envelope is unchanged (`schema_version` - Vendored the cross-SDK conformance suite v1.8.0 (`version_gate_sqs`, `poison_dlq` runners added) and moved the CI pin to it. +### Docs + +- The BabelQueue DLQ on SQS is `-dlq` (FIFO `-dlq.fifo`), not `.dlq`: an SQS queue name cannot contain a dot (ADR-0037 decision 7). README and Javadoc only; `deadLetterQueueUrl` is an explicit URL, so no code change. + ## [1.3.0] - 2026-10-24 ### Added diff --git a/README.md b/README.md index 83cdfe7..c8d1a2b 100644 --- a/README.md +++ b/README.md @@ -72,7 +72,7 @@ delete — observe via `onError` / `onUnknownUrn`. The envelope is unchanged Any other value that is not exactly `1` (no trimming — `"2"`, `"x"`, `" 1"`) → the body is **not** decoded, the handler is not called, no hook runs, `onError` is notified with a `null` envelope, and the message is settled on its **first** delivery (ADR-0037): with a `deadLetterQueueUrl` the body is sent to -`.dlq` **exactly as received** (no `dead_letter` block) and deleted after the send succeeded (a failed +`-dlq` **exactly as received** (no `dead_letter` block) and deleted after the send succeeded (a failed send leaves it for redelivery); under an explicit `delete` strategy it is deleted; without a DLQ queue it is left undeleted so SQS redrives it (`RedrivePolicy`). Only an explicitly set `unknownUrnStrategy` is read — an `onUnknownUrn` hook never drops a gated message. @@ -90,7 +90,7 @@ SqsConsumer.builder(sqs, url) //.retryBackoff(5).retryBackoffMultiplier(2).retryBackoffMax(300).retryJitter("full") .unknownUrnStrategy(UnknownUrnStrategy.RELEASE) // fail | delete | release | dead_letter .unknownUrnReleaseDelaySeconds(60) // default 0 - .deadLetterQueueUrl(dlqUrl) // opt-in BabelQueue DLQ (.dlq / .dlq.fifo) + .deadLetterQueueUrl(dlqUrl) // opt-in BabelQueue DLQ (-dlq / -dlq.fifo) .build(); ``` diff --git a/src/main/java/com/babelqueue/sqs/SqsConsumer.java b/src/main/java/com/babelqueue/sqs/SqsConsumer.java index 26e7fa5..9a821b8 100644 --- a/src/main/java/com/babelqueue/sqs/SqsConsumer.java +++ b/src/main/java/com/babelqueue/sqs/SqsConsumer.java @@ -48,7 +48,7 @@ * (see below), redelivered immediately under the default {@code 0} delay, ends in a native DLQ. * *

Poison and version gate (ADR-0037). "Dead-lettering on" means a {@link Builder#deadLetterQueueUrl DLQ - * queue URL} is configured (opt-in; a FIFO DLQ is {@code .dlq.fifo}). Only the explicitly configured + * queue URL} is configured (opt-in; a FIFO DLQ is {@code -dlq.fifo}). Only the explicitly configured * {@link Builder#unknownUrnStrategy strategy} is read by these two rules (none configured means {@code fail}); an * {@code onUnknownUrn} hook, and the {@code delete} it implies for an unknown URN, never affect them. *

    @@ -649,11 +649,11 @@ public Builder unknownUrnStrategy(String strategy) { } /** - * Turns BabelQueue dead-lettering on (opt-in, ADR-0037 decision 2): the URL of the {@code .dlq} queue - * (a FIFO one is {@code .dlq.fifo}). A poison body is sent to it as the ADR-0035 record, a - * version-gated delivery as its raw body, and an unknown URN under the {@code dead_letter} strategy as an - * annotated envelope; the source message is deleted only after the send succeeded. Unset (the default): a - * poison or version-gated delivery is left to the queue's native {@code RedrivePolicy}. The queue must + * Turns BabelQueue dead-lettering on (opt-in, ADR-0037 decision 2): the URL of the {@code -dlq} queue + * (an SQS name cannot contain a dot; a FIFO one is {@code -dlq.fifo}). A poison body is sent to it as the + * ADR-0035 record, a version-gated delivery as its raw body, and an unknown URN under the + * {@code dead_letter} strategy as an annotated envelope; the source message is deleted only after the send + * succeeded. Unset (the default): a poison or version-gated delivery is left to the queue's native {@code RedrivePolicy}. The queue must * exist; this consumer never creates it. */ public Builder deadLetterQueueUrl(String url) { From 73baf00c2cab76aad2dd4114e14c676d46796d26 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Muhammet=20=C5=9Eafak?= Date: Mon, 26 Oct 2026 08:10:59 +0300 Subject: [PATCH 3/3] Release v1.4.0 --- CHANGELOG.md | 2 ++ README.md | 2 +- pom.xml | 2 +- 3 files changed, 4 insertions(+), 2 deletions(-) diff --git a/CHANGELOG.md b/CHANGELOG.md index c18b8ec..661589d 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -9,6 +9,8 @@ The envelope wire format is versioned separately by `meta.schema_version` ## [Unreleased] +## [1.4.0] - 2026-10-26 + MINOR (ADR-0037): the consumer gains an opt-in BabelQueue DLQ and settles poison bodies and version-gated deliveries on their first delivery. The envelope is unchanged (`schema_version` stays `1`). diff --git a/README.md b/README.md index c8d1a2b..99fa91c 100644 --- a/README.md +++ b/README.md @@ -15,7 +15,7 @@ as the PHP/Laravel, Python, Go, Node and .NET SDKs. Implements com.babelqueue babelqueue-sqs - 1.3.0 + 1.4.0 ``` diff --git a/pom.xml b/pom.xml index 37ba3e4..b87c0af 100644 --- a/pom.xml +++ b/pom.xml @@ -6,7 +6,7 @@ com.babelqueue babelqueue-sqs - 1.3.0 + 1.4.0 jar BabelQueue Amazon SQS