Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension


Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
7 changes: 7 additions & 0 deletions CHANGELOG.md
Original file line number Diff line number Diff line change
Expand Up @@ -9,6 +9,13 @@ The envelope wire format is versioned separately by `meta.schema_version`

## [Unreleased]

## [1.14.0] - 2026-10-26

### Added
- `PoisonRecord.build(body, originalQueue, attempts, error, exception)` (ADR-0035): builds the poison `dead_letter`
block itself (`reason`, `error`, `exception`, `failed_at`, `original_queue`, `attempts`, `lang`) and delegates to
`build(body, deadLetter)`, so transports no longer hand-build it.

## [1.13.0] - 2026-10-24

### Added
Expand Down
2 changes: 1 addition & 1 deletion pom.xml
Original file line number Diff line number Diff line change
Expand Up @@ -6,7 +6,7 @@

<groupId>com.babelqueue</groupId>
<artifactId>babelqueue-core</artifactId>
<version>1.13.0</version>
<version>1.14.0</version>
<packaging>jar</packaging>

<name>BabelQueue Core</name>
Expand Down
35 changes: 35 additions & 0 deletions src/main/java/com/babelqueue/PoisonRecord.java
Original file line number Diff line number Diff line change
Expand Up @@ -5,6 +5,8 @@
import java.nio.charset.CodingErrorAction;
import java.nio.charset.StandardCharsets;
import java.util.Base64;
import java.util.LinkedHashMap;
import java.util.Map;

/**
* Builds the DLQ record of a poison body (ADR-0035, error-handling.md section 6).
Expand All @@ -16,13 +18,46 @@
* "dead_letter": ...}}: {@code raw} is the body as plain text when it is valid UTF-8, otherwise the standard padded
* base64 of the exact bytes with {@code "raw_encoding": "base64"}. U+FFFD substitution never happens.
*
* <p>Transports should call {@link #build(byte[], String, int, String, String)}: it assembles the ADR-0035
* {@code dead_letter} block itself (so no transport hand-builds it) and delegates here. The two-argument
* {@link #build(byte[], String)} stays for callers that already hold the block JSON.
*
* <p>The strict scan is iterative (no recursion), so a body nested thousands of levels deep cannot overflow the stack.
*/
public final class PoisonRecord {

private PoisonRecord() {
}

/**
* Builds the DLQ record of a poison body, assembling the ADR-0035 {@code dead_letter} block itself. Transports
* should use this instead of hand-building the block.
*
* <p>The block members are written in this order: {@code reason} ({@code "poison"}), {@code error},
* {@code exception}, {@code failed_at} (epoch milliseconds, taken now), {@code original_queue}, {@code attempts}
* and {@code lang} ({@link EnvelopeCodec#SOURCE_LANG}). A {@code null} {@code error} or {@code exception} is
* written as JSON {@code null}, not omitted. The block is then spliced or wrapped by
* {@link #build(byte[], String)}.
*
* @param body the exact record value bytes
* @param originalQueue the queue the body was consumed from
* @param attempts informational delivery attempts so far
* @param error the failure message, or {@code null}
* @param exception the failure type name, or {@code null}
* @return the DLQ record JSON text
*/
public static String build(byte[] body, String originalQueue, int attempts, String error, String exception) {
Map<String, Object> block = new LinkedHashMap<>();
block.put("reason", "poison");
block.put("error", error);
block.put("exception", exception);
block.put("failed_at", System.currentTimeMillis());
block.put("original_queue", originalQueue);
block.put("attempts", attempts);
block.put("lang", EnvelopeCodec.SOURCE_LANG);
return build(body, JsonValues.encode(block));
}

/**
* @param body the exact record value bytes
* @param deadLetter the JSON text of the {@code dead_letter} block: exactly one strict JSON object, spliced into
Expand Down
129 changes: 129 additions & 0 deletions src/test/java/com/babelqueue/PoisonRecordTest.java
Original file line number Diff line number Diff line change
@@ -0,0 +1,129 @@
package com.babelqueue;

import static org.junit.jupiter.api.Assertions.assertEquals;
import static org.junit.jupiter.api.Assertions.assertFalse;
import static org.junit.jupiter.api.Assertions.assertInstanceOf;
import static org.junit.jupiter.api.Assertions.assertNull;
import static org.junit.jupiter.api.Assertions.assertTrue;

import java.nio.charset.StandardCharsets;
import java.util.ArrayList;
import java.util.Base64;
import java.util.LinkedHashMap;
import java.util.List;
import java.util.Map;
import org.junit.jupiter.api.Test;

class PoisonRecordTest {

private static final List<String> BLOCK_ORDER =
List.of("reason", "error", "exception", "failed_at", "original_queue", "attempts", "lang");

private static byte[] bytes(String s) {
return s.getBytes(StandardCharsets.UTF_8);
}

@SuppressWarnings("unchecked")
private static Map<String, Object> parseObject(String json) {
return (Map<String, Object>) assertInstanceOf(Map.class, Json.parse(json));
}

/** Rebuilds the block by hand with the failed_at taken from the produced record, then compares bytes. */
private static void assertBytesEqualToManualBlock(byte[] body, String queue, int attempts, String error,
String exception) {
long before = System.currentTimeMillis();
String actual = PoisonRecord.build(body, queue, attempts, error, exception);
long after = System.currentTimeMillis();

Map<String, Object> dl = deadLetterOf(actual);
Object failedAt = dl.get("failed_at");
long ts = assertInstanceOf(Number.class, failedAt).longValue();
assertTrue(ts >= before && ts <= after, "failed_at is epoch ms taken during the call");

Map<String, Object> block = new LinkedHashMap<>();
block.put("reason", "poison");
block.put("error", error);
block.put("exception", exception);
block.put("failed_at", ts);
block.put("original_queue", queue);
block.put("attempts", attempts);
block.put("lang", EnvelopeCodec.SOURCE_LANG);
assertEquals(PoisonRecord.build(body, JsonValues.encode(block)), actual);
}

private static Map<String, Object> deadLetterOf(String record) {
Object dl = parseObject(record).get("dead_letter");
@SuppressWarnings("unchecked")
Map<String, Object> map = (Map<String, Object>) assertInstanceOf(Map.class, dl);
return map;
}

@Test
void overloadEqualsManuallyBuiltBlock() {
assertBytesEqualToManualBlock(bytes("{\"a\":1}"), "orders", 3, "boom", "java.lang.IllegalStateException");
}

@Test
void blockHasCanonicalOrderAndTypes() {
String record = PoisonRecord.build(bytes("{\"a\":1}"), "orders", 2, "boom", "X");
Map<String, Object> dl = deadLetterOf(record);
assertEquals(BLOCK_ORDER, new ArrayList<>(dl.keySet()));
assertEquals("poison", dl.get("reason"));
assertEquals("boom", dl.get("error"));
assertEquals("X", dl.get("exception"));
assertEquals("orders", dl.get("original_queue"));
assertEquals(2L, ((Number) dl.get("attempts")).longValue());
assertEquals("java", dl.get("lang"));
assertTrue(dl.get("failed_at") instanceof Number);
assertTrue(((Number) dl.get("failed_at")).longValue() > 1_600_000_000_000L);
}

@Test
void nullErrorAndExceptionAreWrittenAsNull() {
String record = PoisonRecord.build(bytes("{\"a\":1}"), "q", 0, null, null);
Map<String, Object> dl = deadLetterOf(record);
assertEquals(BLOCK_ORDER, new ArrayList<>(dl.keySet()));
assertTrue(dl.containsKey("error"));
assertTrue(dl.containsKey("exception"));
assertNull(dl.get("error"));
assertNull(dl.get("exception"));
assertTrue(record.contains("\"error\":null,\"exception\":null"));
assertBytesEqualToManualBlock(bytes("{\"a\":1}"), "q", 0, null, null);
}

@Test
void zeroAttempts() {
String record = PoisonRecord.build(bytes("{}"), "q", 0, "e", "E");
assertEquals(0L, ((Number) deadLetterOf(record).get("attempts")).longValue());
assertBytesEqualToManualBlock(bytes("{}"), "q", 0, "e", "E");
}

@Test
void invalidUtf8TakesBase64Branch() {
byte[] body = {(byte) 0xFF, (byte) 0xFE, 0x41};
String record = PoisonRecord.build(body, "q", 1, "bad", "E");
Map<String, Object> parsed = parseObject(record);
assertEquals(Base64.getEncoder().encodeToString(body), parsed.get("raw"));
assertEquals("base64", parsed.get("raw_encoding"));
assertEquals(BLOCK_ORDER, new ArrayList<>(deadLetterOf(record).keySet()));
assertBytesEqualToManualBlock(body, "q", 1, "bad", "E");
}

@Test
void spliceableObjectIsSplicedAtByteLevel() {
String body = "{\"n\":1e400,\"big\":18446744073709551616,\"n\":2}";
String record = PoisonRecord.build(bytes(body), "q", 4, "e", "E");
assertTrue(record.startsWith(body.substring(0, body.length() - 1) + ",\"dead_letter\":{"), record);
assertTrue(record.endsWith("}}"));
assertBytesEqualToManualBlock(bytes(body), "q", 4, "e", "E");
}

@Test
void nonObjectBodyIsWrappedAsRawText() {
String record = PoisonRecord.build(bytes("not json"), "q", 1, "e", "E");
Map<String, Object> parsed = parseObject(record);
assertEquals("not json", parsed.get("raw"));
assertFalse(parsed.containsKey("raw_encoding"));
assertBytesEqualToManualBlock(bytes("not json"), "q", 1, "e", "E");
}
}
Loading