RabbitMQ de los fundamentos a producción, Parte 2: construir un servicio escalable
Maté el broker a mitad de una ráfaga de 20.000 pedidos. El publisher de la Parte 1 había llamado a channel.publish 10.900 veces sin un solo error. Cuando RabbitMQ volvió, la cola tenía 10.433 pedidos. Faltaban 467, y el proceso no se enteró: ni una excepción, ni una línea de log, ni una promesa rechazada. Para el checkout, a todos esos clientes se les iba a cobrar con normalidad.
La Parte 1 explicaba cómo hacer llegar un mensaje a la cola correcta. Esta parte trata de lo que lo mantiene ahí cuando algo falla: publisher confirms, conexiones que se recuperan solas, consumidores que puedes escalar y apagar sin perder trabajo y, al final, un ensayo medido que mata el broker con carga y cuenta qué sobrevivió. La topología de Aurora Coffee Co. es la misma, y src/topology.ts es el mismo.
El hueco que dejó la Parte 1
La Parte 1 terminaba reconociendo un problema. Así resolvía los reintentos en su primera versión, en src/billing-consumer.ts:
// Lo aparca en la cola de espera con su routing key y sus cabeceras originales, // para que el contador x-death siga acumulando entre intentos. channel.publish(RETRY_EXCHANGE, message.fields.routingKey, message.content, { persistent: true, headers: message.properties.headers, messageId: message.properties.messageId, }); channel.ack(message); Son dos operaciones, no una transacción. Si la copia nunca llega a la cola de reintentos, el ack ya borró el original y el pedido desaparece. Además, que channel.publish devuelva true no significa lo que parece: solo indica que el mensaje cupo en un buffer de tu propio proceso. El broker todavía no lo ha visto, y puede que nunca lo vea.
El publisher del checkout tiene exactamente el mismo punto ciego, y de ahí salen los 467 pedidos perdidos del principio. Para cerrarlo hacen falta los publisher confirms.
Y dos cosas en las que la Parte 1 se equivocó
Antes de construir sobre la Parte 1, comprobé contra RabbitMQ 4.3 qué hacía su código cuando algo fallaba. Dos de sus afirmaciones no aguantaron, y la Parte 1 ya está corregida. Esto es lo que mostraron las pruebas.
El contador de reintentos nunca contaba. La Parte 1 leía el header x-death del broker para contar los reintentos. Este es su consumidor, sin cambios, con el servicio de pagos inalcanzable y un solo pedido que procesar:
$ 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 Reintento 2 de 3, cada 30 segundos, todo el tiempo que lo dejé correr. El pedido nunca llega a orders.billing.dead. Al leer el mensaje en la cola de reintentos se ve el motivo: después de varias vueltas, su x-death seguía diciendo count: 1. Cuando un consumidor republica un mensaje, RabbitMQ 4.x no conserva el x-death que manda el cliente; solo anota la expiración que acaba de ver. La solución es llevar la cuenta en un header propio del consumidor, retry-count, que es lo que lee ahora retriesSoFar. La misma prueba con el consumidor de este artículo:
$ 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 x-delivery-limit no frena un bucle de requeue. La Parte 1 presentaba el límite de entregas como la primera defensa contra un mensaje que falla siempre. En RabbitMQ 4.3 cuenta los mensajes que vuelven a la cola porque se cerró su canal o su conexión: en mi prueba, 6 entregas y luego dead-letter. No cuenta un nack(message, false, true) explícito sobre un canal abierto: 50 requeues seguidos del mismo mensaje y ningún dead-letter. Es una red de seguridad para los consumidores que se caen, no un freno para el código que devuelve mensajes a la cola. Por eso cambia una línea del consumidor de este artículo, como verás más abajo.
Antes de nada, actualiza amqplib
Un paso de mantenimiento antes del código nuevo. La Parte 1 fijó amqplib@^0.10.5, y esa versión ya estaba atrasada cuando se publicó el artículo: amqplib 1.0 salió en marzo de 2026 y hoy la rama vigente es la 2.x. Para este artículo importan tres cambios:
- Desde la 1.2 trae sus propios tipos, así que
@types/amqplibsobra. - La 1.1 añadió recuperación de conexión opcional. Buena parte de este artículo trata de lo que hace y de lo que te deja a ti.
- La 2.0 cambió un valor por defecto:
heartbeat: 0ahora desactiva los heartbeats, en lugar de significar "usa el valor del servidor". Si nunca lo configuraste, no te afecta.
Las dependencias del package.json de la Parte 1:
{ "dependencies": { "amqplib": "^0.10.5" }, "devDependencies": { "@types/amqplib": "^0.10.6", "typescript": "^5.9.0" } } El de la Parte 2, completo:
{ "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" } } @types/node está para que tsc --noEmit reconozca process y Buffer. Compilé el topology.ts y el billing-consumer.ts de la Parte 1 contra los tipos que trae amqplib 2.2, y los dos pasan sin tocar nada.
Publisher confirms: la mitad del contrato que le toca al broker
El consumidor le confirma al broker que procesó un mensaje. Un publisher confirm es lo mismo en sentido contrario: el broker le confirma al publisher que recibió el mensaje. Se activa por canal (createConfirmChannel en amqplib), y a partir de ahí cada publicación recibe exactamente una respuesta:
-
basic.ack: el broker se hace responsable del mensaje. Si es persistente y va a una cola clásica durable, significa que ya está escrito en disco. En una cola quorum, que la mayoría de las réplicas lo tiene en su log de Raft, sea cual sea el modo de entrega, porque las colas quorum persisten todos los mensajes. Con un solo nodo, esa mayoría es el propio nodo, con el fsync hecho. -
basic.nack: el broker no pudo aceptarlo, normalmente por un error interno o porque una cola llena rechaza publicaciones. Es raro, pero pasa. - Ninguna respuesta, porque la conexión se cayó antes. Este es el caso que importa, y no significa que el mensaje se perdiera. Significa que no lo sabes: puede que el broker lo escribiera y perdiera el socket antes de mandar el ack.
De ese tercer caso salen directamente dos consecuencias.
Un mensaje sin confirmar hay que enviarlo otra vez, así que los duplicados forman parte del diseño. Al reenviar tras reconectar, a veces llegará un pedido que el broker ya tenía. Por eso el consumidor de la Parte 1 manda el id del pedido como clave de idempotencia. El ensayo del final cuenta esos duplicados en lugar de fingir que no existen.
Que llegue el confirm no significa que el mensaje se enrutara. El otro problema de la Parte 1 era una routing key mal escrita que no coincidía con ningún binding. Con mandatory: true el broker devuelve el mensaje (basic.return), y después igual manda el basic.ack, porque para él ya terminó su trabajo con ese mensaje. El return siempre llega antes que el ack, así que un publisher que registra los returns puede distinguir "confirmado" de "confirmado y tirado a la basura".
Los confirms no son lentos. Esperarlos de uno en uno, sí.
La objeción habitual contra los confirms es el rendimiento. Medí tres maneras de publicar los mismos 20.000 pedidos persistentes en la cola quorum, dos corridas de cada una:
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 Esperar cada confirm antes de mandar el siguiente hunde el rendimiento en dos órdenes de magnitud, porque cada mensaje espera, él solo, a que el broker lo escriba en disco y responda. Mantener una ventana de mensajes en vuelo e ir emparejando los confirms a medida que llegan conserva la mayor parte del rendimiento: 4.784–5.728 msg/s frente a 6.859–7.425 sin confirms. Esa diferencia es lo que cuesta saberlo, y es poca al lado de la alternativa. Las cifras absolutas son de Docker Desktop en un portátil con Windows, donde un fsync a través de la máquina virtual es lento, así que las tuyas serán otras. Lo que importa es la proporción. Si te quedas con una sola frase de esta sección, que sea esta: nunca hagas await de un confirm dentro de un bucle que publica.
El publisher
Este es el lado del checkout, con confirms, una ventana limitada de mensajes en vuelo y reenvío tras reconectar. Sustituye al channel.publish a secas de la Parte 1 y a su manejo de drain.
// 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; }; /** Acota la memoria y es la contrapresión: quien publica espera si hay tantos sin confirmar. */ const MAX_IN_FLIGHT = 1_000; // amqplib rechaza el callback del confirm con este mensaje cuando el broker hace nack, y // con "channel closed" cuando el canal muere antes. Solo vale la pena reenviar el segundo. const BROKER_NACK = 'message nacked'; type Unconfirmed = { routingKey: string; content: Buffer; options: Options.Publish & { messageId: string }; /** Se escribió en un socket al menos una vez, así que el broker puede tenerlo. */ sent: boolean; resolve: () => void; reject: (error: Error) => void; }; export class UnroutableError extends Error { override name = 'UnroutableError'; } /** El plazo venció antes de que el pedido llegara a un socket. Nunca se publicó. */ export class NotPublishedError extends Error { override name = 'NotPublishedError'; } /** * El plazo venció después de enviar el pedido pero antes del confirm. Sigue en cola y * se reenvía hasta que se confirme, así que no se debe informar como fallido. */ 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; } /** * Se resuelve cuando el broker confirma el pedido. Se rechaza si el broker lo rechaza, * si ninguna cola está enlazada a la routing key o si `signal` se aborta antes. Mientras * el broker no responde, espera: el pedido se reenvía en la siguiente conexión. */ 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); // Sin canal estamos entre dos conexiones. El pedido espera en #unconfirmed y // #attach lo envía en cuanto vuelva a haber canal. if (this.#channel) this.#send(this.#channel, key, message); signal?.addEventListener('abort', () => { if (!this.#unconfirmed.has(key)) return; if (!message.sent) { // Nunca salió de este proceso: se puede descartar y dar por fallido sin riesgo. this.#settle(key); reject(new NotPublishedError(`${order.orderId} was not published before the deadline`)); } else { // El broker quizá ya lo tiene. Se sigue reenviando; quien llamó solo deja de esperar. reject(new PublishPendingError(`${order.orderId} is sent but not yet confirmed`)); } }, { once: true }); return promise; } /** Espera los confirms pendientes (con un plazo máximo) y luego cierra la conexión. */ 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(); } /** Se ejecuta después de cada conexión y reconexión. */ async #attach(model: ChannelModel): Promise<void> { const channel = await model.createConfirmChannel(); await assertTopology(channel); // Los returns pertenecen al canal que los vio. Un return cuyo confirm murió con el canal // viejo no debe marcar el reenvío, que esta vez puede enrutarse sin problema. this.#returned.clear(); // Con `mandatory`, un mensaje sin destino vuelve como basic.return *antes* de su // confirm, y ese confirm sigue siendo un ack. Anotamos el return para leer bien el ack. channel.on('return', (returned) => { const id: unknown = returned.properties.messageId; if (typeof id === 'string') this.#returned.add(id); }); // Un error de canal deja la conexión viva, así que la recuperación nunca se activa y // todo lo no confirmado esperaría para siempre. Cerrar la conexión le pasa el problema // a la recuperación, que espera, reconecta y vuelve a ejecutar este método. 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; // Lo que la conexión anterior dejó sin confirmar sale otra vez. Parte puede estar ya // en disco y solo se perdió el confirm: eso es un duplicado, y la clave de // idempotencia del consumidor es lo que lo vuelve inofensivo. 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; // se reenvía al reconectar 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 { // El canal se cerró mientras publicábamos. El mensaje sigue en #unconfirmed. } } #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(); } } } Lo que conviene mirar, de más a menos importan...
