Consume AMQP messages

How-to. Run an amqp-contract worker under the kernel: handlers built from your own services, one unit per delivery, and a drain with exactly one deadline. For the package's full surface, see @btravstack/amqp-worker; for why neither the kernel nor the starter maps a Result to ack/nack, see The kernel maps nothing.

You have a contract with a consumer on a queue, and a service that should react to each message. What you write is the handlers record and a composition root; the connection, the unit boundary and the release at the kernel's deadline are the package's. Everything below is lifted from examples/order-amqp-worker.

Recipe

1. Implement the handlers with AmqpHandlers(contract)({ inject?, unit?, sync }) — one plain function per consumer, typed by the contract.
2. Decide, per handler, what a domain Err and a Defect become (see the three-way split below).
3. Compose with AmqpModule(name)({ contract, handlers, imports, provides, exports, needs }).
4. await runMain(OrderAmqpWorker); set AMQP_URL in the deployment.

Step 1 — the handlers, as a provider

AmqpHandlers(orderContract) is the builder on the starter's own handlers port, typed for the contract (its service the record the contract wants, WorkerInferHandlers<typeof orderContract>) — no class, no name: a consumer serves one handlers record — so the next call, { inject?, unit?, sync }, declares what the handlers need and closes over it. Nothing is injected per message — the middleware the package installs opens the unit and calls next() unchanged:

A record covers every consumer and rpc the contract declares — orderContract has two, orderNotifications and orderAudit, both reading the one orderChanged event on their own queue:

import { AmqpHandlers } from "@btravstack/amqp-worker";
import { orderContract } from "@btravstack/example-order-amqp-contract";

import { OkAsync } from "unthrown";

export const orderHandlers = AmqpHandlers(orderContract)({
  inject: { logger: Logger },
  // The envelope's own tenant, claimed ONCE by the unit module the worker
  // forks per delivery — see Step 3 — rather than destructured again in each
  // leaf.
  unit: { tenant: Tenant },
  sync: ({ logger }) => ({
    orderNotifications: ({ context, input: message }) => {
      const { id, payload } = message.payload;
      logger.info(
        payload === null
          ? "order gone — notifying"
          : "order placed — notifying",
        {
          tenantId: context.unit.tenant,
          orderId: id,
          ...(payload === null ? {} : { quantity: payload.quantity }),
        },
      );
      return OkAsync();
    },
    orderAudit: ({ context, input: message }) => {
      const { id, occurredAt, payload } = message.payload;
      logger.info("recording an order change", {
        tenantId: context.unit.tenant,
        orderId: id,
        occurredAt,
        change: payload === null ? "removed" : "placed",
      });
      return OkAsync();
    },
  }),
});

examples/order-amqp-worker composes these two consumers from a slice each now, rather than in one function — see Split a worker into slices — but the monolithic form above is unchanged, and right for a worker that has not outgrown it.

Step 2 — the three-way split

amqp-contract's dispatch settles a delivery from the handler's Result, and it is a three-way split, not two:

The handler's Result is…	What happens	Decided by
Ok	ack	amqp-contract
Err(RetryableError)	routed by the queue's retry config (e.g. ttl-backoff, maxRetries)	amqp-contract
Err(NonRetryableError)	nack, straight to the dead-letter exchange	amqp-contract
a Defect	nacked once, immediately, to the dead-letter — never touching retry	amqp-contract

The last row is the one to internalise. An unrecovered infrastructure failure is parked on its first attempt exactly like a permanent domain error. A handler that wants "infrastructure comes back" recovers its own defects into a RetryableError explicitly (an alternative to orderHandlers above, not a second provider next to it — the port is one, so one graph holds one handlers provider):

import { AmqpHandlers } from "@btravstack/amqp-worker";
import { NonRetryableError, RetryableError } from "@amqp-contract/worker";
import { TenantId } from "@btravstack/example-order-domain";
import { ErrAsync, OkAsync, P } from "unthrown";

export const placingHandlers = AmqpHandlers(orderContract)({
  inject: {},
  // `PlaceOrder` is bound to a tenant, so it comes off the fork the worker
  // opened for this delivery — see the unit module in Step 3.
  unit: { place: PlaceOrder },
  sync: () => ({
    orderNotifications: ({ context, input: message }) =>
      context.unit.place
        .execute(
          message.payload.id,
          message.payload.payload?.quantity ?? 0,
        )
        .map(() => undefined)
        .mapErrCases((matcher) =>
          matcher.with(
            P.tag("InvalidQuantity"),
            P.tag("InvalidOrderId"),
            P.tag("DuplicateOrder"),
            (error) => new NonRetryableError(error._tag, error),
          ),
        )
        .recoverDefect((cause) =>
          ErrAsync(new RetryableError("placing the order failed", cause)),
        ),
    // Not the point of this example — a bare ack keeps `placingHandlers`
    // focused on the triage `PlaceOrder` needs.
    orderAudit: () => OkAsync(),
  }),
});

The queue's policy is contract configuration the broker enforces — the example's order-notifications queue declares retry: { mode: "ttl-backoff", maxRetries: 3 } and a dead-letter exchange. Read maxRetries: 3 as four total attempts (the first plus three retries), not the same number Temporal's maximumAttempts: 3 names.

Step 3 — the composition root

import { AmqpMessage, AmqpModule } from "@btravstack/amqp-worker";
import { Module, Provider } from "@btravstack/di";
import { orderContract } from "@btravstack/example-order-amqp-contract";
import { Tenant } from "@btravstack/example-order-application";
import { TenantId } from "@btravstack/example-order-domain";
import {
  OrderDatabase,
  OrderPersistenceModule,
} from "@btravstack/example-order-infrastructure";
import { observability } from "@btravstack/observability";
import { OutboxStore, outbox } from "@btravstack/outbox";

import { orderHandlers } from "./handlers.js";

// Forked per delivery and seeded with the validated message, so the envelope's
// own `tenantId` is claimed once rather than at every handler.
export const MessageUnitModule = Module("MessageUnit")({
  provides: [
    Provider(Tenant)({
      inject: { message: AmqpMessage(orderContract) },
      sync: ({ message }) => TenantId(message.payload.tenantId),
    }),
  ],
  exports: [Tenant],
});

export const OrderAmqpWorker = AmqpModule("OrderAmqpWorker")({
  contract: orderContract,
  handlers: orderHandlers,
  // outbox() relays what the writes committed, through the publisher below —
  // see [Publish a message](/how-to/publish-a-message).
  imports: [OrderPersistenceModule, observability(), otel(), outbox()],
  provides: [orderAmqpClient, orderPublisher],
  unit: { message: MessageUnitModule },
  exports: [OutboxStore, OrderDatabase, Logger],
});

AmqpModule is Module(name)({...}) plus contract and handlers: it imports the AMQP starter for contract, provides the handlers and exports AmqpRuntime. observability() is the other starter in that list — the Logger the handlers write to, bound from LOG_LEVEL, JSON per line on stdout, every line carrying the delivery's own unit. The record is checked against the contract at AmqpHandlers(contract)(…) — a record missing a consumer, or naming one the contract does not declare, fails to typecheck there rather than on the first delivery, silently to the DLQ — and handlers is typed against the module's own contract, so a provider built for another contract is refused here.

Step 4 — main.ts

import { runMain } from "@btravstack/core";

import { OrderAmqpWorker } from "./module.js";

await runMain(OrderAmqpWorker);

Configuration and options

The starter provides AmqpRuntime and AmqpConfig ({ url, connectTimeoutMs }), and needs the handlers port AmqpModule provides. Options on AmqpModule:

Variable / option	Default	Notes
AMQP_URL / url	amqp://127.0.0.1:5672	url pins the broker (a test's container); a blank variable is a ConfigInvalid, exit 78
connectTimeoutMs	5000, from AMQP_CONNECT_TIMEOUT_MS	how long create waits before an unreachable broker is a RuntimeStartFailed, exit 1
connectionOptions	—	AmqpConnectionOptions — heartbeat, reconnect interval, findServers, TLS/socket options
defaultConsumerOptions	—	the library's ConsumerOptions — prefetch (the throughput knob — unset is 10, "unbounded" is no cap), priority, …

Once consuming, the runtime publishes AmqpInfo — { queues }, every queue the contract's consumers and RPCs drain — on Serving.info, read through runtimeInfo().

One unit per delivery

UnitMeta.id is a minted randomUUID() per delivery; traceId is the publisher's messageId, falling back to correlationId, then to the minted id. A delivery tag is not used as the id on purpose: tags are per-channel and restart at 1 after the silent reconnects amqp-connection-manager performs. An adapter reads the trace id from currentUnit().

Honouring the drain deadline

The middleware calls next() unchanged, so a handler has no parameter to receive the unit's AbortSignal through: currentUnit()?.signal is the only route to it, and it is aborted at the kernel's drainTimeoutMs.

orderNotifications: ({ input: message }) => {
  const { id, payload } = message.payload;
  if (currentUnit()?.signal.aborted === true) {
    return ErrAsync(
      new RetryableError(
        `the drain deadline passed before order ${id} was notified`,
      ),
    );
  }
  // …
};

A RetryableError spends one retry from the queue's own budget, which is the transport's answer to "this process stopped waiting". There is no cancellation to defer to here: AMQP has none, and a redelivery is recovery rather than cancellation.

It does not leave the delivery un-acked, and the difference matters on a rollout. @amqp-contract/worker acks the original and republishes a copy with x-retry-count incremented — to the queue itself in immediate-requeue mode, or to the backoff tier's wait queue under ttl-backoff. So every message in flight when a pod drains past its deadline comes back one attempt poorer, and a message that was already near maxRetries reaches the dead-letter queue because of a deploy rather than because of anything wrong with it. The one exception is a quorum queue in immediate-requeue mode, which nacks with requeue: true and genuinely leaves it un-acked.

Size maxRetries with that in mind, or answer a NonRetryableError and put the message somewhere your own code owns. What this arm buys either way is that the work is not silently dropped when the kernel stops waiting.

The drain: one deadline

Serving.drain(signal) calls worker.close({ drainTimeoutMs: null }) — cancel every consumer, let in-flight handlers finish so their acks land on a still-open channel — raced against the kernel's signal. null is deliberate: the library's own default drain timeout is 30 s, above the kernel's 20 s default, and would quietly win. Telling the library to wait forever leaves the kernel's drainTimeoutMs as the only clock in the process.

When the deadline wins, expect a SIGKILL

The kernel reports the unit abandoned (exit 2), but nothing was dropped: the connection stays open and the handler runs on toward its own ack or nack. Redelivery happens only once the connection actually drops — when the process exits, not when the drain deadline passes.

And the process does not end on its own. runMain sets an exit code and never calls process.exit(), and worker.close() closes the connection only once the deliveries it already took have drained — so the event loop stays alive. Under a Kubernetes-initiated shutdown the pod then ends at terminationGracePeriodSeconds, and exit 2 is what the report says rather than what the orchestrator observes; outside that lifecycle nothing kills it and the process waits for the deliveries.

Nothing here forces the connection shut, deliberately: destroying it under an ack in flight loses that ack, which is a worse outcome than a SIGKILL the orchestrator was already going to send. Size drainTimeoutMs so the deadline is the exception, and read the report from kubectl logs --previous.

The publishing half

A producer is not a runtime — a graph holds exactly one of those, and this process's is the consumer. The example's relay is @btravstack/outbox's outbox(), a resourceful provider that claims what committed in the outbox table and publishes it through the deployment's own OutboxPublisher, started as the graph builds and stopped when the application scope closes. That half has a page of its own: Publish a message.

See also

• @btravstack/amqp-worker — options, ports, AmqpInfo.
• Split a worker into slices — several consumers, one handler per consumer, composed at the root.
• Order AMQP worker — the outbox pattern end to end, against a real RabbitMQ.
• Manage a resource's lifetime — acquire/release, as the relay uses it.
• Configure from the environment — AMQP_URL, OUTBOX_TENANTS and OUTBOX_POLL_MS.