// Park it in the delay queue with its original routing key and headers, so the // x-death counter keeps accumulating across attempts. channel.publish(RETRY_EXCHANGE, message.fields.routingKey, message.content, { persistent: true, headers: message.properties.headers, messageId: message.properties.messageId, }); channel.ack(message); 

$ docker run … -e PAYMENTS_URL=http://payments-is-down.invalid:4001 aurora-billing node part1/billing-consumer.ts $ node src/checkout.ts 1 [checkout] no queue bound for order.us.refunded [checkout] 1/1 orders confirmed by the broker $ docker logs <consumer> | grep -E "^\[billing\] (retry|giving up)" [billing] retry 1/3 for ord-be50d472-fee9-4c2d-a88c-a92bcfd568d7 [TypeError: fetch failed] { [billing] retry 2/3 for ord-be50d472-fee9-4c2d-a88c-a92bcfd568d7 [TypeError: fetch failed] { [billing] retry 2/3 for ord-be50d472-fee9-4c2d-a88c-a92bcfd568d7 [TypeError: fetch failed] { [billing] retry 2/3 for ord-be50d472-fee9-4c2d-a88c-a92bcfd568d7 [TypeError: fetch failed] { [billing] retry 2/3 for ord-be50d472-fee9-4c2d-a88c-a92bcfd568d7 [TypeError: fetch failed] { $ docker exec aurora-rabbitmq rabbitmqctl -q list_queues name messages name messages orders.billing 0 orders.billing.retry 1 orders.billing.dead 0 

$ docker run … -e PAYMENTS_URL=http://payments-is-down.invalid:4001 aurora-billing $ node src/checkout.ts 1 [checkout] no queue bound for order.us.refunded [checkout] 1/1 orders confirmed by the broker $ docker logs <consumer> | grep -E "^\[billing\] (retry|giving up)" [billing] retry 1/3 for ord-14f839be-4815-4639-88b2-d8deb9302923 [TypeError: fetch failed] { [billing] retry 2/3 for ord-14f839be-4815-4639-88b2-d8deb9302923 [TypeError: fetch failed] { [billing] retry 3/3 for ord-14f839be-4815-4639-88b2-d8deb9302923 [TypeError: fetch failed] { [billing] giving up on ord-14f839be-4815-4639-88b2-d8deb9302923 after 3 retries [TypeError: fetch failed] { $ docker exec aurora-rabbitmq rabbitmqctl -q list_queues name messages name messages orders.billing 0 orders.billing.retry 0 orders.billing.dead 1 

{ "dependencies": { "amqplib": "^0.10.5" }, "devDependencies": { "@types/amqplib": "^0.10.6", "typescript": "^5.9.0" } } 

{ "name": "aurora-orders-queue", "private": true, "type": "module", "engines": { "node": ">=24" }, "dependencies": { "amqplib": "^2.2.0" }, "devDependencies": { "typescript": "^5.9.0", "@types/node": "^24.0.0" } } 

run 1 no confirms (fire-and-forget) 6859 msg/s confirms, awaiting each one 70 msg/s confirms, 1,000 in flight (OrderPublisher) 5728 msg/s run 2 no confirms (fire-and-forget) 7425 msg/s confirms, awaiting each one 78 msg/s confirms, 1,000 in flight (OrderPublisher) 4784 msg/s 

// src/order-publisher.ts import type { ChannelModel, ConfirmChannel, Options, RecoveringChannelModel } from 'amqplib'; import { connectWithRecovery } from './connection.ts'; import { assertTopology, ORDERS_EXCHANGE } from './topology.ts'; export type OrderPlaced = { orderId: string; sku: string; quantity: number; totalCents: number; }; /** Caps memory, and is the backpressure: callers wait once this many are unconfirmed. */ const MAX_IN_FLIGHT = 1_000; // amqplib rejects a confirm callback with this message when the broker nacks, and with // "channel closed" when the channel dies first. Only the second one is worth resending. const BROKER_NACK = 'message nacked'; type Unconfirmed = { routingKey: string; content: Buffer; options: Options.Publish & { messageId: string }; /** Written to a socket at least once, so the broker may have it. */ sent: boolean; resolve: () => void; reject: (error: Error) => void; }; export class UnroutableError extends Error { override name = 'UnroutableError'; } /** The deadline passed before the order reached any socket. It was never published. */ export class NotPublishedError extends Error { override name = 'NotPublishedError'; } /** * The deadline passed after the order was sent but before the broker confirmed it. * It stays queued and is resent until confirmed, so it must not be reported as failed. */ export class PublishPendingError extends Error { override name = 'PublishPendingError'; } export class OrderPublisher { #connection: RecoveringChannelModel | null = null; #channel: ConfirmChannel | null = null; #closed = false; #nextKey = 0; readonly #unconfirmed = new Map<number, Unconfirmed>(); readonly #returned = new Set<string>(); readonly #waitingForCapacity: Array<() => void> = []; readonly #waitingForIdle: Array<() => void> = []; static async start(): Promise<OrderPublisher> { const publisher = new OrderPublisher(); publisher.#connection = await connectWithRecovery('checkout-publisher', (model) => publisher.#attach(model), ); return publisher; } /** * Resolves once the broker has confirmed the order. Rejects if the broker refuses it, * if no queue is bound for the routing key, or when `signal` aborts first. While the * broker is unreachable it waits: the order is resent on the next connection. */ async publish(routingKey: string, order: OrderPlaced, { signal }: { signal?: AbortSignal } = {}): Promise<void> { while (this.#unconfirmed.size >= MAX_IN_FLIGHT && !this.#closed && !signal?.aborted) { await new Promise<void>((resolve) => { this.#waitingForCapacity.push(resolve); signal?.addEventListener('abort', () => resolve(), { once: true }); }); } if (this.#closed) throw new NotPublishedError('publisher is closed'); if (signal?.aborted) throw new NotPublishedError(`no capacity before the deadline for ${order.orderId}`); const { promise, resolve, reject } = Promise.withResolvers<void>(); const key = this.#nextKey++; const message: Unconfirmed = { routingKey, content: Buffer.from(JSON.stringify(order)), options: { persistent: true, mandatory: true, contentType: 'application/json', messageId: order.orderId, }, sent: false, resolve, reject, }; this.#unconfirmed.set(key, message); // No channel means we are between connections. The order waits in #unconfirmed // and #attach sends it as soon as there is a channel again. if (this.#channel) this.#send(this.#channel, key, message); signal?.addEventListener('abort', () => { if (!this.#unconfirmed.has(key)) return; if (!message.sent) { // Never left this process, so it is safe to drop and safe to report as failed. this.#settle(key); reject(new NotPublishedError(`${order.orderId} was not published before the deadline`)); } else { // The broker may already have it. Keep resending; the caller only stops waiting. reject(new PublishPendingError(`${order.orderId} is sent but not yet confirmed`)); } }, { once: true }); return promise; } /** Waits for outstanding confirms (up to a deadline), then closes the connection. */ async close(timeoutMs = 10_000): Promise<void> { this.#closed = true; for (const wake of this.#waitingForCapacity.splice(0)) wake(); if (this.#unconfirmed.size > 0) { const idle = new Promise<void>((resolve) => this.#waitingForIdle.push(resolve)); await Promise.race([idle, new Promise((resolve) => setTimeout(resolve, timeoutMs))]); } await this.#connection?.close(); for (const message of this.#unconfirmed.values()) { message.reject(new PublishPendingError(`publisher closed before ${message.options.messageId} was confirmed`)); } this.#unconfirmed.clear(); } /** Runs after every (re)connect. */ async #attach(model: ChannelModel): Promise<void> { const channel = await model.createConfirmChannel(); await assertTopology(channel); // Returns belong to the channel that saw them. A return whose confirm died with the // old channel must not mark the resend, which may well be routed this time. this.#returned.clear(); // With `mandatory`, an unroutable message comes back as basic.return *before* its // confirm, and that confirm is still an ack. Note the return so the ack is read right. channel.on('return', (returned) => { const id: unknown = returned.properties.messageId; if (typeof id === 'string') this.#returned.add(id); }); // A channel error leaves the connection up, so recovery never fires and everything // unconfirmed would wait forever. Closing the connection hands the problem to // recovery, which backs off, reconnects, and runs this method again. channel.on('error', (error: Error) => { console.error(`[checkout-publisher] channel error, rebuilding: ${error.message}`); void model.close().catch(() => undefined); }); channel.on('close', () => { if (this.#channel === channel) this.#channel = null; }); this.#channel = channel; // Whatever the previous connection left unconfirmed goes out again. Some of it may // already be on disk with only the confirm lost: that is a duplicate, and the // consumer's idempotency key is what makes a duplicate harmless. for (const [key, message] of this.#unconfirmed) this.#send(channel, key, message); } #send(channel: ConfirmChannel, key: number, message: Unconfirmed): void { try { channel.publish(ORDERS_EXCHANGE, message.routingKey, message.content, message.options, (error) => { if (error instanceof Error && error.message !== BROKER_NACK) return; // resent on reconnect this.#settle(key); const wasReturned = this.#returned.delete(message.options.messageId); if (error) message.reject(new Error(`broker refused ${message.options.messageId}`)); else if (wasReturned) message.reject(new UnroutableError(`no queue bound for ${message.routingKey}`)); else message.resolve(); }); message.sent = true; } catch { // The channel closed under us. The message stays in #unconfirmed for the next one. } } #settle(key: number): void { this.#unconfirmed.delete(key); this.#waitingForCapacity.shift()?.(); if (this.#unconfirmed.size === 0) { for (const wake of this.#waitingForIdle.splice(0)) wake(); } } }