Adapters & transports

Last updated: October 25, 2026

The @babelqueue/core codec is framework-agnostic. Thin packages wire it into common Node stacks: @babelqueue/bullmq (BullMQ jobs) and @babelqueue/nestjs (a NestJS module built on the BullMQ adapter), plus broker transports — @babelqueue/redis (Redis), @babelqueue/rabbitmq (RabbitMQ), @babelqueue/sqs (Amazon SQS), @babelqueue/azure-service-bus (Azure Service Bus), @babelqueue/pulsar (Apache Pulsar), @babelqueue/kafka (Apache Kafka) and @babelqueue/artemis (Apache ActiveMQ Artemis).

BullMQ — @babelqueue/bullmq

npm install @babelqueue/bullmq bullmq

bullmq ^5 is a peer dependency; @babelqueue/core is pulled in for you.

It exports two main functions, plus header helpers (headersOf, encodeMetadata, decodeMetadata):

• publish(queue, urn, data, options?) → Promise<string> — adds a BullMQ job whose name is the URN and whose data is the canonical envelope; returns meta.id. options is { traceId?, jobsOptions?, headers? } (jobsOptions is BullMQ’s own — delay, attempts, backoff…; headers is an out-of-band carrier such as a W3C traceparent, stored in the job’s telemetry.metadata beside the untouched envelope).
• processor(handlers, options?) → (job) => Promise — a BullMQ processor function that validates each envelope, resolves its URN and routes to handlers[urn]. options.onUnknownUrn(envelope, job) handles URNs with no mapped handler.

Produce

import { Queue } from "bullmq";
import { publish } from "@babelqueue/bullmq";

const queue = new Queue("orders", { connection: { host: "localhost", port: 6379 } });

const id = await publish(queue, "urn:babel:orders:created", { order_id: 1042 });

Consume

import { Worker } from "bullmq";
import { processor } from "@babelqueue/bullmq";

new Worker(
  "orders",
  processor(
    {
      "urn:babel:orders:created": async (env, job) => {
        console.log(env.data.order_id, env.trace_id);
      },
    },
    { onUnknownUrn: (env, job) => console.warn("no handler for", job.name) },
  ),
  { connection: { host: "localhost", port: 6379 } },
);

A handler is (envelope, job, headers) => unknown | Promise<unknown> (headers is the out-of-band carrier published with the job, empty when there is none; headersOf(job) reads it elsewhere). A non-conformant envelope is poison: it fails the job with BullMQ’s UnrecoverableError, so BullMQ does not retry it whatever the job’s attempts say, and the job data stays in the queue’s failed set. An unmapped URN throws unless onUnknownUrn is supplied, and a throwing handler is retried per the job’s options, as before.

Behaviour change (ADR-0037, since @babelqueue/bullmq 1.2.0). A poison job used to throw a BabelQueueError, which BullMQ retried as an ordinary failure up to the job’s attempts. Code that caught BabelQueueError from the processor for this case must now handle UnrecoverableError.

To keep a copy of poison jobs where a polyglot reader can find them, give the processor a deadLetterQueue:

import { Queue, Worker } from "bullmq";
import { processor } from "@babelqueue/bullmq";

new Worker(
  "orders",
  processor(handlers, {
    deadLetterQueue: true, // adds the poison record to "orders.dlq" before the job fails
  }),
  { connection },
);

deadLetterQueue is opt-in and takes a Queue (used as given), a queue name (opened on the worker’s connection) or true (<queue>.dlq, <queue> being the worker’s queue; set the processor’s queue option if the name should differ). The poison record has reason: "poison" and its body is the job data exactly as BullMQ delivered it, already parsed, serialised as JSON. A failed add is reported through onError and the job still fails as unrecoverable. The processor has a close() that closes a Queue the adapter created.

NestJS — @babelqueue/nestjs

npm install @babelqueue/nestjs @nestjs/common bullmq

@nestjs/common ^10 || ^11 and bullmq ^5 are peers; it builds on @babelqueue/bullmq.

Register the module and inject the publisher:

import { Module } from "@nestjs/common";
import { BabelQueueModule } from "@babelqueue/nestjs";

@Module({
  imports: [
    BabelQueueModule.forRoot({
      queue: "orders",
      connection: { host: "localhost", port: 6379 },
    }),
  ],
})
export class AppModule {}

import { Injectable } from "@nestjs/common";
import { BabelQueuePublisher } from "@babelqueue/nestjs";

@Injectable()
export class Orders {
  constructor(private readonly babelQueue: BabelQueuePublisher) {}

  create() {
    return this.babelQueue.publish("urn:babel:orders:created", { order_id: 1042 });
  }
}

forRoot({ queue, connection, queueOptions? }) provides an injectable BabelQueuePublisher (publish(urn, data, { traceId?, headers? }) → Promise<string>) over the BullMQ queue. For consuming, build a plain BullMQ Worker with the processor re-exported from @babelqueue/nestjs:

import { Worker } from "bullmq";
import { processor } from "@babelqueue/nestjs";

new Worker(
  "orders",
  processor({ "urn:babel:orders:created": async (env) => { /* ... */ } }),
  { connection: { host: "localhost", port: 6379 } },
);

Redis — @babelqueue/redis

npm install @babelqueue/redis ioredis

ioredis is an optional peer — you provide the client (an ioredis instance satisfies the adapter structurally). It implements the §1 reliable-queue pattern: the list element is the canonical envelope JSON, byte-for-byte, with no wrapping by default (unlike @babelqueue/bullmq, which uses BullMQ’s own job layout) and no property projection — Redis lists carry no native metadata, so routing and tracing read the body directly. Produce is RPUSH; consume reserves the head into a <queue>:processing list (LMOVE <queue> <queue>:processing LEFT RIGHT inside a Lua script, so an in-flight message survives a crash and the queue is FIFO), routes by URN, and LREMs it on success.

Out-of-band headers (e.g. traceparent) are opt-in: passing { headers } to publish stores a __bq_frame wrapper around the envelope; without headers the element is the bare envelope. The handler receives the carried headers as a third argument.

Produce

import Redis from "ioredis";
import { RedisPublisher } from "@babelqueue/redis";

const client = new Redis("redis://localhost:6379/0");

const id = await RedisPublisher.create(client, "orders")
  .publish("urn:babel:orders:created", { order_id: 1042 });

publish(urn, data, { traceId?, headers? }) returns the message meta.id.

Consume

import { RedisConsumer, type BabelHandlers } from "@babelqueue/redis";

const handlers: BabelHandlers = {
  "urn:babel:orders:created": (envelope, raw) => {
    console.log(envelope.data.order_id, envelope.trace_id);
  },
};

const consumer = new RedisConsumer(client, "orders", handlers, {
  maxTries: 3, // requeue with attempts+1, then <queue>.dlq
  onError: (err) => console.error(err),
});

await consumer.run(() => true); // reserve → process → LREM, until you stop it

A successful handler LREMs the element from <queue>:processing. A throwing handler requeues the envelope with attempts + 1 (at-least-once) up to maxTries, then dead-letters to <queue>.dlq with a dead_letter block — the body owns the attempt count. dead_letter.attempts is the counted value, the delivered attempts plus one for the current delivery, so with maxTries: 3 it is 3 (the same value the Go and Python runtimes write; before, this consumer wrote the delivered value, one lower). Unknown-URN strategy is one of fail / delete / release / dead_letter. A body that does not decode, or a JSON object with no valid job / urn, is poison: it never reaches a handler, whatever the strategy is, and with the default <queue>.dlq it is dead-lettered as a poison record (reason: "poison", original body kept).

deadLetterQueue: null turns off only the dead-letter of an exhausted message and of an unknown URN (they are acknowledged and dropped, and the dead_letter strategy degrades to delete). It no longer drops a poison body: that is still dead-lettered as the poison record, to the configured name, and with null to the default <queue>.dlq. This matches the Go and Python transports and the Redis binding.

Behaviour change (ADR-0037, since @babelqueue/redis 1.4.0). Until now deadLetterQueue: null acknowledged and dropped a poison body without a record. If you set null to keep poison bodies out of Redis, read or trim <queue>.dlq from now on.

This is a Node-owned reliable queue. Full parity with Laravel’s reserved-sorted-set reservation on a shared Redis queue is a separate task — for a mixed PHP+Node fleet, prefer a queue this consumer owns end-to-end. See the Redis binding.

RabbitMQ — @babelqueue/rabbitmq

npm install @babelqueue/rabbitmq amqplib

amqplib is an optional peer — you provide the channel (an amqplib Channel satisfies the adapter structurally). It implements §2 of the broker-bindings contract: the envelope JSON is the message body, and the contract fields are projected onto native AMQP 0-9-1 properties so a consumer routes without decoding the body — type = URN, correlation_id = trace_id, message_id = meta.id, app_id = babelqueue, plus the native-typed x-schema-version / x-source-lang / x-attempts headers (AMQP field-tables carry typed values, so integers stay integers). Consume is basic.get + manual ack (at-least-once).

Produce

import amqp from "amqplib";
import { RabbitMQPublisher } from "@babelqueue/rabbitmq";

const conn = await amqp.connect("amqp://guest:guest@localhost:5672/");
const channel = await conn.createChannel();
await channel.assertQueue("orders", { durable: true });

const id = await RabbitMQPublisher.create(channel, "orders")
  .publish("urn:babel:orders:created", { order_id: 1042 });

publish(urn, data, { traceId? }) returns the message meta.id. Messages are persistent (delivery_mode = 2).

Consume

import { RabbitMQConsumer, type BabelHandlers } from "@babelqueue/rabbitmq";

const handlers: BabelHandlers = {
  "urn:babel:orders:created": (envelope, message) => {
    console.log(envelope.data.order_id, envelope.trace_id);
  },
};

const consumer = new RabbitMQConsumer(channel, "orders", handlers, {
  maxTries: 3,
  onError: (err) => console.error(err),
});

await consumer.run(() => true); // basic.get → process → ack, until you stop it

A successful handler acks the message. A throwing handler republishes the envelope with attempts + 1 (at-least-once) up to maxTries, then dead-letters to <queue>.dlq with a dead_letter block. The consumer routes on properties.type (falling back to the body URN). Unknown-URN strategy is one of fail / delete / release / dead_letter. See the RabbitMQ binding.

Amazon SQS — @babelqueue/sqs

npm install @babelqueue/sqs @aws-sdk/client-sqs

@aws-sdk/client-sqs is an optional peer — you provide the SQS client (the aggregated SQS class satisfies the transport structurally). It sends the canonical envelope as the MessageBody with the §3 MessageAttributes, and consumes by routing each message to a handler by URN.

Produce

import { SQS } from "@aws-sdk/client-sqs";
import { SqsPublisher } from "@babelqueue/sqs";

const sqs = new SQS({ region: "eu-central-1" });
const url = "https://sqs.eu-central-1.amazonaws.com/123456789012/orders";

const id = await new SqsPublisher(sqs, url).publish("urn:babel:orders:created", { order_id: 1042 });

publish(urn, data, { traceId? }) returns the message meta.id. FIFO queues: new SqsPublisher(sqs, url, { fifo: true }) (the queue URL must end in .fifo).

Consume

import { SqsConsumer } from "@babelqueue/sqs";

const consumer = new SqsConsumer(
  sqs,
  url,
  {
    "urn:babel:orders:created": async (env, message) => {
      console.log(env.data.order_id, env.trace_id);
    },
  },
  { onUnknownUrn: (env, msg) => {}, onError: (err, env, msg) => {} },
);

await consumer.poll(); // receive one batch, route, delete handled; loop this

A throwing handler’s message is released, never deleted: the consumer calls ChangeMessageVisibility with releaseDelay (default 0 s, so SQS redelivers it at once, at-least-once). Give the queue a native RedrivePolicy (maxReceiveCount → a DLQ) so a message that always fails cannot loop forever; attempts is reconciled to ApproximateReceiveCount − 1. Point the client’s endpoint at LocalStack/ElasticMQ for local testing.

A poison body (not JSON, not an envelope, no valid job or urn, or an unsupported meta.schema_version) and a version-gated message can never succeed, so the consumer settles them on their first delivery, never hands them to a handler and never retries them (ADR-0037, since @babelqueue/sqs 1.4.0).

Before decoding, the consumer version-gates on the bq-schema-version message attribute (§3.7). A missing or blank value (the empty string, or only ASCII whitespace: space, tab, LF, VT, FF, CR) changes nothing: the body is decoded as usual. Any other value is compared with "1" exactly as sent, with no trimming, so "2", "x", "01" and " 1" are all rejected without decoding the body, and onError is called. A "1" value does not skip the body check: the decoded envelope still goes through the schema_version validation, and a body that fails it is a poison body.

What happens next depends on whether you set deadLetterQueueUrl:

const consumer = new SqsConsumer(sqs, url, handlers, {
  deadLetterQueueUrl: "https://sqs.eu-central-1.amazonaws.com/123456789012/orders-dlq",
});

The original is deleted only after the send to the DLQ succeeded; a failed send leaves it undeleted, so the write is tried again when the visibility timeout lapses. Name the DLQ <queue>-dlq (SQS does not accept a dot in a queue name, so not <queue>.dlq). For a .fifo DLQ (<queue>-dlq.fifo) the consumer uses the SQS message id as the deduplication id. deadLetterQueueUrl is opt-in: with none, nothing changes and the native RedrivePolicy stays your safety net; with one set, leaving a poison or version-gated message to the RedrivePolicy is no longer a conformant settle. Only an explicitly set unknownUrn strategy is read here: an onUnknownUrn hook (and the delete it implies for an unknown URN) never applies, so a hook-only consumer behaves as fail and never drops one silently.

Azure Service Bus — @babelqueue/azure-service-bus

npm install @babelqueue/azure-service-bus @azure/service-bus

@azure/service-bus is an optional peer — you provide the sender/receiver (a ServiceBusSender / ServiceBusReceiver satisfies the adapter structurally). It sends the canonical envelope as the message body with the native §4 projection (subject = URN, correlationId = trace_id, messageId = meta.id, plus the bq- application properties), and consumes by routing each message to a handler by URN.

import { ServiceBusClient } from "@azure/service-bus";
import { AsbPublisher, AsbConsumer } from "@babelqueue/azure-service-bus";

const client = new ServiceBusClient(connectionString); // or (namespace, credential)

// produce
const id = await new AsbPublisher(client.createSender("orders"))
  .publish("urn:babel:orders:created", { order_id: 1042 });

// consume (PeekLock)
const consumer = new AsbConsumer(
  client.createReceiver("orders"),
  {
    "urn:babel:orders:created": async (env, message) => {
      console.log(env.data.order_id, env.trace_id);
    },
  },
  { onError: (err) => console.error(err) },
);
await consumer.run();

A poison body (not JSON, not an envelope, no valid job or urn, or an unsupported meta.schema_version) and a version-gated message can never succeed, so the consumer settles them on their first delivery, never hands them to a handler and never retries them (ADR-0037, since @babelqueue/azure-service-bus 1.3.0).

Before decoding, the consumer version-gates on the bq-schema-version application property (§4.7). A string "1" or the number or bigint 1 passes; any other type is rejected. The spec rejects a float 1.0, but the AMQP client hands JavaScript a 1.0 as the number 1, so this consumer cannot tell them apart and accepts it (the one exception §4.7 allows). A missing or blank value (the empty string, or only ASCII whitespace: space, tab, LF, VT, FF, CR) changes nothing: the body is decoded as usual. A property present with a null value counts as missing too. Any other value is compared with "1" exactly as sent, with no trimming, so "2", "x", "01" and " 1" are all rejected without decoding the body, and onError is called. A "1" value does not skip the body check: the decoded envelope still goes through the schema_version validation, and a body that fails it is a poison body.

What happens next depends on whether you gave the consumer a deadLetterSender, a sender on the <queue>.dlq entity (an ordinary Service Bus queue that must exist):

const consumer = new AsbConsumer(client.createReceiver("orders"), handlers, {
  deadLetterSender: client.createSender("orders.dlq"),
  queue: "orders", // dead_letter.original_queue; defaults to the receiver's entityPath
});

The original is completed only after the send succeeded; a failed send is reported and abandons it, so the write is tried again. deadLetterSender is opt-in: with none, nothing changes and the native sub-queue stays your safety net. The <queue>.dlq copy is sent as bytes: a delivery whose body the SDK already parsed (the default receiver parses a JSON body) has no exact bytes, so it is reported and abandoned rather than completed into a lossy record; use a receiver created with skipParsingBodyAsJson: true when you configure a deadLetterSender. The version gate never reads an onUnknownUrn hook.

Fixed (wire behaviour, GR-5). AsbPublisher used to put the envelope JSON in body as a string, and @azure/service-bus JSON-encodes every body that is not a Buffer, so the data section held a quoted string literal instead of the envelope bytes the .NET, Java, Go and Python SDKs write and read. The body is now a Buffer of the envelope’s UTF-8 bytes. Upgrade the Node consumers first, then the publishers. A new publisher against an old consumer (<= 1.2.0) makes every message look like poison, so work stops. A new consumer still opens an old, quoted-string body once (a migration step, only when the quoted string holds a valid envelope; any other quoted string stays poison), on both the default receiver and skipParsingBodyAsJson: true, so messages already on the queue drain. The step may be dropped in a later major.

Delayed delivery: publish(urn, data, { delayMs: 300000 }) → native scheduledEnqueueTimeUtc. A throwing handler abandons the message (the broker redelivers, incrementing deliveryCount); attempts is reconciled to max(body.attempts, deliveryCount − 1). See the Azure Service Bus binding.

Apache Pulsar — @babelqueue/pulsar

npm install @babelqueue/pulsar pulsar-client

pulsar-client is an optional peer — you provide the producer/consumer (a Producer / Consumer satisfies the adapter structurally). It sends the canonical envelope as the message payload with the §5 property projection (bq-job = URN, bq-trace-id = trace_id, bq-message-id = meta.id, plus bq-schema-version / bq-source-lang / bq-attempts, all string→string), and consumes by routing each message to a handler by URN.

import Pulsar from "pulsar-client";
import { PulsarPublisher, PulsarConsumer } from "@babelqueue/pulsar";

const client = new Pulsar.Client({ serviceUrl: "pulsar://localhost:6650" });

// produce
const producer = await client.createProducer({ topic: "orders" });
const id = await new PulsarPublisher(producer)
  .publish("urn:babel:orders:created", { order_id: 1042 });

// consume
const sub = await client.subscribe({
  topic: "orders",
  subscription: "babelqueue",
  subscriptionType: "Shared",
});
// optional: a producer on the dead-letter topic turns dead-lettering on
const dlq = await client.createProducer({ topic: "orders.dlq" });
const consumer = new PulsarConsumer(
  sub,
  {
    "urn:babel:orders:created": async (env, message) => {
      console.log(env.data.order_id, env.trace_id);
    },
  },
  {
    client,               // the consumer builds its own retry producer from it...
    topic: "orders",      // ...on this topic (required with client: pulsar-client's Consumer has no getTopic())
    maxAttempts: 3,       // the default: 3 deliveries in total, the first included
    deadLetterProducer: dlq,
    onError: (err) => console.error(err),
  },
);
await consumer.run();

Delayed delivery: publish(urn, data, { delayMs: 300000 }) → native deliverAfter. attempts is reconciled to max(body.attempts, redeliveryCount) — the redelivery count is 0-based, so no −1.

Retry, max attempts and the DLQ

Redelivery is bounded by the SDK. Pulsar has no broker-side redelivery limit and the consumer sets no native DeadLetterPolicy (one you add yourself is a second, uncoordinated limit), so it counts the attempts itself. maxAttempts is the total number of deliveries a message gets, the first included; the default is 3 and 1 means no retry. A value below 1 (or a non-integer) throws BabelQueueError.

• Counted attempt. On every delivery the consumer takes the reconciled attempts and adds one for the current delivery. A throwing handler, or an unmapped URN without an onUnknownUrn hook, is retried while that counted attempt is below maxAttempts.
• A retry is a re-publish. The consumer re-publishes the message to its own topic with bq-attempts set to the counted value (the body’s attempts in sync, out-of-band headers carried) and then acknowledges the original, but only after the send succeeded. A failed re-publish or producer creation warns through onWarn and negatively acknowledges the original, so nothing is lost. A negative acknowledgement advances getRedeliveryCount() only on Shared and Key_Shared subscriptions, so the re-publish is what makes the cap hold on every subscription type.
• Giving the consumer a producer. The consumer builds the retry producer from client (the pulsar-client Client) only when you also pass topic, the topic the subscription reads: it creates the producer on that topic, lazily on the first retry. topic must be given because the pulsar-client Consumer cannot report its own topic (it has no getTopic()). retryProducer supplies your own producer instead and wins over client. With client alone, or with neither, a retry would be a negative acknowledgement, which only terminates on Shared and Key_Shared; so on Exclusive, Failover and an undeclared subscriptionType (which counts as Exclusive, the fail-closed choice) the constructor throws BabelQueueError, asking for topic. The way out is either client plus topic (or a retryProducer), or declaring subscriptionType: "Shared" (or "Key_Shared") when the subscription really is one.
• At the cap. With a deadLetterProducer (a producer on <queue>.dlq) the message is written there as the canonical envelope plus the dead_letter block (reason failed or unknown_urn, attempts and dead_letter.attempts = the counted value), keeping the bq-job route the delivery arrived on. Without one it is acknowledged and dropped, and the drop is reported through onWarn. The original is acknowledged only after the DLQ write succeeded; a failed write leaves it negatively acknowledged, so it is redelivered and the write is tried again.
• Poison. A body that does not decode into an envelope (not JSON, not an object, an object with no valid job / urn, or any other non-conformant envelope) never reaches a handler or onUnknownUrn. With a deadLetterProducer it is written to <queue>.dlq at once, without retry, as a poison record (reason: "poison", the original bytes kept; invalid UTF-8 as base64 under raw with raw_encoding). Without one it takes the bounded retry and is then dropped.
• onUnknownUrn is unchanged. When the hook is set it runs and the message is acknowledged.

Behaviour change. Before this release a throwing handler was negatively acknowledged without a limit. With the defaults, a message that keeps failing is now dropped (or dead-lettered) at its third delivery; raise maxAttempts to keep retrying longer. A consumer on an Exclusive, Failover or undeclared subscription that is given neither a retryProducer nor client together with topic now throws at construction; pass client and topic, or declare subscriptionType: "Shared" / "Key_Shared".

Version gate. A message whose bq-schema-version property is present, not blank and not "1" is rejected without decoding the body and settled on its first delivery, outside maxAttempts (ADR-0037, since @babelqueue/pulsar 1.4.0): onError is called with the version error, no handler and no onUnknownUrn hook runs, and it is never retried or re-published. With a deadLetterProducer the raw body, byte for byte, with no dead_letter block, is written to <queue>.dlq and the message is acknowledged after the write. Without one it is acknowledged and dropped, and the drop is reported through onWarn. A failed write leaves it negatively acknowledged, so the write is tried again on the redelivery.

Behaviour change (ADR-0037, since @babelqueue/pulsar 1.4.0). A version-gated message used to be negatively acknowledged on every delivery, without limit.

See the Apache Pulsar binding.

Apache Kafka — @babelqueue/kafka

npm install @babelqueue/kafka kafkajs

kafkajs is an optional peer — you provide the producer/consumer. Kafka has no native ack/delay/DLQ/delivery-counter, so the adapter absorbs all four: the record value is the envelope, the contract fields go to bq- headers (bq-job routes), the record timestamp mirrors meta.created_at, and bq-attempts is the authoritative attempt counter. Consume is process-then-commit (manual commit).

import { Kafka } from "kafkajs";
import { KafkaPublisher, KafkaConsumer, RetryTopics } from "@babelqueue/kafka";

const kafka = new Kafka({ brokers: ["localhost:9092"] });

// produce
const producer = kafka.producer();
await producer.connect();
const id = await KafkaPublisher.create(producer, "orders")
  .publish("urn:babel:orders:created", { order_id: 1042 });

// consume (manual commit)
const consumer = kafka.consumer({ groupId: "orders-workers" });
await consumer.connect();
await consumer.subscribe({ topic: "orders" });

const retry = new RetryTopics("orders", [5_000, 60_000]); // orders.retry.1/.2 + orders.dlq
const babel = new KafkaConsumer(
  consumer,
  { "urn:babel:orders:created": async (env, message) => { console.log(env.data.order_id, env.trace_id); } },
  { producer, retryTopics: retry, maxTries: 3, onError: (err) => console.error(err) },
);
await babel.run();

Before decoding, the consumer version-gates on the bq-schema-version record header (§6.7). A missing or blank value (the empty string, or only ASCII whitespace: space, tab, LF, VT, FF, CR) changes nothing: the body is decoded as usual. A header present with a null value counts as missing too (it is skipped, and dropped from the raw record forwarded to <topic>.dlq). Any other value is compared with "1" exactly as sent, with no trimming, so "2", "x", "01" and " 1" are all rejected without decoding the body: onError is called, the raw record is written to <topic>.dlq, and only then is the offset committed. With no DLQ configured the record is committed and dropped. A "1" value does not skip the body check: the decoded envelope still goes through the schema_version validation.

An undecodable (poison) body is never retried and never reaches a handler. It is dead-lettered wrapped, like the Go reference (§6.5): a body that is exactly one JSON object keeps its keys, anything else becomes {"raw": "<body>"}, and a dead_letter block with reason: "poison", error, exception, original_queue, attempts (from bq-attempts) and lang: "node" replaces any existing one. The key is kept, the record carries bq-attempts, bq-original-topic and the bq-* headers projected from the body, and the offset is committed once the DLQ write is confirmed. The schema-version gate rejection above still forwards its record raw. A body holding a number a JS double cannot carry (an integer beyond 2^53, 1e400) is carried verbatim under raw instead of being altered.

A body is poison when, and only when, the Go reference’s typed decode would reject it: invalid JSON, a BOM, trailing data, an empty body or a non-object root; data, meta or dead_letter not an object; job, trace_id or a string member of meta / dead_letter not a string; meta.schema_version, meta.created_at, attempts or dead_letter.failed_at|attempts not an int64 integer literal ("2026", 1.5, 1e2 and true are poison); or a non-finite number inside data. A dedicated scanner decides this, because JSON.parse would lose duplicate keys. Member names are matched exactly as spelled (JOB or Meta is an unknown member, kept as it is) and every duplicate key is checked. A bare null body is not poison (an empty envelope on the unknown-URN path), and a well-typed but non-conformant envelope (a missing data, an empty trace_id, a schema_version other than 1) is routed by its URN like any other record. No decode failure can wedge a partition. On a poison record dead_letter.attempts is the bq-attempts header (Go writes 0) and original_queue is the bq-original-topic header when present; a float literal such as 1.0 is re-formatted where the body is re-encoded.

A throwing handler republishes the envelope to the next <topic>.retry.<n> tier with bq-attempts + 1, then commits; at maxTries it goes to <topic>.dlq with a dead_letter block. A delay with no retry topics throws (Kafka has no native delay). The retry republish and the envelope <topic>.dlq write carry the consumed record’s key unchanged (the raw DLQ record already did): a null key stays null, an empty key stays empty, and a key is never invented. Partition equality across SDKs is not guaranteed. See the Apache Kafka binding.

onError never blocks settlement. On every failure path the consumer settles first (the retry republish or the <topic>.dlq write, then the commit) and calls onError afterwards, even when settling itself failed. A callback that throws, or returns a promise that rejects, is swallowed, so it cannot skip the DLQ write or the commit and the consumer keeps running.

To replay <topic>.dlq, pass a KafkaRedriveIO to redrive() from @babelqueue/core. You supply the pop (a bounded fetch with manual commit whose ack commits the offset) and the record’s key; the adapter carries that key byte for byte on the write back to the work topic and on every restore write back to the DLQ (a null key stays null, an empty key stays empty):

import { redrive } from "@babelqueue/core";
import { KafkaRedriveIO } from "@babelqueue/kafka";

const io = new KafkaRedriveIO({ producer, consumer, pop: (queue) => fetchNextDlqRecord(queue) });
await redrive(io, "orders.dlq");

A restore writes the consumed record back verbatim: its original value bytes, headers and key, with a fresh timestamp, so the restored copy keeps its provenance (bq-original-topic, …) and does not age out of the DLQ early. For that, have pop return the record’s value and headers along with its body and key, and set topic: a write is treated as a restore when its queue is the record’s own topic and its body is unchanged. A record without value and headers still restores, with the body as the value and the bq-* headers projected from it. The write back to the work topic is unchanged (bq-* headers projected from the body, meta.created_at as the timestamp). A header the consumer decoded as null is not re-written on a restore.

Kafka commits are cumulative, so a run that stops on a failed message must not leave the group past it. Pass the same kafkajs consumer (anything with seek) and have pop return each record’s topic, partition and offset: the IO then offers the core’s optional RedriveIO.rewind, which seeks every affected partition back to its first unacknowledged record (redrive pops the whole batch before processing it). A record whose restore failed is never acked, and if the rewind fails too both errors are thrown together (AggregateError). A record that was acknowledged is skipped by the rewind. Without consumer nothing is rewound. pop returns a tracked copy and does not mutate the message you give it. Rewinding needs the @babelqueue/core release that adds RedriveIO.rewind; older cores never call it.

KafkaRedriveIO does not offer publishWithHeaders, so bypass has no effect on this transport.

Apache ActiveMQ Artemis — @babelqueue/artemis

npm install @babelqueue/artemis rhea

rhea is an optional peer — you provide the sender/receiver. Artemis speaks AMQP 1.0 (not RabbitMQ’s 0-9-1), with native settlement, scheduled delivery, a delivery counter and a dead-letter address. The envelope is the message body; the contract fields ride the slots a JMS peer reads — x-opt-jms-type = URN (routes), correlation-id = trace_id, creation-time = meta.created_at — plus the string bq_ application properties (underscored, since JMS property names must be valid Java identifiers).

import { Container } from "rhea";
import { ArtemisPublisher, ArtemisConsumer } from "@babelqueue/artemis";

const connection = new Container().connect({ host: "localhost", port: 5672 });

// produce
const sender = connection.open_sender("orders");
const id = await ArtemisPublisher.create(sender, "orders")
  .publish("urn:babel:orders:created", { order_id: 1042 });

// consume (the consumer owns the disposition — autoaccept off)
const receiver = connection.open_receiver({ source: "orders", autoaccept: false, credit_window: 10 });
const dlqSender = connection.open_sender("orders.dlq");

const babel = new ArtemisConsumer(
  { "urn:babel:orders:created": (env, message) => { console.log(env.data.order_id, env.trace_id); } },
  { deadLetterSender: dlqSender, maxTries: 3, onError: (err) => console.error(err) },
);
babel.listen(receiver); // wires receiver.on("message") → accept / release / dead-letter

A poison body (not JSON, not an envelope, no valid job or urn, or an unsupported meta.schema_version) and a version-gated message can never succeed, so the consumer settles them on their first delivery, never hands them to a handler and never retries them (ADR-0037, since @babelqueue/artemis 1.3.0).

Before decoding, the consumer version-gates on the bq_schema_version message property (§7.7). A missing or blank value (the empty string, or only ASCII whitespace: space, tab, LF, VT, FF, CR) changes nothing: the body is decoded as usual. A property present with a null value counts as missing too. Any other value is compared with "1" exactly as sent, with no trimming, so "2", "x", "01" and " 1" are all rejected without decoding the body, and onError is called. A "1" value does not skip the body check: the decoded envelope still goes through the schema_version validation, and a body that fails it is a poison body.

dead_letter.original_queue of a poison record is the consumer’s queue option, else the source.address of the receiver given to listen(). Only an explicitly set unknownUrn strategy is read here.

Behaviour change (ADR-0037, since @babelqueue/artemis 1.3.0). A poison body used to be forwarded to the DLQ as its raw bytes, without a dead_letter block, so a DLQ reader written against that form must now handle the poison record. Under unknownUrn: "delete" a version-gated message is now dropped without a record; its raw body used to be written even then.

A successful handler accepts the message; a throwing handler releases it (the broker redelivers and bumps delivery-count); at maxTries the envelope goes to <queue>.dlq with a dead_letter block. attempts reconciles to max(body, delivery-count) — the AMQP counter is 0-based, so no −1.

The same confirm-then-accept rule covers every other dead-letter path: a handler that has run out of tries, an unknown URN with the dead-letter strategy, and a poison body. The consumer waits for the broker to confirm the <queue>.dlq write (an accepted verdict) before it accepts the message. If the write is refused, fails or times out, onError is called and the message is released after deadLetterRetryDelayMs, so it is redelivered and the handler may run again; keep handlers idempotent. With no deadLetterSender the message is accepted and dropped, as before. See the Apache ActiveMQ Artemis binding.

Whatever a Node service produces is the canonical envelope, so it is consumed natively by any other BabelQueue SDK — see the wire contract.