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 importante:
-
#unconfirmedes todo el diseño. Un mensaje entra ahí antes de enviarse y solo sale cuando el broker responde. Lo que siga dentro cuando una conexión muere,#attachlo reenvía en la siguiente. Ese buffer en memoria no es una caché: es la lista de pedidos cuyo destino todavía no conoces. -
La ventana también frena a quien publica. La Parte 1 esperaba el evento
draincuando se llenaba el buffer del socket. Aquí quien llama espera en cuanto hay 1.000 mensajes sin confirmar, y eso también pone techo al buffer del socket. Con el broker caído,publish()se queda esperando, y la petición del checkout con él. Es lo honesto: si una petición no se puede guardar de forma duradera, no debería responder que todo salió bien. - Distingue un nack de un canal muerto. amqplib avisa de las dos cosas por el mismo callback, con mensajes distintos. Un nack es el broker diciendo que no, así que quien llamó recibe un error. Un canal cerrado no dice nada sobre el mensaje, así que se queda en la cola para la siguiente conexión.
- Ante un error de canal, cierra la conexión a propósito. Parece exagerado, y es deliberado. Lo explico en la siguiente sección.
-
Un plazo solo puede cancelar lo que nunca se envió. Si pasas
{ signal: AbortSignal.timeout(…) }, recibes uno de dos errores.NotPublishedErrorsignifica que el pedido nunca llegó a un socket: no existe para el broker, y quien llamó puede informar del fallo sin riesgo.PublishPendingErrorsignifica que se envió y que el broker quizá ya lo tiene, así que sigue en cola y se reenvía hasta que llegue el confirm. Si tratas ese segundo caso como un fallo, el cliente ve "el pago falló" y se le cobra igual.
Desde el checkout se usa como cualquier otra llamada asíncrona:
// src/checkout.ts
import { randomUUID } from 'node:crypto';
import { OrderPublisher, UnroutableError } from './order-publisher.ts';
const COUNT = Number(process.argv[2] ?? 10);
const publisher = await OrderPublisher.start();
const results = await Promise.allSettled(
Array.from({ length: COUNT }, () =>
publisher.publish('order.us.placed', {
orderId: `ord-${randomUUID()}`,
sku: 'BEAN-ETH-250',
quantity: 1,
totalCents: 1450,
}),
),
);
// Una routing key sin ningún binding: el confirm llega, pero no es un éxito.
try {
await publisher.publish('order.us.refunded', {
orderId: `ord-${randomUUID()}`, sku: 'BEAN-ETH-250', quantity: 1, totalCents: -1450,
});
} catch (error) {
if (!(error instanceof UnroutableError)) throw error;
console.warn(`[checkout] ${error.message}`);
}
const confirmed = results.filter((result) => result.status === 'fulfilled').length;
console.log(`[checkout] ${confirmed}/${COUNT} orders confirmed by the broker`);
await publisher.close();
$ node src/checkout.ts 100
[checkout] no queue bound for order.us.refunded
[checkout] 100/100 orders confirmed by the broker
La línea del reembolso es la routing key mal escrita de la Parte 1, que esta vez sí se detecta: el confirm llegó, y aun así el publisher sabía que no era un éxito.
Y los dos resultados posibles del plazo, con una prueba que publica una vez con el broker detenido y otra con el broker congelado (docker pause: el pedido llega al socket, pero el confirm nunca vuelve). Las líneas de reconexión están filtradas:
$ node src/deadline-probe.ts | grep -v '^[checkout-publisher]'
broker stopped -> NotPublishedError: ord-deadline-unsent was not published before the deadline
broker frozen -> PublishPendingError: ord-deadline-sent is sent but not yet confirmed
in orders.billing -> ord-deadline-sent
El pedido que nunca salió del proceso no se publicó. El que estaba pendiente cuando venció su plazo llegó de todos modos, y por eso no se dio por fallido.
Conexiones, canales y qué arrastra a qué
Una conexión es un socket TCP con su negociación AMQP: abrirla es caro, y lo normal es que dure lo mismo que el proceso. Un canal es una sesión virtual ligera que viaja multiplexada dentro de esa conexión. Casi todo el protocolo ocurre en los canales, y por eso el canal es la unidad de fallo.
De ahí salen tres reglas. En una de ellas, Node matiza el consejo habitual.
1. Un error de canal cierra el canal, y solo el canal. Si redeclaras una cola con otros argumentos, publicas en un exchange que no existe o haces ack de un delivery tag que el canal no conoce, el broker cierra ese canal con un error. La conexión sigue viva. Todos los consumidores y todos los confirms pendientes de ese canal desaparecen, y nada se reconecta, porque desde el punto de vista de la conexión no pasó nada. Tu consumidor simplemente deja de recibir mensajes. Por eso tanto el publisher como el consumidor de este artículo reaccionan a channel.on('error') cerrando la conexión: convierten un fallo a medias, silencioso, en un fallo completo, y la recuperación sabe qué hacer con un fallo completo.
2. Conexiones separadas para publicar y para consumir. Cuando RabbitMQ se queda corto de memoria o de disco, levanta una alarma y bloquea las conexiones que publican hasta que la alarma se apaga. Si tus consumidores comparten esa conexión, sus acks también se quedan atascados detrás del bloqueo, y la cola que más necesita vaciarse deja de vaciarse. Una conexión por rol evita ese bloqueo mutuo. El connection_name del código de abajo hace que cada rol aparezca con su nombre en la pestaña Connections de la UI de administración.
3. Los delivery tags pertenecen a un canal. Un delivery tag es un contador local al canal que entregó el mensaje. Si haces ack en cualquier otro canal, incluido el canal nuevo después de reconectar, estás confirmando un mensaje ajeno. Lo normal es que el tag no exista ahí, y eso es un error de canal (regla 1).
El consejo que leerás en otros sitios, "nunca compartas un canal entre hilos", es literal en .NET y en Java, donde dos hilos que escriben frames en el mismo canal los mezclan. Node tiene un solo hilo, así que ese error concreto no puede pasar aquí. Las reglas 1 y 3 son la versión para Node de la misma lección: todo lo que comparte un canal corre la misma suerte. Pon cosas en el mismo canal solo si te parece bien que caigan juntas.
Reconexión: lo que hace amqplib y lo que te deja a ti
amqplib 1.1 añadió recuperación: le pasas opciones de recovery a connect, y cada vez que el socket se cae la conexión vuelve a intentarlo, con esperas que crecen de forma exponencial (backoff) y una variación aleatoria (jitter). Hasta ahí funciona como promete. Lo que no hace, y que solo descubrí leyendo lib/recovery.js, es esto:
- No recupera los canales. Un canal pertenece a un socket concreto. Cuando ese socket muere, todos sus canales quedan cerrados para siempre, y los objetos que guardas ya no apuntan a nada.
- No restaura consumidores ni topología. Viven en los canales.
-
No reenvía nada. Los mensajes sin confirmar son asunto tuyo, y por eso
OrderPublisherlos guarda.
A cambio te da setup: una función que se ejecuta después de cada conexión exitosa, incluida la primera. Todo lo que vive en un canal se reconstruye ahí. La Parte 1 hizo que assertTopology fuera idempotente justo por esto.
// src/connection.ts
import amqp from 'amqplib';
import type { ChannelModel, RecoveringChannelModel } from 'amqplib';
export const RABBIT_URL = process.env.RABBIT_URL ?? 'amqp://aurora:aurora-dev@localhost:5672';
// Un broker que se cae cierra el socket y se nota al instante. Uno detrás de una red rota
// no manda nada, y solo los heartbeats perdidos lo delatan. RabbitMQ sugiere 60 s; con
// 15 s, un broker congelado se detectó a los 45 s.
const HEARTBEAT_SECONDS = 15;
/**
* Una conexión por proceso *y por rol*. El nombre aparece en la pestaña Connections de
* la UI de administración, y así distingues tres réplicas de facturación de un publisher.
*
* `setup` se ejecuta después de cada conexión exitosa, incluida la primera. amqplib
* reconecta el socket por su cuenta; canales, consumidores y topología los reconstruyes
* tú, y este es el único lugar donde se reconstruyen.
*/
export async function connectWithRecovery(
role: string,
setup: (model: ChannelModel) => Promise<void>,
): Promise<RecoveringChannelModel> {
const url = new URL(RABBIT_URL);
url.searchParams.set('heartbeat', String(HEARTBEAT_SECONDS));
const connection = await amqp.connect(url.toString(), {
clientProperties: { connection_name: role },
// Sin esto, un intento contra un host que nunca responde espera al timeout de TCP del
// sistema, y la recuperación no programa el siguiente intento hasta que ese se rinda.
timeout: 10_000,
recovery: {
initialDelay: 500,
maxDelay: 15_000,
// Un broker inalcanzable al arrancar es un error de configuración: que falle el despliegue.
initialMaxRetries: 5,
// Un broker que desaparece más tarde es un incidente: seguir intentando hasta que vuelva.
maxRetries: Infinity,
setup,
},
});
// Sin este listener, la recuperación nunca llega a actuar: el error del socket se
// reemite como 'error' en la conexión, y un 'error' sin manejar termina el proceso.
connection.on('error', (error) => console.warn(`[${role}] connection error: ${error.message}`));
connection.on('disconnect', (error) => console.warn(`[${role}] disconnected: ${error.message}`));
connection.on('reconnect-scheduled', ({ attempt, delay }) =>
console.warn(`[${role}] reconnect attempt ${attempt} in ${delay} ms`),
);
connection.on('connect', () => console.log(`[${role}] reconnected`));
connection.on('blocked', (reason) => console.warn(`[${role}] blocked by broker: ${reason}`));
connection.on('unblocked', () => console.log(`[${role}] unblocked`));
connection.on('reconnect-failed', (error) => {
console.error(`[${role}] giving up on the broker: ${error.message}`);
process.exitCode = 1;
});
return connection;
}
Ese listener de error es la línea que estuve a punto de dejar fuera. El ejemplo de recuperación del README no lo incluye, y la primera vez que mi ensayo mató el broker con la recuperación activada, el proceso murió igual. Aquí está reproducido en un contenedor: un cliente que se conecta con opciones de recovery, solo registra un listener de disconnect y luego pierde el broker.
connected with recovery enabled
/app/node_modules/amqplib/lib/safe_emit.js:8
throw e;
^
Error: read ECONNRESET
at TCP.onStreamRead (node:internal/stream_base_commons:216:20)
Emitted 'error' event on RecoveringPromiseModel instance at:
at RecoveringCore.<anonymous> (/app/node_modules/amqplib/lib/recovery.js:445:46)
at RecoveringCore.emit (node:events:514:28)
at ChannelModel.onConnectionError (/app/node_modules/amqplib/lib/recovery.js:269:45)
at ChannelModel.emit (node:events:514:28)
at Connection.emit (node:events:514:28)
at safeEmit (/app/node_modules/amqplib/lib/safe_emit.js:3:13)
at Connection.onSocketError (/app/node_modules/amqplib/lib/connection.js:300:7)
at Socket.emit (node:events:526:24)
at emitErrorNT (node:internal/streams/destroy:170:8)
at emitErrorCloseNT (node:internal/streams/destroy:129:3) {
errno: -104,
code: 'ECONNRESET',
syscall: 'read'
}
Node.js v24.21.0
La conexión con recuperación reemite el error del socket como 'error', y un EventEmitter sin listener de 'error' lanza una excepción. El listener de disconnect nunca llegó a ejecutarse. La recuperación estaba lista para reconectar, y el proceso ya no existía cuando le tocaba. Una conexión que se reconecta sola es una función muy convincente, hasta el momento en que se lleva el proceso por delante.
Hay cuatro ajustes más que vale la pena copiar:
-
initialMaxRetriesaparte demaxRetries. Si no puedes llegar al broker al arrancar, casi siempre es una URL o una contraseña equivocada, y el despliegue debería fallar de forma visible. Si el broker desaparece más tarde, es un incidente, y el servicio debería esperar a que pase. -
Un heartbeat lo bastante corto para notar el silencio. Un broker al que matan cierra el socket, y por eso el ensayo de más abajo detecta cada caída al instante. Un broker detrás de una red rota no manda nada, y solo los heartbeats perdidos lo delatan. En la prueba de abajo, con un heartbeat de 15 segundos, el publisher notó el broker congelado a los 45 segundos. RabbitMQ sugiere 60, y con eso el mismo silencio dura minutos, mientras los pedidos se acumulan en
#unconfirmed. -
Un timeout de conexión. Sin
timeout, un intento contra un host que nunca responde espera al timeout de TCP del sistema operativo, y la recuperación no puede programar el siguiente hasta que ese se rinda. En la prueba, el intento 2 se programó diez segundos después del intento 1. - El jitter no es un adorno. Cuando el ensayo de más abajo reinicia el broker con tres réplicas consumiendo, los logs muestran a cada una reintentando a su propio ritmo (522 ms, 466 ms y 427 ms en el primer intento). Sin jitter, las tres se reconectan en el mismo milisegundo, y con cien réplicas el pico de carga también llega todo junto.
La prueba detrás de las cifras del heartbeat y del timeout: congela el broker con docker pause, lo más parecido en local a una red rota, y publica un pedido.
[+0.2 s] [probe] broker frozen with docker pause
[+45.0 s] [checkout-publisher] connection error: Heartbeat timeout
[+45.0 s] [checkout-publisher] disconnected: Heartbeat timeout
[+45.0 s] [checkout-publisher] reconnect attempt 1 in 587 ms
[+55.6 s] [checkout-publisher] reconnect attempt 2 in 904 ms
[+60.4 s] [probe] broker unfrozen
[checkout-publisher] reconnected
[+60.5 s] [probe] order confirmed
El consumidor
El consumidor recibe cuatro cambios. Lo que no cambia (parseOrder, retriesSoFar y chargeCard) pasa tal cual de la Parte 1 a src/billing.ts, con export añadido. Está en el apéndice del final, para que el consumidor de abajo se centre en lo nuevo.
// src/billing-consumer.ts
import type { ChannelModel, ConfirmChannel, ConsumeMessage, Options } from 'amqplib';
import { chargeCard, parseOrder, RETRY_COUNT_HEADER, retriesSoFar } from './billing.ts';
import { connectWithRecovery } from './connection.ts';
import { assertTopology, BILLING_QUEUE, RETRY_EXCHANGE } from './topology.ts';
const MAX_RETRIES = 3;
// En Node no solo reparte la carga: limita cuántos handlers corren a la vez.
const PREFETCH = Number(process.env.BILLING_PREFETCH ?? 20);
const inFlight = new Set<Promise<void>>();
// Ids de mensajes que el broker devolvió por no tener destino, por canal. Ver publishConfirmed.
const returnedOn = new WeakMap<ConfirmChannel, Set<string>>();
let active: { channel: ConfirmChannel; consumerTag: string } | null = null;
let shuttingDown = false;
/**
* Solo se resuelve si el broker confirmó la copia *y* la enrutó a una cola. Un mensaje
* sin destino también recibe ack del broker; sin `mandatory` y sin revisar el return,
* un binding que falta haría que se confirmara el original y se perdiera el pedido.
*/
function publishConfirmed(
channel: ConfirmChannel,
exchange: string,
routingKey: string,
content: Buffer,
options: Options.Publish & { messageId: string },
): Promise<void> {
return new Promise((resolve, reject) => {
channel.publish(exchange, routingKey, content, { ...options, mandatory: true }, (error) => {
const returned = returnedOn.get(channel)?.delete(options.messageId) ?? false;
if (error) reject(error);
else if (returned) reject(new Error(`no queue bound for ${routingKey} on ${exchange}`));
else resolve();
});
});
}
/**
* Los delivery tags pertenecen al canal que entregó el mensaje. Si ese canal murió a
* mitad del handler, el broker ya devolvió el mensaje a la cola y el ack no tiene nada
* que confirmar. Nunca hagas ack en el canal *nuevo*: ahí un tag desconocido es un error.
*/
function settle(action: () => void): void {
try {
action();
} catch {
// Canal cerrado: el mensaje se volverá a entregar con `redelivered: true`.
}
}
async function handle(channel: ConfirmChannel, message: ConsumeMessage): Promise<void> {
let order;
try {
order = parseOrder(message);
} catch (error) {
console.error('[billing] unparseable message, dead-lettering', error);
settle(() => channel.nack(message, false, false));
return;
}
try {
const chargeId = await chargeCard(order);
console.log(`[billing] charged ${order.orderId} -> ${chargeId}`);
settle(() => channel.ack(message));
return;
} catch (error) {
const attempts = retriesSoFar(message);
if (attempts >= MAX_RETRIES) {
console.error(`[billing] giving up on ${order.orderId} after ${attempts} retries`, error);
settle(() => channel.nack(message, false, false));
return;
}
console.warn(`[billing] retry ${attempts + 1}/${MAX_RETRIES} for ${order.orderId}`, error);
try {
// La Parte 1 publicaba y hacía ack seguidos. Ahora el ack espera a que el broker
// confirme que la copia está a salvo en la cola de reintentos.
await publishConfirmed(channel, RETRY_EXCHANGE, message.fields.routingKey, message.content, {
persistent: true,
headers: { ...message.properties.headers, [RETRY_COUNT_HEADER]: attempts + 1 },
messageId: order.orderId,
});
settle(() => channel.ack(message));
} catch (publishError) {
// La copia para reintento no está confirmada, así que el original no puede recibir
// ack. Devolverlo a la cola sería un bucle: en RabbitMQ 4.x un requeue explícito no
// cuenta para x-delivery-limit. Se manda a dead-letter, donde una persona lo verá.
console.error(`[billing] could not park ${order.orderId} for retry, dead-lettering`, publishError);
settle(() => channel.nack(message, false, false));
}
}
}
/** Se ejecuta tras cada (re)conexión: un canal nuevo, la topología y el consumidor. */
async function attach(model: ChannelModel): Promise<void> {
if (shuttingDown) return;
const channel = await model.createConfirmChannel();
await assertTopology(channel);
await channel.prefetch(PREFETCH);
const returned = new Set<string>();
returnedOn.set(channel, returned);
channel.on('return', (message) => {
const id: unknown = message.properties.messageId;
if (typeof id === 'string') returned.add(id);
});
// La misma regla que en el publisher: un canal muerto en una conexión viva es un
// consumidor que se detuvo en silencio. Lo escalamos para que la recuperación rehaga todo.
channel.on('error', (error: Error) => {
console.error(`[billing] channel error, rebuilding: ${error.message}`);
void model.close().catch(() => undefined);
});
const { consumerTag } = await channel.consume(BILLING_QUEUE, (message) => {
if (message === null) {
// El broker canceló el consumidor (por ejemplo, borraron la cola). Al rehacer, se redeclara.
void model.close().catch(() => undefined);
return;
}
const task: Promise<void> = handle(channel, message)
.catch((error: unknown) => console.error('[billing] handler crashed', error))
.finally(() => inFlight.delete(task));
inFlight.add(task);
});
active = { channel, consumerTag };
}
const connection = await connectWithRecovery('billing-consumer', attach);
console.log(`[billing] consuming ${BILLING_QUEUE} with prefetch ${PREFETCH}`);
async function shutdown(): Promise<void> {
shuttingDown = true;
console.log(`[billing] shutting down with ${inFlight.size} in flight`);
// 1. Cortar las entregas nuevas, pero dejar el canal abierto para que lo que está en curso haga ack.
if (active) await active.channel.cancel(active.consumerTag).catch(() => undefined);
// 2. Dejar terminar lo que ya corre. chargeCard expira a los 5 s, así que esto tiene límite.
await Promise.allSettled([...inFlight]);
// 3. Cerrar primero el canal. Su cierre queda en cola detrás de los acks del mismo canal;
// cerrar la conexión directamente puede adelantarse al último ack y devolver un pedido ya cobrado.
if (active) await active.channel.close().catch(() => undefined);
await connection.close();
}
for (const signal of ['SIGINT', 'SIGTERM'] as const) {
process.once(signal, () => {
void shutdown().then(() => process.exit(0));
});
}
Los cuatro cambios respecto a la Parte 1:
-
La copia para reintento se confirma, y se enruta, antes de hacer ack del original. Esto cierra el hueco que reconocía la Parte 1:
publishConfirmedespera al broker, y solo entoncesackborra el original. Además publica conmandatoryy trata un return como un fallo, porque si faltara el binding de la cola de reintentos el broker confirmaría igual y el pedido se perdería. Si la copia no se puede aparcar, el original va a dead-letter en lugar de volver a la cola. Un requeue no cuenta parax-delivery-limit, así que se repetiría tan rápido como lo permita la red. Con el binding de reintentos borrado, el pedido llegó aorders.billing.deadtras un solo intento. Los reintentos se cuentan enretry-count, la corrección que explico al principio. -
Todo lo que vive en un canal se reconstruye en
attach: el canal, la topología, el prefetch y el consumidor. Es todo lo que la recuperación necesita de ti. -
El ack va al canal que entregó el mensaje, y solo a ese.
handlerecibe su canal como argumento y nunca lee una variable de "canal actual". Si ese canal murió a mitad del handler,settleabsorbe el ack fallido. El broker devolvió el mensaje a la cola en cuanto se cerró el canal, así que ya no quedaba nada que confirmar. - El apagado termina el trabajo antes de cerrar. Eso merece sección propia.
El apagado, y el ack que nunca llegó
Este era el apagado de la Parte 1:
for (const signal of ['SIGINT', 'SIGTERM'] as const) {
process.once(signal, () => {
// Cerrar el canal devuelve a la cola todo lo que quedó sin confirmar, así que un
// despliegue progresivo le pasa el trabajo en curso al siguiente pod en vez de
// perderlo.
void channel
.close()
.then(() => connection.close())
.then(() => process.exit(0));
});
}
El comentario es cierto, pero oculta un costo. Cerrar el canal con handlers todavía en marcha significa que cada pedido en curso que ya llamó al servicio de pagos, pero aún no hizo ack, vuelve a la cola y se cobra otra vez en la siguiente réplica. La clave de idempotencia convierte ese segundo cobro en una operación sin efecto, así que nadie paga dos veces. Aun así es trabajo desperdiciado, y además es apostar a que todos los servicios siguientes respetan las claves de idempotencia.
El shutdown nuevo lo hace en tres pasos: cancela el consumidor para que no lleguen mensajes nuevos, espera a los handlers en curso y luego cierra. Medí las dos versiones de la misma forma: una réplica, prefetch 20, 1.000 pedidos y docker stop a los 2,5 segundos, contando después cuántos cobros vio repetidos el stub de pagos. La tabla la armé con esas corridas:
| Consumidor | Corrida 1 | Corrida 2 | Corrida 3 |
|---|---|---|---|
| Parte 1 (cierra el canal de inmediato) | 16 | 5 | 8 |
| Parte 2, cerrando la conexión tras terminar lo pendiente | 1 | 1 | 2 |
| Parte 2, cerrando el canal y después la conexión | 0 | 0 | 0 |
La fila del medio me costó casi una hora. Esperar a lo pendiente funcionaba: la primera vez que me pasó, el log decía que había 15 handlers en curso, y los 15 cobraron y llamaron a ack. Aun así, uno o dos pedidos se volvían a entregar en cada corrida. El origen estaba en cómo escribe amqplib los frames: cada canal tiene su propio buffer de escritura, y un multiplexor los va intercalando en el socket. El ack espera en el buffer del canal 1, mientras que connection.close() sale por el buffer del canal 0. El cierre puede adelantarse al último ack. El broker ve que la conexión se cierra y devuelve a la cola un mensaje que ya estaba terminado.
Cerrar primero el canal lo resuelve, porque channel.close() queda en cola detrás de los acks de ese mismo canal. Es una línea, y solo apareció porque el stub cuenta las repeticiones.
Escalar: consumidores en competencia y prefetch
Si varios consumidores leen la misma cola, RabbitMQ les reparte los mensajes por turnos. Es el patrón de consumidores en competencia (competing consumers), y es todo lo que hace falta para escalar una cola horizontalmente: no hay particiones que planificar ni coordinador. Arrancas otra réplica y empieza a recibir mensajes.
El archivo de compose suma los servicios nuevos. El de RabbitMQ es el de la Parte 1 con una línea más, que explico en la sección sobre reinicios:
# docker-compose.yml
services:
rabbitmq:
image: rabbitmq:4-management
container_name: aurora-rabbitmq
# El nodo se llama rabbit@<hostname>, y su directorio de datos lleva el nombre del nodo.
# Sin un hostname fijo, un contenedor recreado es un nodo nuevo y sin colas.
hostname: aurora-rabbitmq
ports:
- "5672:5672" # AMQP — aquí se conecta tu app
- "15672:15672" # UI de administración — aquí te conectas tú
environment:
RABBITMQ_DEFAULT_USER: aurora
RABBITMQ_DEFAULT_PASS: aurora-dev
volumes:
- rabbitmq-data:/var/lib/rabbitmq
healthcheck:
test: ["CMD", "rabbitmq-diagnostics", "-q", "check_running"]
interval: 10s
timeout: 5s
retries: 5
payments:
build: .
command: ["node", "src/payments-stub.ts"]
environment:
PAYMENTS_LATENCY_MS: "50"
ports:
- "4001:4001"
billing:
build: .
environment:
RABBIT_URL: amqp://aurora:aurora-dev@rabbitmq:5672
PAYMENTS_URL: http://payments:4001
BILLING_PREFETCH: ${BILLING_PREFETCH:-20}
depends_on:
rabbitmq:
condition: service_healthy
payments:
condition: service_started
# Margen para que shutdown() termine lo que está en curso antes de que Docker mande SIGKILL.
stop_grace_period: 30s
volumes:
rabbitmq-data:
FROM node:24-alpine
WORKDIR /app
COPY package.json package-lock.json ./
RUN npm ci --omit=dev
COPY src ./src
USER node
CMD ["node", "src/billing-consumer.ts"]
Node 24 ejecuta los .ts directamente, así que la imagen no necesita paso de compilación, igual que no lo necesita tu máquina. El stub de pagos, un servidor HTTP de unas 50 líneas que tarda 50 ms por cobro, respeta las claves de idempotencia incluso con peticiones simultáneas con la misma clave y cuenta las repeticiones, está en el apéndice.
El prefetch es tu límite de concurrencia
En la Parte 1, el prefetch servía para repartir la carga: sin él, el primer consumidor que se conecta se queda con toda la cola. En Node es algo más. Fíjate en el callback del consumidor: lanza handle() y retorna enseguida, sin esperar a nada. Por eso el prefetch pone techo al número de handlers que corren a la vez en un proceso: el broker no manda más mensajes sin ack que ese número. Tres réplicas con prefetch 20 suponen hasta 60 llamadas simultáneas al servicio de pagos.
Los mismos 3.000 pedidos, tres réplicas y el stub tardando 50 ms por cobro:
$ BILLING_PREFETCH=1 docker compose up -d --wait --scale billing=3 billing
$ BILLING_PREFETCH=1 node src/measure-drain.ts 3000
prefetch 1: 3000 orders charged in 66.3 s (45 orders/s)
$ BILLING_PREFETCH=20 docker compose up -d --wait --scale billing=3 billing
$ BILLING_PREFETCH=20 node src/measure-drain.ts 3000
prefetch 20: 3000 orders charged in 5.8 s (516 orders/s)
$ BILLING_PREFETCH=100 docker compose up -d --wait --scale billing=3 billing
$ BILLING_PREFETCH=100 node src/measure-drain.ts 3000
prefetch 100: 3000 orders charged in 3.7 s (821 orders/s)
Con prefetch 1, tres réplicas se comportan como tres trabajadores de un solo hilo: en teoría 3 × (1 / 50 ms) da 60 pedidos por segundo, y medí 45. Con prefetch 20 va once veces más rápido. Con prefetch 100 va más rápido todavía, pero mucho menos de cinco veces, porque 300 handlers simultáneos podrían cobrar en teoría 6.000 pedidos por segundo, y algo que no son los consumidores se agota antes. Aun así, significa 300 peticiones simultáneas contra el proveedor de pagos. Con un proveedor real que te limita las peticiones, así es como llegan los 429 y una tormenta de reintentos. Calcula el prefetch a partir de lo que acepta el servicio más lento, dividido entre el número de réplicas, no a partir de lo que la cola puede entregar. La cola siempre va a entregar más rápido de lo que el proveedor puede cobrar.
Lo que pierdes: el orden
Ya lo decía la Parte 1: el orden dentro de una cola solo se mantiene mientras la lee un único consumidor. Los consumidores en competencia procesan pedidos en paralelo, así que el pedido 41 puede terminar antes que el 40. Para facturación no importa. Si tu consumidor sí necesita orden (por ejemplo, para aplicar cambios de saldo de una cuenta), declara la cola con x-single-active-consumer: true. Todas las réplicas siguen conectadas, pero solo una recibe mensajes, y otra toma el relevo cuando esa se va. Obtienes conmutación por error, no paralelismo. Y como es un argumento de la cola, se aplica el otro aviso de la Parte 1: decídelo antes de que la cola exista, porque después no se puede añadir.
Cuando el broker se reinicia
El reinicio que conserva tus datos y el que no
El compose de la Parte 1 declaraba un volumen con nombre para /var/lib/rabbitmq, lo que da a entender que los datos persisten. Persisten mientras el contenedor siga existiendo. Esto es lo que pasa cuando no:
$ docker compose up -d --wait && node src/checkout.ts 100
[checkout] no queue bound for order.us.refunded
[checkout] 100/100 orders confirmed by the broker
$ docker exec aurora-rabbitmq rabbitmqctl -q list_queues name messages
name messages
orders.billing 100
orders.billing.retry 0
orders.billing.dead 0
$ docker compose down && docker compose up -d --wait
$ docker exec aurora-rabbitmq rabbitmqctl -q list_queues name messages
$ docker exec aurora-rabbitmq ls /var/lib/rabbitmq/mnesia
rabbit@8dfd00d4dcd6
rabbit@8dfd00d4dcd6-feature_flags
rabbit@8dfd00d4dcd6-plugins-expand
rabbit@ffe1f6cff9d8
rabbit@ffe1f6cff9d8-feature_flags
rabbit@ffe1f6cff9d8-plugins-expand
rabbit@ffe1f6cff9d8.pid
Cien pedidos confirmados, y después ninguna cola, con el volumen todavía guardando los datos viejos. Un nodo de RabbitMQ se llama rabbit@<hostname>, su directorio de datos lleva el nombre del nodo, y el hostname por defecto de un contenedor es su id aleatorio. docker compose down seguido de up crea un contenedor nuevo: arranca un nodo nuevo, no encuentra ningún directorio con su nombre y empieza vacío. Los datos del nodo anterior siguen ahí al lado, intactos e ignorados. docker restart conserva el contenedor y su hostname, y por eso el problema no aparece en las pruebas del día a día y salta con la primera actualización de imagen, que sí recrea el contenedor.
La solución es la línea que añade el compose nuevo:
container_name: aurora-rabbitmq
hostname: aurora-rabbitmq
La misma prueba con esa línea: 100 mensajes antes del down, 100 después y un único directorio de datos. Los StatefulSets de Kubernetes dan a los pods hostnames estables por la misma razón.
Lo que un solo nodo puede prometer y lo que no
Una cola quorum en un solo nodo te da durabilidad: un mensaje confirmado ya pasó por el fsync antes de que salga el ack. Lo que no te da es disponibilidad: mientras ese nodo está caído, no se publica ni se consume nada. La razón de ser de las colas quorum es la replicación: en un clúster de tres nodos, una cola sobrevive a la caída de uno y sigue atendiendo. Eso cambia lo que pasa durante la caída, no el código del cliente. El cliente sí necesita un cambio. amqp.connect recibe una sola URL y la recuperación sigue llamando a esa misma dirección, así que si ese nodo es el que cayó, el cliente espera a que vuelva en lugar de pasarse a otro. Pon delante de los nodos un balanceador TCP como HAProxy, o un nombre DNS que solo resuelva a nodos sanos, y conéctate a eso. Todo lo demás de este artículo funciona igual contra un clúster. Montar uno en local da para un compose propio.
El ensayo: matar el broker a mitad del lote y contar
La preparación es esta. Nadie consume la cola durante la corrida. El script publica 20.000 pedidos a unos 10.000 por segundo. Tras el pedido 10.000 ejecuta docker kill aurora-rabbitmq (SIGKILL, sin apagado ordenado) y luego docker start, sin dejar de publicar en ningún momento, como haría el tráfico real del checkout. Cuando el broker vuelve, vacía orders.billing y cuenta los ids de pedido distintos. El script completo está en el apéndice.
Conviene tener claro qué prueba esto y qué no. docker kill termina el proceso del broker, pero la caché de páginas del host sobrevive, así que lo que el broker escribió sin hacer fsync sigue ahí después. El ensayo mide lo que se pierde en tránsito: en el buffer del cliente y en la memoria del broker. No puede demostrar que un mensaje confirmado pasó por el fsync, porque estas corridas darían el mismo resultado aunque no hubiera pasado. Para probar eso hay que perder la máquina entera, por ejemplo apagando una VM de golpe. Hay tres escenarios.
A: el publisher de la Parte 1. Un canal normal. Que publish() retorne sin error es la única señal que recibe. Una concesión a la medición: el publisher de la Parte 1 no registraba ningún listener de error, así que el real se habría caído en el momento del kill. El ensayo añade los listeners para poder seguir contando. Los pedidos que faltan ya se habían aceptado de todos modos.
scenario: fire-and-forget
orders published: 20000, broker killed after #10000
broker down for 9.6 s, run took 16.2 s
publish() returned normally 10900
publish() threw 9100
messages in orders.billing 10433
distinct orders in orders.billing 10433
orders lost 9567
duplicates 0
Los 9.100 que lanzaron excepción no son el problema: el checkout vio un error y pudo avisar al cliente. El problema es la diferencia entre 10.900 y 10.433. Se aceptaron 467 pedidos sin ningún error y nunca llegaron. Estaban en el buffer del cliente, o en la memoria del broker, cuando mataron el proceso.
B: confirms, pero sin reenvío.
scenario: confirms
orders published: 20000, broker killed after #10000
broker down for 11.0 s, run took 17.3 s
confirmed by the broker 10665
confirm failed (channel closed) 1335
publish() threw 8000
messages in orders.billing 10793
distinct orders in orders.billing 10793
orders lost 9207
duplicates 0
Los 10.665 pedidos confirmados están todos en la cola. Los 1.335 cuyo confirm nunca llegó son el caso de "no lo sabes", y el conteo lo demuestra: 128 llegaron de todos modos (10.793 − 10.665) y 1.207 no. Los confirms no evitan la pérdida, pero la hacen visible, y sin verla no hay forma de corregirla.
C: OrderPublisher. Confirms, recuperación y reenvío de todo lo no confirmado.
[checkout-publisher] connection error: Unexpected close
[checkout-publisher] disconnected: Unexpected close
[checkout-publisher] reconnect attempt 1 in 530 ms
[checkout-publisher] reconnect attempt 2 in 1007 ms
[checkout-publisher] reconnect attempt 3 in 1760 ms
[checkout-publisher] reconnect attempt 4 in 3332 ms
[checkout-publisher] reconnect attempt 5 in 9301 ms
[checkout-publisher] reconnected
scenario: recovery
orders published: 20000, broker killed after #10000
broker down for 9.9 s, run took 24.8 s
confirmed by the broker 20000
rejected 0
seconds from kill to last confirm 18.4
messages in orders.billing 20128
distinct orders in orders.billing 20000
orders lost 0
duplicates 128
Están los 20.000 pedidos, más 128 duplicados: pedidos que el broker ya había escrito, cuyo confirm murió con el socket y que se enviaron otra vez. Esta corrida enseña también lo que cuesta el backoff. El broker volvió a los 9,9 segundos, pero el intento 5 ya estaba programado para dentro de 9,3, así que el publisher se quedó varios segundos parado frente a un broker sano. maxDelay es un equilibrio entre no saturar a un broker que todavía está arrancando y no tardar de más cuando ya se recuperó. Quince segundos es mi término medio. Elige el tuyo pensando en un incidente real.
Corrí cada escenario tres veces. La tabla la armé con las nueve capturas:
| Escenario | Aceptados sin error que nunca llegaron | Confirmados que nunca llegaron | Perdidos en total | Duplicados |
|---|---|---|---|---|
| A: sin confirmación | 467 / 1.406 / 1.125 | n/a | 9.567 / 9.606 / 9.125 | 0 / 0 / 0 |
| B: solo confirms | n/a | 0 / 0 / 0 | 9.207 / 10.934 / 9.400 | 0 / 0 / 0 |
C: OrderPublisher
|
n/a | 0 / 0 / 0 | 0 / 0 / 0 | 128 / 64 / 322 |
A pierde en silencio una cantidad distinta en cada corrida, según lo que hubiera en vuelo justo cuando cae el broker. B nunca pierde un mensaje confirmado y nunca reenvía uno sin confirmar. C cambia pérdidas por duplicados, que es el único cambio que permite la entrega al menos una vez.
El mismo ensayo, del lado del consumidor
La misma caída, esta vez con tres réplicas de facturación consumiendo 6.000 pedidos con prefetch 20:
$ curl -s localhost:4001/stats
{"charges":9000,"replays":0}
$ (sleep 4; docker kill aurora-rabbitmq; docker start aurora-rabbitmq) & BILLING_PREFETCH=20 node src/measure-drain.ts 6000
prefetch 20: 6000 orders charged in 28.9 s (208 orders/s)
$ curl -s localhost:4001/stats
{"charges":15000,"replays":53}
$ docker exec aurora-rabbitmq rabbitmqctl -q list_queues name messages
name messages
orders.billing 0
orders.billing.retry 0
orders.billing.dead 0
$ docker compose logs billing --since 1m --no-log-prefix | grep -v charged
[billing] consuming orders.billing with prefetch 20
[billing] consuming orders.billing with prefetch 20
[billing] consuming orders.billing with prefetch 20
[billing-consumer] connection error: read ECONNRESET
[billing-consumer] disconnected: read ECONNRESET
[billing-consumer] reconnect attempt 1 in 522 ms
[billing-consumer] reconnect attempt 2 in 938 ms
[billing-consumer] reconnect attempt 3 in 2182 ms
[billing-consumer] reconnect attempt 4 in 4550 ms
[billing-consumer] reconnect attempt 5 in 8630 ms
[billing-consumer] reconnected
[billing-consumer] connection error: read ECONNRESET
[billing-consumer] disconnected: read ECONNRESET
[billing-consumer] reconnect attempt 1 in 466 ms
[billing-consumer] reconnect attempt 2 in 1148 ms
[billing-consumer] reconnect attempt 3 in 1829 ms
[billing-consumer] reconnect attempt 4 in 4379 ms
[billing-consumer] reconnect attempt 5 in 8235 ms
[billing-consumer] reconnected
[billing-consumer] connection error: read ECONNRESET
[billing-consumer] disconnected: read ECONNRESET
[billing-consumer] reconnect attempt 1 in 427 ms
[billing-consumer] reconnect attempt 2 in 997 ms
[billing-consumer] reconnect attempt 3 in 1823 ms
[billing-consumer] reconnect attempt 4 in 3324 ms
[billing-consumer] reconnect attempt 5 in 8230 ms
[billing-consumer] reconnected
Exactamente 6.000 cobros nuevos y distintos, las tres colas vacías y 53 repeticiones. Son pedidos que una réplica ya había cobrado cuando el socket murió, antes de que su ack llegara al broker. El broker los volvió a entregar, otra réplica llamó al stub de pagos con la misma clave de idempotencia, y el stub devolvió el cobro original. A nadie se le cobró dos veces, y eso quedó decidido en la Parte 1, con un header HTTP. Todos los mecanismos de fiabilidad de este artículo generan duplicados en algún punto. La clave de idempotencia es lo que hace que no cuesten nada.
Qué usar en cada caso
Usa confirms con un buffer de reenvío cuando:
- Un mensaje perdido es una venta perdida, un cobro que no se hace o un pedido que nunca sale. Es decir, cualquier publisher cuyos mensajes alguien vaya a reclamar.
- Puedes hacer que los consumidores sean idempotentes. Si no puedes, arregla eso primero: reenviar sin idempotencia convierte los mensajes perdidos en cobros dobles.
Quédate con algo más simple cuando:
- El mensaje es telemetría, como una visita a una página o una muestra de una métrica. Perder 467 de 20.000 durante una caída del broker es aceptable, y publicar sin confirmación cuesta varias veces menos, tanto en operación como en complejidad.
-
La publicación y una escritura en base de datos tienen que ser atómicas.
OrderPublisherguarda en memoria los pedidos sin confirmar. Si el que se cae es el propio proceso del checkout, esos pedidos desaparecen con él, igual que el pedido que ya estaba guardado en Postgres. La solución es una bandeja de salida transaccional (transactional outbox): escribes el evento en una tabla de salida dentro de la misma transacción que el pedido, y un proceso aparte lo publica desde esa tabla con confirms. Es el siguiente nivel de fiabilidad, y es otra arquitectura, no un ajuste de esta.
Sobre construirlo o usar una librería:
-
La recuperación de amqplib más tu propio
setupes lo que usa este artículo: poco código, y se ve todo. -
amqp-connection-manager envuelve los canales para que sobrevivan a las reconexiones, vuelve a ejecutar tu configuración y encola las publicaciones mientras no hay conexión. Es anterior a la recuperación de amqplib y hace más trabajo por ti. Merece la pena si prefieres no mantener tu propio
#unconfirmed. Antes de confiarle una caída larga, revisa cómo limita esa cola. - Los brokers gestionados (Amazon MQ for RabbitMQ, CloudAMQP) te quitan de encima el clúster, las actualizaciones y el problema del hostname. No te quitan nada del trabajo del lado del cliente: todo lo anterior sigue valiendo, porque la conexión se puede seguir cayendo.
Y sobre los clústeres: un solo nodo basta hasta que el costo de "no se publica nada durante ocho segundos" supere el de mantener tres nodos. En un checkout, ese momento suele llegar pronto.
Puntos clave
-
Que
publish()devuelvatruesignifica que lo aceptó tu buffer, no el broker. En el ensayo, 467 pedidos se perdieron así sin un solo error. Los confirms son la única señal de que un mensaje ya es problema del broker y no tuyo, y en paralelo cuestan una parte del rendimiento (4.784–5.728 msg/s frente a 6.859–7.425). Esperados uno a uno, fueron 70–78. -
La recuperación de amqplib reconecta el socket y nada más. Reconstruye canales, topología y consumidores en
setup, reenvía lo que quedó sin confirmar y añade un listener de'error', o el proceso muere antes de que la recuperación pueda hacer nada. - En Node, el prefetch limita tu concurrencia. Réplicas × prefetch es el máximo de peticiones que mandas a la vez a los servicios siguientes. Ajústalo según la dependencia más lenta, no según la cola.
- Todos los mecanismos de fiabilidad de este artículo producen duplicados. 128 por reenvío y 53 por reentrega, todos absorbidos por una clave de idempotencia. Haz idempotentes los consumidores primero, y lo demás se puede añadir sin riesgo.
-
Provoca de verdad cada fallo que tu código dice manejar, en tu versión del broker, antes de confiar en él. El contador de reintentos de la Parte 1 leía
x-death, y su protección contra bucles se apoyaba enx-delivery-limit. En RabbitMQ 4.3 no se sostuvo ninguna de las dos cosas, y solo una prueba con el servicio de pagos caído lo mostró.
Apéndice: el resto del código
Todo lo que sigue hace falta para ejecutar los ejemplos y el ensayo, pero nada de esto es el patrón en sí.
src/billing.ts — las funciones de la Parte 1, sin cambios salvo el export
// src/billing.ts
import type { ConsumeMessage } from 'amqplib';
import type { OrderPlaced } from './order-publisher.ts';
const PAYMENTS_URL = process.env.PAYMENTS_URL ?? 'http://localhost:4001';
export const RETRY_COUNT_HEADER = 'retry-count';
/**
* Cuántas veces se ha aparcado ya este mensaje en la cola de reintentos. Lo cuenta el
* propio consumidor: RabbitMQ 4.x ignora el x-death que manda un cliente al republicar,
* así que x-death nunca pasa de 1 y un pedido que falla se reintentaría para siempre.
*/
export function retriesSoFar(message: ConsumeMessage): number {
const count: unknown = message.properties.headers?.[RETRY_COUNT_HEADER];
return typeof count === 'number' ? count : 0;
}
export function parseOrder(message: ConsumeMessage): OrderPlaced {
const parsed: unknown = JSON.parse(message.content.toString('utf8'));
if (
typeof parsed !== 'object' || parsed === null ||
typeof (parsed as OrderPlaced).orderId !== 'string' ||
typeof (parsed as OrderPlaced).totalCents !== 'number'
) {
throw new SyntaxError('order payload is missing orderId or totalCents');
}
return parsed as OrderPlaced;
}
export async function chargeCard(order: OrderPlaced): Promise<string> {
const response = await fetch(`${PAYMENTS_URL}/charges`, {
method: 'POST',
headers: {
'content-type': 'application/json',
// La entrega es al menos una vez, así que el mismo pedido puede llegar dos
// veces. Usar el id del pedido como clave de idempotencia hace que el segundo
// intento devuelva el primer cargo en vez de cobrarle otra vez al cliente.
'idempotency-key': order.orderId,
},
body: JSON.stringify({ amountCents: order.totalCents, reference: order.orderId }),
signal: AbortSignal.timeout(5_000),
});
if (!response.ok) {
throw new Error(`payments service returned ${response.status} for ${order.orderId}`);
}
const { chargeId } = (await response.json()) as { chargeId: string };
return chargeId;
}
src/payments-stub.ts — el sustituto del servicio de pagos
// src/payments-stub.ts
// Un sustituto del proveedor de pagos, para que el consumidor tenga a quién llamar.
// Respeta las claves de idempotencia como lo haría uno real, y cuenta las repeticiones.
import { randomUUID } from 'node:crypto';
import { createServer } from 'node:http';
import { setTimeout as sleep } from 'node:timers/promises';
const PORT = Number(process.env.PORT ?? 4001);
const LATENCY_MS = Number(process.env.PAYMENTS_LATENCY_MS ?? 50);
// El cobro pendiente se guarda antes de la espera, así que dos peticiones simultáneas con
// la misma clave comparten un cobro en lugar de no encontrarse en el mapa y crear dos.
const chargesByKey = new Map<string, Promise<string>>();
let replays = 0;
const server = createServer(async (request, response) => {
const reply = (status: number, body: unknown): void => {
response.writeHead(status, { 'content-type': 'application/json' });
response.end(JSON.stringify(body));
};
if (request.method === 'GET' && request.url === '/stats') {
reply(200, { charges: chargesByKey.size, replays });
return;
}
if (request.method !== 'POST' || request.url !== '/charges') {
reply(404, { error: 'not found' });
return;
}
for await (const _chunk of request); // el cuerpo no se usa, solo se consume
const key = request.headers['idempotency-key'];
if (typeof key !== 'string' || key.length === 0) {
reply(400, { error: 'idempotency-key header is required' });
return;
}
const existing = chargesByKey.get(key);
if (existing) {
replays++;
console.log(`[payments] replayed ${key}`);
reply(200, { chargeId: await existing });
return;
}
const charge = sleep(LATENCY_MS).then(() => `ch_${randomUUID()}`);
chargesByKey.set(key, charge);
reply(201, { chargeId: await charge });
});
server.listen(PORT, () => console.log(`[payments] listening on ${PORT}, ${LATENCY_MS} ms per charge`));
src/drill.ts — el ensayo que mata el broker
// src/drill.ts
// Publica TOTAL pedidos, mata el broker después de KILL_AFTER, lo reinicia y cuenta lo
// que queda en orders.billing. Nadie consume la cola durante la corrida.
//
// node src/drill.ts <fire-and-forget|confirms|recovery> [total] [killAfter]
import amqp from 'amqplib';
import { execFile } from 'node:child_process';
import { once } from 'node:events';
import { setTimeout as sleep } from 'node:timers/promises';
import { promisify } from 'node:util';
import { RABBIT_URL } from './connection.ts';
import { OrderPublisher } from './order-publisher.ts';
import type { OrderPlaced } from './order-publisher.ts';
import {
assertTopology, BILLING_DEAD_QUEUE, BILLING_QUEUE, BILLING_RETRY_QUEUE, ORDERS_EXCHANGE,
} from './topology.ts';
const SCENARIOS = ['fire-and-forget', 'confirms', 'recovery'] as const;
type Scenario = (typeof SCENARIOS)[number];
const scenario = process.argv[2] as Scenario;
if (!SCENARIOS.includes(scenario)) {
console.error(`usage: node src/drill.ts <${SCENARIOS.join('|')}> [total] [killAfter]`);
process.exit(2);
}
const TOTAL = Number(process.argv[3] ?? 20_000);
const KILL_AFTER = Number(process.argv[4] ?? TOTAL / 2);
const CONTAINER = 'aurora-rabbitmq';
const ROUTING_KEY = 'order.us.placed';
const SENTINEL = 'drill-sentinel';
// Unos 10.000 pedidos por segundo: un checkout con mucho tráfico, no un bucle de benchmark.
const BATCH = 100;
const BATCH_PAUSE_MS = 10;
const docker = promisify(execFile);
function orderNumber(i: number): OrderPlaced {
return { orderId: `ord-${String(i).padStart(6, '0')}`, sku: 'BEAN-ETH-250', quantity: 1, totalCents: 1450 };
}
async function waitForBroker(): Promise<void> {
for (;;) {
try {
const probe = await amqp.connect(RABBIT_URL);
await probe.close();
return;
} catch {
await sleep(250);
}
}
}
let killedAt = 0;
let backAt = 0;
async function killAndRestartBroker(): Promise<void> {
killedAt = performance.now();
await docker('docker', ['kill', CONTAINER]);
await docker('docker', ['start', CONTAINER]);
await waitForBroker();
backAt = performance.now();
}
async function resetQueues(): Promise<void> {
const connection = await amqp.connect(RABBIT_URL);
const channel = await connection.createChannel();
await assertTopology(channel);
for (const queue of [BILLING_QUEUE, BILLING_RETRY_QUEUE, BILLING_DEAD_QUEUE]) await channel.purgeQueue(queue);
await connection.close();
}
/** Vacía orders.billing e informa cuántos mensajes, y cuántos pedidos distintos, contenía. */
async function countQueue(): Promise<{ total: number; unique: number }> {
// Una cola quorum recién reiniciada elige líder y reproduce su log antes de informar su
// profundidad real, y contar antes de tiempo da 0. Un write confirmado prueba que ya hay
// líder, así que primero publicamos un centinela y luego esperamos una profundidad estable.
const connection = await amqp.connect(RABBIT_URL);
const confirms = await connection.createConfirmChannel();
confirms.sendToQueue(BILLING_QUEUE, Buffer.from('{}'), { persistent: true, messageId: SENTINEL });
await confirms.waitForConfirms();
await confirms.close();
const channel = await connection.createChannel();
let depth = -1;
for (;;) {
await sleep(1_000);
const { messageCount } = await channel.checkQueue(BILLING_QUEUE);
if (messageCount === depth) break;
depth = messageCount;
}
const seen = new Set<string>();
let total = 0;
let lastMessageAt = performance.now();
await channel.consume(BILLING_QUEUE, (message) => {
if (message === null) return;
lastMessageAt = performance.now();
if (message.properties.messageId === SENTINEL) return;
total++;
seen.add(String(message.properties.messageId));
}, { noAck: true });
// Terminamos cuando no llega nada en dos segundos y el broker confirma que la cola está vacía.
for (;;) {
await sleep(500);
if (performance.now() - lastMessageAt < 2_000) continue;
const { messageCount } = await channel.checkQueue(BILLING_QUEUE);
if (messageCount === 0) break;
}
await connection.close();
return { total, unique: seen.size };
}
const pace = async (i: number): Promise<void> => {
if (i % BATCH === BATCH - 1) await sleep(BATCH_PAUSE_MS);
};
/** El publisher de la Parte 1: que publish() devuelva true es la única señal que recibe. */
async function fireAndForget(): Promise<Record<string, number>> {
const connection = await amqp.connect(RABBIT_URL);
connection.on('error', () => undefined); // mantiene vivo el ensayo para contar; el código de la Parte 1 se cae aquí
const channel = await connection.createChannel();
channel.on('error', () => undefined);
let accepted = 0;
let threw = 0;
let restart: Promise<void> | undefined;
for (let i = 0; i < TOTAL; i++) {
if (i === KILL_AFTER) restart = killAndRestartBroker();
const order = orderNumber(i);
try {
const writable = channel.publish(ORDERS_EXCHANGE, ROUTING_KEY, Buffer.from(JSON.stringify(order)), {
persistent: true, contentType: 'application/json', messageId: order.orderId,
});
accepted++;
if (!writable) await Promise.race([once(channel, 'drain'), once(channel, 'close')]);
} catch {
threw++;
}
await pace(i);
}
await restart;
return { 'publish() returned normally': accepted, 'publish() threw': threw };
}
/** Un confirm channel, pero nada reenvía lo que quedó sin confirmar. */
async function confirmsOnly(): Promise<Record<string, number>> {
const connection = await amqp.connect(RABBIT_URL);
connection.on('error', () => undefined);
const channel = await connection.createConfirmChannel();
channel.on('error', () => undefined);
let confirmed = 0;
let failed = 0;
let threw = 0;
let restart: Promise<void> | undefined;
for (let i = 0; i < TOTAL; i++) {
if (i === KILL_AFTER) restart = killAndRestartBroker();
const order = orderNumber(i);
try {
channel.publish(
ORDERS_EXCHANGE, ROUTING_KEY, Buffer.from(JSON.stringify(order)),
{ persistent: true, contentType: 'application/json', messageId: order.orderId },
(error) => (error ? failed++ : confirmed++),
);
} catch {
threw++;
}
await pace(i);
}
await restart;
return { 'confirmed by the broker': confirmed, 'confirm failed (channel closed)': failed, 'publish() threw': threw };
}
/** OrderPublisher: confirms, recuperación de la conexión y reenvío de todo lo no confirmado. */
async function withRecovery(): Promise<Record<string, number>> {
const publisher = await OrderPublisher.start();
const results: Array<Promise<boolean>> = [];
let restart: Promise<void> | undefined;
for (let i = 0; i < TOTAL; i++) {
if (i === KILL_AFTER) restart = killAndRestartBroker();
results.push(publisher.publish(ROUTING_KEY, orderNumber(i)).then(() => true, () => false));
await pace(i);
}
const outcomes = await Promise.all(results);
const allConfirmedAt = performance.now();
await restart;
await publisher.close();
return {
'confirmed by the broker': outcomes.filter(Boolean).length,
'rejected': outcomes.filter((ok) => !ok).length,
'seconds from kill to last confirm': Math.round((allConfirmedAt - killedAt) / 100) / 10,
};
}
await resetQueues();
const started = performance.now();
const counters =
scenario === 'fire-and-forget' ? await fireAndForget()
: scenario === 'confirms' ? await confirmsOnly()
: await withRecovery();
const { total, unique } = await countQueue();
console.log(`\nscenario: ${scenario}`);
console.log(`orders published: ${TOTAL}, broker killed after #${KILL_AFTER}`);
console.log(`broker down for ${((backAt - killedAt) / 1000).toFixed(1)} s, run took ${((performance.now() - started) / 1000).toFixed(1)} s`);
for (const [label, value] of Object.entries(counters)) console.log(` ${label.padEnd(36)} ${value}`);
console.log(` ${'messages in orders.billing'.padEnd(36)} ${total}`);
console.log(` ${'distinct orders in orders.billing'.padEnd(36)} ${unique}`);
console.log(` ${'orders lost'.padEnd(36)} ${TOTAL - unique}`);
console.log(` ${'duplicates'.padEnd(36)} ${total - unique}`);
src/pause-probe.ts y src/deadline-probe.ts — las dos pruebas
// src/pause-probe.ts — congela el broker (sin RST, como una red rota) y mide cuánto tarda
// el publisher en notarlo y en entregar un pedido cuando el broker vuelve.
import { execFile } from 'node:child_process';
import { setTimeout as sleep } from 'node:timers/promises';
import { promisify } from 'node:util';
import { OrderPublisher } from './order-publisher.ts';
const docker = promisify(execFile);
const publisher = await OrderPublisher.start();
const started = performance.now();
const elapsed = (): string => `${((performance.now() - started) / 1000).toFixed(1)} s`;
const log = console.warn;
console.warn = (...args: unknown[]) => log(`[+${elapsed()}]`, ...args);
await docker('docker', ['pause', 'aurora-rabbitmq']);
console.warn('[probe] broker frozen with docker pause');
const confirmed = publisher.publish('order.us.placed', {
orderId: `ord-probe-${Date.now()}`, sku: 'BEAN-ETH-250', quantity: 1, totalCents: 1450,
});
await sleep(60_000);
await docker('docker', ['unpause', 'aurora-rabbitmq']);
console.warn('[probe] broker unfrozen');
await confirmed;
console.warn('[probe] order confirmed');
await publisher.close();
// src/deadline-probe.ts — qué significa para un pedido el plazo de quien llama, en los dos casos.
import amqp from 'amqplib';
import { execFile } from 'node:child_process';
import { setTimeout as sleep } from 'node:timers/promises';
import { promisify } from 'node:util';
import { RABBIT_URL } from './connection.ts';
import { OrderPublisher } from './order-publisher.ts';
import { BILLING_QUEUE } from './topology.ts';
const docker = promisify(execFile);
const order = (id: string) => ({ orderId: id, sku: 'BEAN-ETH-250', quantity: 1, totalCents: 1450 });
const outcome = (promise: Promise<void>): Promise<string> =>
promise.then(() => 'confirmed', (error: Error) => `${error.name}: ${error.message}`);
async function idsInQueue(): Promise<string[]> {
const connection = await amqp.connect(RABBIT_URL);
const channel = await connection.createChannel();
const ids: string[] = [];
for (;;) {
const message = await channel.get(BILLING_QUEUE, { noAck: true });
if (message === false) break;
ids.push(String(message.properties.messageId));
}
await connection.close();
return ids;
}
const publisher = await OrderPublisher.start();
// 1. Broker detenido, sin canal: el pedido nunca sale del proceso.
await docker('docker', ['stop', 'aurora-rabbitmq']);
await sleep(1_000);
console.log('broker stopped ->', await outcome(publisher.publish('order.us.placed', order('ord-deadline-unsent'), { signal: AbortSignal.timeout(2_000) })));
await docker('docker', ['start', 'aurora-rabbitmq']);
// 2. Broker congelado, canal vivo: el pedido se escribe en el socket y el confirm nunca llega.
await sleep(20_000);
await docker('docker', ['pause', 'aurora-rabbitmq']);
console.log('broker frozen ->', await outcome(publisher.publish('order.us.placed', order('ord-deadline-sent'), { signal: AbortSignal.timeout(2_000) })));
await docker('docker', ['unpause', 'aurora-rabbitmq']);
await publisher.close();
await sleep(2_000);
console.log('in orders.billing ->', (await idsInQueue()).join(', ') || '(empty)');
src/bench-confirms.ts y src/measure-drain.ts — las dos mediciones
// src/bench-confirms.ts — lo que cuestan los confirms en rendimiento, de tres formas. Nadie consume la cola.
import amqp from 'amqplib';
import { once } from 'node:events';
import { RABBIT_URL } from './connection.ts';
import { OrderPublisher } from './order-publisher.ts';
import { assertTopology, BILLING_QUEUE, ORDERS_EXCHANGE } from './topology.ts';
const COUNT = Number(process.argv[2] ?? 20_000);
const KEY = 'order.us.placed';
const body = (i: number): Buffer =>
Buffer.from(JSON.stringify({ orderId: `ord-${i}`, sku: 'BEAN-ETH-250', quantity: 1, totalCents: 1450 }));
async function purge(): Promise<void> {
const connection = await amqp.connect(RABBIT_URL);
const channel = await connection.createChannel();
await assertTopology(channel);
await channel.purgeQueue(BILLING_QUEUE);
await connection.close();
}
async function timed(label: string, run: () => Promise<void>): Promise<void> {
await purge();
const started = performance.now();
await run();
const seconds = (performance.now() - started) / 1000;
console.log(`${label.padEnd(40)} ${String(Math.round(COUNT / seconds)).padStart(6)} msg/s`);
}
await timed('no confirms (fire-and-forget)', async () => {
const connection = await amqp.connect(RABBIT_URL);
const channel = await connection.createChannel();
for (let i = 0; i < COUNT; i++) {
if (!channel.publish(ORDERS_EXCHANGE, KEY, body(i), { persistent: true })) await once(channel, 'drain');
}
await connection.close();
});
await timed('confirms, awaiting each one', async () => {
const connection = await amqp.connect(RABBIT_URL);
const channel = await connection.createConfirmChannel();
for (let i = 0; i < COUNT; i++) {
await new Promise<void>((resolve, reject) =>
channel.publish(ORDERS_EXCHANGE, KEY, body(i), { persistent: true }, (error) => (error ? reject(error) : resolve())));
}
await connection.close();
});
await timed('confirms, 1,000 in flight (OrderPublisher)', async () => {
const publisher = await OrderPublisher.start();
await Promise.all(Array.from({ length: COUNT }, (_, i) =>
publisher.publish(KEY, { orderId: `ord-${i}`, sku: 'BEAN-ETH-250', quantity: 1, totalCents: 1450 })));
await publisher.close();
});
await purge();
// src/measure-drain.ts — publica N pedidos y mide cuánto tarda facturación en cobrarlos todos.
import { randomUUID } from 'node:crypto';
import { setTimeout as sleep } from 'node:timers/promises';
import { OrderPublisher } from './order-publisher.ts';
const COUNT = Number(process.argv[2] ?? 3_000);
const stats = async (): Promise<{ charges: number; replays: number }> =>
(await fetch('http://localhost:4001/stats')).json() as Promise<{ charges: number; replays: number }>;
const before = await stats();
const publisher = await OrderPublisher.start();
const started = performance.now();
await Promise.all(Array.from({ length: COUNT }, () =>
publisher.publish('order.us.placed', { orderId: `ord-${randomUUID()}`, sku: 'BEAN-ETH-250', quantity: 1, totalCents: 1450 })));
await publisher.close();
let now = await stats();
while (now.charges - before.charges < COUNT) { await sleep(100); now = await stats(); }
const seconds = (performance.now() - started) / 1000;
console.log(`prefetch ${process.env.BILLING_PREFETCH}: ${COUNT} orders charged in ${seconds.toFixed(1)} s (${Math.round(COUNT / seconds)} orders/s)`);
Con esto se cierra la serie. La Parte 1 llevaba un mensaje a la cola correcta, y esta lo mantuvo ahí durante la caída del broker, una reconexión, un escalado y un despliegue gradual, con cifras para cada caso. Si te llevas un solo hábito de las dos partes, que sea el ensayo: tumba el sistema a propósito y cuenta lo que vuelve. Casi siempre, la sorpresa está en ese conteo.
Top comments (0)