RabbitMQ Messaging skill for Codex

A RabbitMQ integration is sound when publishers know the broker accepted and routed each message, consumers acknowledge only after the work is durable, every queue has a dead-letter path and a redelivery limit, and the topology is declared in code or definitions rather than clicked together in the UI.

Workflow

- [ ] 1. Inspect broker version, topology and client libraries
- [ ] 2. Design exchanges, routing keys and queues
- [ ] 3. Publish with confirms and mandatory routing
- [ ] 4. Consume with manual acks, prefetch and idempotency
- [ ] 5. Add dead-lettering, delayed retry and limits
- [ ] 6. Manage connections, then monitor

1. Inspect

• Read AGENTS.md, then find connection setup, topology declarations (assertExchange, queue_declare, definitions.json) and the broker version (rabbitmq-diagnostics status or the management UI).
• RabbitMQ 4.x removed classic mirrored queues. Any ha-mode policy is a migration item to quorum queues.
• List existing queues and their state if you have access:

rabbitmqctl list_queues name type messages_ready messages_unacknowledged consumers
rabbitmqctl list_exchanges name type
rabbitmqctl list_bindings source_name destination_name routing_key

These commands need access to the broker host or container; in a sandbox without it, ask the user to run them and paste the output.

2. Topology

Need	Exchange	Routing
Events many services may subscribe to	topic (e.g. orders)	order.created, order.cancelled.eu
Commands to exactly one handler type	direct	queue name or command name
Broadcast to every bound queue regardless of key	fanout	ignored

• One exchange per domain or bounded context; one queue per consumer service per purpose (billing.order-created). Never let two different services share a queue unless they are replicas of the same consumer.
• Routing keys are entity.event[.qualifier]. Bind with * (one word) or # (zero or more) deliberately; # bindings silently receive every future event.

Queue type	Use for
Quorum (x-queue-type: quorum)	Default for durable work queues; replicated, poison-message limit built in
Stream (x-queue-type: stream)	Replay, large fan-out, many consumers reading the same history
Classic, non-durable or exclusive	Temporary reply queues, transient per-connection data

Queue type and other arguments are fixed at declaration; redeclaring with different arguments fails with PRECONDITION_FAILED. Changing them means a new queue and a migration. Tunables that can change (DLX, delivery limit, TTL, max length) belong in policies.

3. Publish reliably

import amqp from 'amqplib';

const conn = await amqp.connect(process.env.AMQP_URL!);
const ch = await conn.createConfirmChannel();
await ch.assertExchange('orders', 'topic', { durable: true });

ch.on('return', (msg) => {
  log.error({ routingKey: msg.fields.routingKey }, 'unroutable message returned');
});

export async function publishOrderCreated(order: Order) {
  ch.publish('orders', 'order.created', Buffer.from(JSON.stringify({ orderId: order.id })), {
    persistent: true,
    mandatory: true,                // return it if no queue is bound
    messageId: `order.created:${order.id}`,
    contentType: 'application/json',
    timestamp: Math.floor(Date.now() / 1000),
    headers: { 'schema-version': '1' },
  });
  await ch.waitForConfirms();       // broker has taken responsibility
}

• Publisher confirms are the only way to know the broker accepted a message. Batch publishes and wait once per batch for throughput.
• mandatory: true plus a return handler (or an alternate exchange) catches messages that matched no binding; otherwise they are silently dropped.
• Publishing after a database commit can still lose the message if the process dies in between; write to an outbox in the same transaction and relay (see outbox-pattern).
• Set a stable messageId so consumers can deduplicate.

4. Consume safely

const ch = await conn.createChannel();
await ch.assertQueue('billing.order-created', {
  durable: true,
  arguments: { 'x-queue-type': 'quorum' },
});
await ch.bindQueue('billing.order-created', 'orders', 'order.created');
await ch.prefetch(20);

await ch.consume('billing.order-created', async (msg) => {
  if (!msg) return;                                   // consumer cancelled by broker
  try {
    const body = JSON.parse(msg.content.toString());
    await handleOnce(msg.properties.messageId, () => createInvoice(body.orderId));
    ch.ack(msg);
  } catch (err) {
    if (isTransient(err)) return scheduleRetry(ch, msg);   // see step 5
    ch.nack(msg, false, false);                            // to the dead-letter exchange
  }
}, { noAck: false });

• Manual acks only. Ack after the side effect is committed, never before.
• Prefetch bounds unacked messages per consumer: 10-50 for typical I/O-bound handlers, 1-5 for slow or heavy ones. Unlimited prefetch makes one consumer hoard the queue.
• Delivery is at-least-once: dedupe on messageId in the same transaction as the effect (an inbox table with a unique key).
• Ordering holds only per queue with one consumer and no requeues. If order matters, use x-single-active-consumer: true or a version check in the handler.

5. Dead letters, retries, poison messages

rabbitmqctl set_policy --apply-to queues billing-dlx "^billing\." \
  '{"dead-letter-exchange":"billing.dlx","delivery-limit":5}'

• Every work queue gets a dead-letter exchange and a bound *.dead queue that someone monitors. Quorum queues in 4.x default to a delivery limit of 20; set it explicitly.
• Never nack with requeue=true on errors that will repeat; that is a hot loop.
• Delayed retry pattern: on transient failure, republish to a retry exchange with a per-delay wait queue (x-message-ttl 5 s, 30 s, 5 min) whose dead-letter exchange routes back to the work exchange; increment your own x-retry-count header and ack the original. After N retries, publish to the dead queue instead.
• For quorum queues that must not lose dead-lettered messages, use dead-letter-strategy: at-least-once, which requires overflow: reject-publish.
• Cap queue growth with max-length or max-length-bytes policies plus alerts; unbounded queues eventually trigger memory or disk alarms that block all publishers.

6. Connections and monitoring

• One long-lived connection per process (or one for publishing and one for consuming); channels per thread or task. Never open a connection per message.
• Channels are not thread-safe in most clients; do not share one across threads.
• Handle reconnects: Java and .NET clients and aio-pika’s connect_robust recover automatically; with amqplib or pika you must re-create connection, channels and consumers on close/error.
• Heartbeats on (default 60 s negotiation) so dead TCP connections are detected.
• Monitor via the Prometheus plugin (port 15692) or management API: messages ready, unacked, consumer count per queue, publish vs deliver rates, memory and disk alarms, dead-letter queue depth. Alert on “ready messages growing with zero consumers” and on any dead queue depth above zero.

Gotchas

• Declaring a queue from both producer and consumer with different arguments causes PRECONDITION_FAILED on whichever starts second.
• Auto-ack (noAck: true) loses every in-flight message when a consumer crashes.
• A slow handler with a large prefetch holds hundreds of messages invisible to other consumers, while the queue looks idle.
• Long-running handlers exceeding the consumer acknowledgement timeout (30 minutes by default) get their channel closed and messages redelivered.
• Purging or deleting queues destroys messages permanently; confirm with the user before running rabbitmqctl purge_queue or deleting anything in production.
• Using RabbitMQ as a database (millions of messages parked for days) degrades the broker; store state elsewhere and keep queues short.

Verification

• Integration test against a disposable broker (docker run -p 5672:5672 rabbitmq:4-management, needs Docker and an image pull, or testcontainers): publish with confirms, consume, assert one effect.
• Publish to a routing key with no binding and confirm the return handler fires.
• Throw a permanent error in the handler: message lands in the dead queue once, no loop.
• Kill the consumer mid-message: message is redelivered and processed once in effect.
• rabbitmqctl list_queues shows the expected type, consumers and zero unacked at rest.

Output template

## Messaging change
- Exchange(s): <name, type>
- Routing keys: <keys>
- Queues: <name, type, bindings, consumers>
- Policies: <DLX, delivery-limit, max-length>
- Publisher: confirms <yes>, mandatory <yes>, outbox <yes/no>
- Consumer: prefetch <n>, dedupe on <messageId>, retry schedule <delays>
- Alerts: <metrics and thresholds>