Claude Code with RabbitMQ: AMQP Without Stuck Queues

Why RabbitMQ without CLAUDE.md ships consumers that block the broker

RabbitMQ is the most widely deployed AMQP broker and the easiest message system to misuse. The defaults of amqplib and most client libraries are "throughput first": no publisher confirms, auto-ack on consume, unlimited prefetch, in-memory queues, no dead letter routing. Each of those defaults is a way to lose messages or stall the broker.

Asked to "wire up a RabbitMQ producer and consumer in Node," Claude generates code that creates a fresh connection for every publish, declares a queue without durability, consumes with noAck: true, and has no prefetch limit. The first time the consumer crashes mid-handler, every unacknowledged message is gone. The first time the broker restarts, every queue and every message in it is gone. The first time a consumer is slower than the producer, the broker buffers messages in memory until it runs out and pages start failing.

This guide covers the CLAUDE.md configuration that locks Claude Code into RabbitMQ patterns that survive: one long-lived connection per process with a channel pool, durable exchanges and queues with explicit declarations, publisher confirms on every publish, manual acks with a prefetch limit, dead letter exchanges for poison messages, and shutdown handlers that drain in-flight work. For the queue-style alternative when AMQP topology is overkill, Claude Code with BullMQ covers job queues on Redis. For the streaming alternative when ordered, replayable event logs are the right shape, Claude Code with Kafka covers the event-streaming model.

The RabbitMQ CLAUDE.md template

The CLAUDE.md at your project root is read at the start of every Claude Code session. For a RabbitMQ integration it needs to declare: the client library, the connection topology, the exchange and queue conventions, the publisher confirm policy, the consumer ack and prefetch rules, the dead letter setup, and the hard rules that block the message-loss patterns Claude generates by default.

# RabbitMQ rules

## Stack
- amqplib ^0.10.x (Node.js, callback API) or amqp-connection-manager ^4.x (auto-reconnect wrapper)
- Server: RabbitMQ 3.13+ or compatible (CloudAMQP, AmazonMQ for RabbitMQ)
- AMQP_URL env var pins the connection (amqps:// in production)
- VHOST prefix per environment (e.g., /prod, /staging)

## Project structure
- src/rabbit/connection.ts     , singleton connection + channel factory (long-lived)
- src/rabbit/topology.ts        , exchanges + queues + bindings declarations
- src/producers/                , one file per producer (typed, publisher-confirms enabled)
- src/consumers/                , one file per consumer (manual ack, prefetch set)
- src/handlers/                 , pure functions called by consumers (testable)
- tests/                        , unit tests for handlers + integration tests with real broker

## Connection pattern (MANDATORY)
- ONE connection per process, NEVER one per publish or consume
- Use amqp-connection-manager for automatic reconnect with backoff
- Connections are TCP-expensive; channels are cheap, open one per logical concern
- Heartbeat: 60 seconds in production
- Connection name: set via clientProperties.connection_name = '<service>-<env>-<pid>' for broker debugging

## Channel pattern (MANDATORY)
- Producer channel: dedicated, with publisher confirms enabled
- Consumer channel: dedicated per consumer, with prefetch set
- NEVER share a channel across producer and consumer
- NEVER share a channel across consumer groups

## Exchange and queue declarations (MANDATORY)
- Declared explicitly in src/rabbit/topology.ts, idempotent on startup
- Exchanges: durable: true for production traffic
- Queues: durable: true, with dead-letter routing configured at declaration time
- Bindings: declared alongside the queue, not in the producer or consumer
- NEVER use the default exchange '' for routed work
- NEVER auto-delete production queues

## Publisher confirms (MANDATORY for every publisher)
- channel.confirmSelect() called at channel creation
- Every publish awaits the confirm callback (or use confirmChannel.publish + waitForConfirms)
- On confirm failure: retry with backoff, log, eventually DLQ at the application level

## Consumer ack and prefetch (MANDATORY for every consumer)
- channel.prefetch(<N>) set before consume, default 10
- noAck: false on consume call (manual ack required)
- ack on success, nack with requeue: false on poison message (routes to DLX)
- nack with requeue: true on transient error (broker re-delivers)

## Dead letter handling (MANDATORY)
- Every production queue declares x-dead-letter-exchange + x-dead-letter-routing-key
- The DLX routes to a DLQ for human investigation
- DLQ messages carry headers: original-exchange, original-routing-key, failure-reason

## Graceful shutdown (MANDATORY)
- Consumer cancel() on SIGTERM, wait for in-flight handlers
- Close producer channel after pending confirms drain
- Close connection last
- Shutdown timeout matches longest handler duration plus a buffer

## Hard rules
- NEVER create a new connection per publish or consume
- NEVER use noAck: true in production
- NEVER omit prefetch on a consumer (default is unlimited, broker memory exhaustion)
- NEVER declare a queue with durable: false in production
- NEVER publish to a fanout/topic exchange without publisher confirms
- NEVER use the same channel for producer and consumer
- ALWAYS log connection_name + queue + delivery_tag on every consumer event

Six rules here prevent the majority of production incidents Claude generates without them.

The one-connection rule is the foundation. Opening a new AMQP connection per publish or per consumer is the most common Claude pattern, and it is also the fastest way to exhaust the broker's file descriptor limit. Each connection consumes one TCP socket, one heartbeat thread on the broker, and at least one Erlang process. A web tier with 100 instances each opening a connection per HTTP request will hit the broker's connection limit within minutes. The right pattern is one long-lived connection per process, with channels (which are cheap virtual streams over the connection) for each producer and consumer.

The manual ack rule is the most important consumer rule. With noAck: true, RabbitMQ removes the message from the queue the moment it is delivered to the consumer. If the consumer crashes after receiving the message but before processing it, the message is gone. With noAck: false, the message stays in the queue until the consumer sends ack. A crash before ack causes the broker to redeliver the message to another consumer.

The prefetch rule is the consumer back-pressure mechanism. Without prefetch, the broker pushes messages to the consumer as fast as the TCP socket allows. The consumer buffers them in memory. A slow handler with a fast producer means the consumer's memory grows without bound, then crashes, and the broker redelivers everything. With prefetch(10), the consumer holds at most 10 unacked messages at a time. The broker holds the rest in the queue, where they belong.

The durable queue rule is the broker-restart insurance. A non-durable queue and the messages in it are lost when RabbitMQ restarts. In development this is fine. In production it means a planned upgrade or an unexpected crash empties every queue. Durable queues survive a broker restart, and messages published with persistent: true survive too.

The publisher confirms rule is the produce-success insurance. Without confirms, a publish returns as soon as the message is handed to the broker's TCP buffer. If the broker dies before fsync, the message is gone and the publisher does not know. With confirms, the broker sends back an ack when the message has been routed and persisted. The publisher knows to retry on nack or a timeout.

Install and connection setup

Install amqp-connection-manager for the auto-reconnect wrapper, plus amqplib underneath:

pnpm add amqp-connection-manager amqplib
pnpm add -D @types/amqplib

Set the environment:

# .env.local
AMQP_URL=amqp://guest:guest@localhost:5672/local

Create the singleton connection module:

// src/rabbit/connection.ts
import amqp, { AmqpConnectionManager, ChannelWrapper } from 'amqp-connection-manager';

if (!process.env.AMQP_URL) {
  throw new Error('AMQP_URL must be set');
}

const connectionName = `${process.env.SERVICE_NAME ?? 'unknown'}-${process.env.NODE_ENV ?? 'dev'}-${process.pid}`;

export const connection: AmqpConnectionManager = amqp.connect([process.env.AMQP_URL], {
  heartbeatIntervalInSeconds: 60,
  reconnectTimeInSeconds: 5,
  connectionOptions: {
    clientProperties: {
      connection_name: connectionName,
    },
  },
});

connection.on('connect', () => {
  console.log('rabbit_connected', { connectionName });
});

connection.on('disconnect', ({ err }) => {
  console.warn('rabbit_disconnected', { connectionName, message: err.message });
});

The connection_name field shows up in the RabbitMQ management UI. When debugging which application opened a leaking connection, this is the first thing to look for. Including the PID makes it unique per process even on the same host.

amqp-connection-manager handles automatic reconnection with backoff. If the broker restarts, the manager reconnects, re-declares topology (via the setup function we will use shortly), and re-attaches consumers. Without the manager, an amqplib connection that drops requires application-level reconnection logic that is almost always wrong.

Declaring topology

Exchanges, queues, and bindings are declared once at startup. Putting the declarations in a single topology.ts file makes the message routing visible at a glance, and lets the connection manager redeclare on reconnect.

// src/rabbit/topology.ts
import { Channel } from 'amqplib';

export const EXCHANGES = {
  ORDERS: 'orders.topic',
  DLX_ORDERS: 'orders.dlx',
} as const;

export const QUEUES = {
  CHARGE_PROCESSOR: 'charge-processor',
  CHARGE_PROCESSOR_DLQ: 'charge-processor.dlq',
  INVENTORY_RESERVATION: 'inventory-reservation',
  INVENTORY_RESERVATION_DLQ: 'inventory-reservation.dlq',
} as const;

export async function declareTopology(channel: Channel) {
  await channel.assertExchange(EXCHANGES.ORDERS, 'topic', { durable: true });
  await channel.assertExchange(EXCHANGES.DLX_ORDERS, 'topic', { durable: true });

  await channel.assertQueue(QUEUES.CHARGE_PROCESSOR_DLQ, {
    durable: true,
  });
  await channel.bindQueue(QUEUES.CHARGE_PROCESSOR_DLQ, EXCHANGES.DLX_ORDERS, 'charge-processor.#');

  await channel.assertQueue(QUEUES.CHARGE_PROCESSOR, {
    durable: true,
    deadLetterExchange: EXCHANGES.DLX_ORDERS,
    deadLetterRoutingKey: 'charge-processor.failed',
    messageTtl: 1000 * 60 * 60 * 24,
    maxLength: 1_000_000,
  });
  await channel.bindQueue(QUEUES.CHARGE_PROCESSOR, EXCHANGES.ORDERS, 'order.placed');

  await channel.assertQueue(QUEUES.INVENTORY_RESERVATION_DLQ, { durable: true });
  await channel.bindQueue(
    QUEUES.INVENTORY_RESERVATION_DLQ,
    EXCHANGES.DLX_ORDERS,
    'inventory-reservation.#',
  );

  await channel.assertQueue(QUEUES.INVENTORY_RESERVATION, {
    durable: true,
    deadLetterExchange: EXCHANGES.DLX_ORDERS,
    deadLetterRoutingKey: 'inventory-reservation.failed',
  });
  await channel.bindQueue(QUEUES.INVENTORY_RESERVATION, EXCHANGES.ORDERS, 'order.placed');
}

Four design decisions here. The topic exchange orders.topic routes by routing key, so the same order.placed message lands in every interested queue. The dead-letter exchange orders.dlx is the destination for messages that fail processing, routed to per-consumer DLQs. The deadLetterExchange argument on the primary queue declarations wires up the DLX automatically. The messageTtl and maxLength arguments bound the queue size; messages older than 24 hours or beyond the millionth queued message route to the DLX instead of accumulating.

Producing with publisher confirms

The producer opens a confirm channel and awaits acknowledgement on every publish.

// src/producers/order-placed.ts
import { connection } from '../rabbit/connection';
import { EXCHANGES, declareTopology } from '../rabbit/topology';

const channel = connection.createChannel({
  json: true,
  confirm: true,
  setup: async (ch: any) => {
    await declareTopology(ch);
  },
});

export type OrderPlacedEvent = {
  orderId: string;
  userId: string;
  amount: number;
  currency: string;
  placedAt: number;
};

export async function publishOrderPlaced(event: OrderPlacedEvent) {
  await channel.publish(EXCHANGES.ORDERS, 'order.placed', event, {
    persistent: true,
    messageId: event.orderId,
    headers: {
      'event-type': 'order.placed.v1',
    },
  });
}

Three CLAUDE.md rules visible here. The channel is created with confirm: true, which enables publisher confirms. The setup function declares topology on this channel and is re-run automatically on reconnect. The publish uses persistent: true so the broker fsyncs the message to disk; without this, even a durable queue does not save messages across a broker restart.

The messageId field is set to the business identifier. Consumers can use this for idempotency without parsing the payload. Setting headers explicitly makes the routing intent visible in monitoring tools.

The channel.publish() call from amqp-connection-manager returns a promise that resolves only when the broker has confirmed the message. Internally this is the publisher confirm protocol: send the publish, wait for basic.ack or basic.nack. If the broker nacks, the promise rejects and the caller can retry.

Consuming with manual ack and prefetch

The consumer sets prefetch before the consume call and acks each message after the handler returns.

// src/consumers/charge-processor.ts
import { connection } from '../rabbit/connection';
import { QUEUES, declareTopology } from '../rabbit/topology';
import { processCharge } from '../handlers/process-charge';

const channel = connection.createChannel({
  json: true,
  setup: async (ch: any) => {
    await declareTopology(ch);
    await ch.prefetch(10);
    await ch.consume(QUEUES.CHARGE_PROCESSOR, async (msg: any) => {
      if (!msg) return;

      const deliveryTag = msg.fields.deliveryTag;
      const event = JSON.parse(msg.content.toString());

      try {
        await processCharge(event, {
          messageId: msg.properties.messageId,
        });
        ch.ack(msg);
      } catch (err) {
        if (isNonRetryable(err)) {
          console.error('charge_processor_poison', {
            deliveryTag,
            messageId: msg.properties.messageId,
            error: (err as Error).message,
          });
          ch.nack(msg, false, false);
        } else {
          console.warn('charge_processor_retry', {
            deliveryTag,
            messageId: msg.properties.messageId,
            error: (err as Error).message,
          });
          ch.nack(msg, false, true);
        }
      }
    });
  },
});

function isNonRetryable(err: unknown): boolean {
  if (!(err instanceof Error)) return false;
  return err.name === 'SchemaError' || err.name === 'NotFoundError';
}

The prefetch(10) cap means the broker delivers at most 10 messages before requiring acks. The consume call runs the handler for each delivered message. On success, ack(msg) releases the message from the broker. On poison (non-retryable) failure, nack(msg, false, false) rejects without requeue, which routes the message to the DLX. On transient failure, nack(msg, false, true) rejects with requeue, which puts the message back on the queue for redelivery.

The classification of error types is the application's responsibility. Schema errors, references to deleted entities, and validation failures are non-retryable. Network timeouts, upstream 503s, and transient broker errors are retryable. Getting the classification wrong in either direction is bad: classifying a retryable error as poison loses recoverable work, and classifying a poison message as retryable creates an infinite redelivery loop.

The handler itself sits in src/handlers/process-charge.ts and is the same idempotency-bounded pure function pattern that Claude Code with Kafka describes for stream consumers. The handler dedupes by messageId against a Redis SET with TTL, so a redelivery (after a nack with requeue, or after a broker crash) does not double-process.

Connection pooling for high-throughput producers

A single producer channel can sustain a few thousand publishes per second. Beyond that, the publisher-confirm round trip becomes the bottleneck. Solution: open a pool of producer channels and round-robin across them.

// src/rabbit/producer-pool.ts
import { connection } from './connection';
import { EXCHANGES, declareTopology } from './topology';

const POOL_SIZE = 4;

const channels = Array.from({ length: POOL_SIZE }, () =>
  connection.createChannel({
    json: true,
    confirm: true,
    setup: async (ch: any) => {
      await declareTopology(ch);
    },
  }),
);

let cursor = 0;

export function getChannel() {
  const ch = channels[cursor];
  cursor = (cursor + 1) % POOL_SIZE;
  return ch;
}

Each channel maintains its own publisher-confirm pipeline. The pool spreads load across them, giving roughly N-times the throughput of a single channel up to the per-connection limit (which is typically a few tens of thousands of publishes per second).

For consumer pooling, the model is different: each consumer has its own channel because the prefetch limit is per-channel. To increase consumer throughput, run more consumer processes or open multiple consumer channels on the same connection, each binding to the same queue. RabbitMQ round-robins delivery across consumers on a queue.

Graceful shutdown

A consumer that exits without draining loses in-flight work to redelivery. A producer that exits without waiting for confirms loses messages that were sent but not yet acked.

// src/main.ts
import { connection } from './rabbit/connection';
import { startChargeProcessor } from './consumers/charge-processor';
import { startInventoryReservation } from './consumers/inventory-reservation';

const consumers = [await startChargeProcessor(), await startInventoryReservation()];

async function shutdown(signal: string) {
  console.log('shutdown_start', { signal });

  for (const consumer of consumers) {
    await consumer.cancel();
  }

  await connection.close();

  console.log('shutdown_complete');
  process.exit(0);
}

process.on('SIGTERM', () => shutdown('SIGTERM'));
process.on('SIGINT', () => shutdown('SIGINT'));

consumer.cancel() tells the broker to stop delivering new messages to this consumer. In-flight handlers continue to run and ack as they complete. After all handlers complete, connection.close() shuts down the underlying TCP connection cleanly, after waiting for pending publisher confirms.

Match the Kubernetes terminationGracePeriodSeconds to the longest handler duration plus a buffer. If a handler can take five minutes and the grace period is 30 seconds, the orchestrator sends SIGKILL before the consumer has a chance to drain.

Dead letter handling and replay

The dead letter exchange routes failed messages to per-consumer DLQs. A dedicated DLQ consumer can either alert on every poison message (low-volume) or batch them for manual review (high-volume).

// src/consumers/charge-processor-dlq.ts
import { connection } from '../rabbit/connection';
import { QUEUES } from '../rabbit/topology';
import { reportPoisonMessage } from '../alerting/poison-alert';

connection.createChannel({
  setup: async (ch: any) => {
    await ch.prefetch(50);
    await ch.consume(QUEUES.CHARGE_PROCESSOR_DLQ, async (msg: any) => {
      if (!msg) return;
      try {
        await reportPoisonMessage({
          queue: QUEUES.CHARGE_PROCESSOR_DLQ,
          messageId: msg.properties.messageId,
          firstFailureReason: msg.properties.headers?.['x-first-death-reason'],
          payload: JSON.parse(msg.content.toString()),
        });
        ch.ack(msg);
      } catch (err) {
        ch.nack(msg, false, true);
      }
    });
  },
});

The x-first-death-reason header is set by RabbitMQ when a message is dead-lettered, and tells you whether it was rejected, expired, or hit the queue length limit. The DLQ consumer is its own service-level component; if it fails, messages accumulate in the DLQ where an operator can replay them via the management UI or a CLI tool.

Common Claude Code mistakes with RabbitMQ

Six patterns Claude generates incorrectly without CLAUDE.md constraints.

1. Connection per publish

Claude generates: const conn = await amqp.connect(url); const ch = await conn.createChannel(); ch.publish(...) inside a function called per request.

Correct pattern: import the singleton connection from src/rabbit/connection.ts, reuse the long-lived producer channel.

2. noAck: true on consumers

Claude generates: channel.consume('queue', handler, { noAck: true }).

Correct pattern: channel.consume('queue', handler, { noAck: false }) with manual ack/nack inside the handler.

3. No prefetch

Claude generates: await channel.consume(...) without await channel.prefetch(N) first.

Correct pattern: await channel.prefetch(10); await channel.consume(...).

4. Non-durable queues and non-persistent messages

Claude generates: assertQueue('queue', { durable: false }) or publish(..., { persistent: false }).

Correct pattern: assertQueue('queue', { durable: true }) and publish(..., { persistent: true }).

5. No publisher confirms

Claude generates: connection.createChannel() without confirmSelect(), then publishes without awaiting confirmation.

Correct pattern: connection.createChannel({ confirm: true }) (amqp-connection-manager) or await channel.confirmSelect() (raw amqplib).

6. Topology declared per producer

Claude generates: assertExchange and assertQueue calls inside each producer or consumer file.

Correct pattern: declare topology once in src/rabbit/topology.ts, re-run on reconnect via the setup function.

Add each pattern to CLAUDE.md with the corrected form.

Permission hooks for RabbitMQ CLI

rabbitmqctl and the RabbitMQ HTTP API have a set of operations that are dangerous in production. Queue purge, exchange delete, user delete, and policy changes can take a system down or lose data. Permission hooks gate them.

In .claude/settings.local.json:

{
  "permissions": {
    "allow": [
      "Bash(rabbitmqctl list_queues*)",
      "Bash(rabbitmqctl list_consumers*)",
      "Bash(rabbitmqctl list_connections*)",
      "Bash(rabbitmqctl status*)",
      "Bash(rabbitmqadmin list*)"
    ],
    "deny": [
      "Bash(rabbitmqctl delete_queue*)",
      "Bash(rabbitmqctl purge_queue*)",
      "Bash(rabbitmqctl delete_exchange*)",
      "Bash(rabbitmqctl delete_user*)",
      "Bash(rabbitmqctl set_policy*)",
      "Bash(rabbitmqadmin delete*)",
      "Bash(rabbitmqadmin purge*)"
    ]
  }
}

The allow list permits read-only inspection. The deny list forces destructive operations into the explicit-confirmation path. Combined with the Claude Code hooks system for project-wide policy, the result is a setup where Claude can investigate a stuck queue but cannot purge it.

Observability

Every RabbitMQ deployment needs four sets of metrics: queue depth (ready, unacked, total), connection and channel counts, publish rates and confirm latencies, and broker resource usage (memory, disk, file descriptors).

Queue depth is the most important consumer-health signal. Alert on ready > 10_000 (consumer is too slow) or unacked > prefetch * consumer_count * 2 (consumer is stuck). The unacked signal catches a consumer that has stopped processing but has not closed the channel; the broker still thinks it is alive.

Connection and channel counts catch leaks. A healthy service has a stable connection count (one per process) and a stable channel count (one per producer/consumer per process). A growing count means someone is opening connections in a request handler.

Publish rates and confirm latencies tell you whether the producer is healthy. A growing confirm latency means the broker is overwhelmed or the disk is slow. A drop in publish rate with a stable confirm rate means the upstream stopped publishing.

Broker resource metrics are cluster-health. RabbitMQ slows publishes when memory exceeds the watermark (default 40% of RAM); alert before the watermark hits. Disk free space below the disk watermark causes the broker to block publishes entirely; this is a hard failure mode. For the observability layer that surfaces these in dashboards, Claude Code with Datadog covers the integration.

Choosing RabbitMQ versus the alternatives

RabbitMQ is the right tool for routed messaging with rich topology: direct, topic, fanout, and headers exchanges with bindings that route by routing key. It is not the right tool for replayable event streams (Kafka is) or for request-response style background jobs (BullMQ is).

Use RabbitMQ when: you need flexible message routing with multiple consumer types per producer, you have moderate throughput (tens of thousands of messages per second per broker, not millions), and you need a mature, well-supported broker that runs on every cloud.

Use Kafka when: you need an append-only log of events that can be replayed by new consumers, you have very high throughput (hundreds of thousands of messages per second per broker), and you need strict per-key ordering across consumer groups.

Use BullMQ when: your workload is request/response style background jobs, you already run Redis, and you want a simpler operational model than a dedicated message broker.

The CLAUDE.md rules for each are different. Picking the right tool first is the rule that prevents the most work later.

Building RabbitMQ code that does not stall the broker

The RabbitMQ CLAUDE.md in this guide produces code where every process has exactly one long-lived connection, every producer uses publisher confirms, every queue is durable with a dead letter exchange, every consumer sets prefetch before consume, every message is acked or nacked manually, and shutdown drains in-flight work.

The underlying principle is the same as any messaging infrastructure with Claude Code. RabbitMQ without a CLAUDE.md produces code that works against a local broker and falls apart the first time the broker restarts, the consumer is slower than the producer, or a publisher retry sends a duplicate. The CLAUDE.md is what makes the production-safe defaults the default for Claude.

For the streaming alternative, Claude Code with Kafka covers replayable event logs. For the job-queue alternative on Redis, Claude Code with BullMQ covers request/response style background work. For Redis as the consumer-idempotency backing store, Claude Code with Redis covers the connection patterns.

Get Claudify. Ship RabbitMQ producers and consumers that hold up under real load.