Error Handling

Overview

Handlers are async functions that return Promise<void>. To signal failure, you throw. The framework decides what happens next based on the error type and the retry policy configured on transport.consumer — there is no return-value convention to memorise.

The problem

Distributed messaging amplifies bug blast radius. A handler that throws on a malformed payload, if left unconfigured, will loop forever: redelivery, throw, redelivery, throw. A handler that crashes on a transient network blip should retry a few times, not give up immediately. A handler that hits an unrecoverable schema problem should jump straight to a dead-letter queue so it doesn’t waste retry budget. And every category should be observable, so on-call can see “fifty messages in the error queue overnight” without grep-ing logs.

The defaults shipped with the transport handle the common case well; tuning them is a five-line job once you understand the model.

The solution

The default consumer policy is:

• maxRetries is the total number of deliveries before error-queue routing (3 by default), with retryDelay ms (3000) in between. So maxRetries=3 means the initial delivery plus 2 redeliveries — 3 attempts in total — before the message is routed to the error queue. Each retry is a fresh broker delivery, so the RetryCount AMQP header (read in a handler as ctx.headers.RetryCount) increments and the message goes back through your whole pipeline.
• After the final delivery fails, the framework publishes the message to the error queue ('errors' by default, configurable via transport.consumer.errorQueue) along with headers describing the failure (retry count, last error, and a terminal-failure flag).
• Terminal classification happens only in the deserialize step: a TerminalDeserializationError thrown by the serializer, or a Standard Schema ValidationError, routes the message directly to the error queue with zero retries. A handler that throws is always treated as a retryable failure — there is no way to short-circuit retries from inside a handler.
• Audit queue (opt-in): setting auditEnabled: true and an auditQueue mirrors every consumed message to a separate queue (default 'audit'). Off by default (auditEnabled defaults to false) because the disk cost is real.
• Unhandled message types: if no handler is registered for a message type, the consumer either acks (default) or dead-letters depending on deadLetterUnhandled.

Inside a handler, you just throw to signal failure. Every handler throw is treated the same way — as a retryable failure — regardless of the error type, so the message is redelivered until maxRetries is reached and then routed to the error queue. There is no way to skip retries from inside a handler; zero-retry routing only happens for deserialization and schema-validation failures (see below).

bus.handle<ChargeCard>('ChargeCard', async (msg, ctx) => {
  // Any throw here is retried up to maxRetries, then routed to the error queue.
  await payments.charge(msg.orderId, msg.total);
});

The transport-side knob lives on transport.consumer:

const transport = rabbitMQWithRegistry(
  {
    url: process.env.AMQP_URL!,
    consumer: {
      maxRetries: 5,
      retryDelay: 5_000,
      errorQueue: 'orders.errors',
      auditQueue: 'orders.audit',
      auditEnabled: true,
      deadLetterUnhandled: true,
    },
  },
  registry,
);

Setting errorQueue: null disables error-queue routing entirely — failures are logged and acked. This is occasionally useful in dev or for fire-and-forget telemetry pipelines, but in production it means you lose every record of a failed message. Don’t do it without a replacement observability path.

Redriving from the error queue (re-publishing a stuck message to its original queue) is intentionally outside this package — every team has their own approval/audit requirements for that workflow. Treat the error queue as an inbox and build whatever drainage tooling you need on top.

Headers on dead-lettered messages

When the framework routes a message to the error queue, it adds headers that describe the failure. The most useful ones in a postmortem are:

• RetryCount — how many attempts were made before giving up.
• TerminalFailure — set to 'true' when the failure was terminal-classified (a deserialization/validation failure routed straight to errors with zero retries).
• Exception — the last thrown error’s message followed by its stack trace (stack truncated to 2048 characters).

The framework does not add a header carrying the source queue name. If a redrive tool needs to know where a message originated, derive it another way (e.g. from broker metadata) — there is no OriginalQueue header to rely on.

These are plain AMQP headers, visible in the management UI and accessible from any consumer that subscribes to the error queue. Build alerting on top of the queue depth (messages_ready on the error queue itself) so a sudden spike pages you immediately rather than being discovered the next morning.

Audit vs error

The audit queue is a tee on the consume path: every message the consumer successfully handles is copied to the audit queue with a TimeProcessed header (an ISO timestamp) and a Success: 'true' header. Auditing is off by default (auditEnabled defaults to false); when enabled without an explicit auditQueue, the default queue name is 'audit'. It is not a retry-tracker — failed messages still go to the error queue, audit just gives you a copy of every successful processing event. Use it for compliance (“did we ever process this order?”) or replay scenarios; it is not a cheap dev tool, since it doubles your durable write volume on the broker.

Error classes the framework throws

Most of these are caught and translated into queue routing by the consumer; a few surface to your code on the outbound side. They all extend the common base ServiceConnectError, so a single catch (e: ServiceConnectError) covers the family:

• ServiceConnectError — the base class for every framework-thrown error. Use it as the catch-all when you want to distinguish “this came from the bus” from “this came from my handler”.
• ArgumentError / ArgumentOutOfRangeError — thrown at configuration time when an option is malformed (e.g. negative timeoutMs, missing endpoint on send). Always a bug in calling code; fail loud at startup.
• InvalidOperationError — thrown when the bus is asked to do something inconsistent with its lifecycle (e.g. publishing after stop, or sendToMany with an empty endpoint list).
• ValidationError — thrown when a Standard Schema validator attached via registerMessage rejects an inbound payload. The message goes straight to the error queue without a retry.
• HandlerNotRegisteredError — thrown on shutdown / diagnostic paths when a code path expects a handler for a type and none was attached. In normal operation the consumer policy (deadLetterUnhandled) decides what happens to unhandled inbound traffic; this error is for cases where the absence is a programmer error.
• MessageTypeNotRegisteredError — thrown on the outgoing side when publish / send / sendRequest references a type string that was never registerMessage’d. Always a wiring bug.
• AggregatorConfigurationError — thrown by registerAggregator when an aggregator’s options are inconsistent (e.g. no store, a batchSize() that is not a positive integer, or a timeout() that is not a positive finite number of ms).
• RoutingSlipDestinationError — thrown by bus.route(...) when the destinations list is empty or contains a malformed entry. The framework’s destination helpers (assertValidDestination, isValidDestination, destinationFailureReason) are exported for callers that need to pre-validate a slip before handing it to the bus.
• OutgoingFiltersBlockedError — thrown on the outbound path when an outgoing-pipeline filter returns FilterAction.Stop. The publish or send never reached the wire.
• RequestTimeoutError / RequestSendCancelledError / AbortError — request/reply outcomes. See Cancellation for the full failure taxonomy.
• ConcurrencyError / DuplicateSagaError — saga persistence outcomes; see Process Manager.
• StreamFaultedError / StreamSequenceError — streaming outcomes; see Streaming.
• TerminalDeserializationError — serializer-side. Thrown by a serializer during the deserialize step to mark a payload as unrecoverable; the framework routes it straight to the error queue with zero retries. It is not a handler signal — throwing it from a handler behaves like any other handler throw (a normal retryable failure).

When to set deadLetterUnhandled

If your queue can receive message types you do not have handlers for (a published event where your service is one of many subscribers, but you only care about a subset), leave deadLetterUnhandled: false (the default) — the consumer acks unknowns and moves on. If your queue is point-to-point and a message of an unknown type indicates a real bug (the sender published the wrong type to the wrong endpoint), set deadLetterUnhandled: true so those messages land somewhere you can investigate.

Common pitfalls

Caution

• Returning a string like 'retry' from a handler does nothing. Throw to retry.
• Setting maxRetries=0 does not mean “no error queue routing” — the first failure still goes to errors. To suppress error-queue routing, set errorQueue: null (and accept the loss of visibility).
• Treating errorQueue: null as production-safe. Combine with audit + alerting if you really want to go silent.
• Expecting a thrown error type to skip retries. A handler throw is always retried up to maxRetries; zero-retry routing only happens for deserialization/validation failures, not handler throws.
• Forgetting to register a handler and assuming the message will retry forever — without deadLetterUnhandled: true, the default is to ack and drop.

See also

• Idempotency
• Observability
• ConsumeResult
• Transport configuration