* notes / bullmq

BullMQ y las colas: cuando un trabajo no debe vivir en tu request handler

— Reflexión técnica sobre el diseño de sistemas asíncronos con BullMQ: ciclo de vida de los trabajos, reintentos exponenciales, scheduling, observabilidad y un episodio doloroso de concurrencia descontrolada.

September 12, 20237 min read#bullmq#redis#nodejs#colas

Contexto: por qué existen las colas

Cualquier operación que un manejador HTTP no pueda completar en, digamos, doscientos milisegundos (generar una factura PDF, transcodificar un vídeo, sincronizar con un CRM externo, enviar un email, ejecutar un algoritmo de recomendación) pertenece a una cola de fondo. Las razones son tres: la latencia percibida por el usuario debe permanecer baja; los fallos transitorios del sistema externo no deben propagarse como errores HTTP 500; y la carga debe poder amortiguarse en picos de tráfico sin que el servidor web se desplome.

BullMQ es la evolución moderna de Bull, escrita en TypeScript, con un modelo de actores claramente separados: Queue produce trabajos, Worker los consume, QueueEvents emite señales del ciclo de vida, y FlowProducer orquesta flujos jerárquicos. Todo se apoya sobre Redis, que actúa como almacén persistente: los trabajos pendientes viven en una lista, los activos en otra estructura, los completados y fallidos en sets ordenados con timestamps, y los retrasados en un sorted set que el worker consulta periódicamente.

Cabe destacar que, frente a alternativas como Sidekiq (Ruby, también Redis), RabbitMQ (broker dedicado AMQP) o AWS SQS (servicio gestionado), BullMQ ocupa un nicho concreto: ecosistema Node, Redis ya disponible, y necesidad de primitivas de scheduling y retries sin desplegar infraestructura adicional. No es la mejor opción para topologías pub/sub complejas, pero para "ejecuta esto fuera del request" es difícil de batir.

Producerqueue.add()Redislistas + sorted setswaitingdelayedactivecompletedfailedWorker Aconcurrency: 4Worker Bconcurrency: 4Worker Cconcurrency: 2

Implementación: cola mínima, worker con backoff, scheduling

Empezamos por lo esencial. Una cola y un worker en TypeScript, con conexión compartida y reintentos exponenciales:

import { Queue, Worker, type ConnectionOptions } from "bullmq";
 
const connection: ConnectionOptions = {
  host: process.env.REDIS_HOST ?? "localhost",
  port: 6379,
};
 
export const emailsQueue = new Queue("emails", { connection });
 
type EmailJob = {
  to: string;
  template: "welcome" | "receipt" | "reset-password";
  payload: Record<string, unknown>;
};
 
export async function enqueueEmail(data: EmailJob) {
  await emailsQueue.add("send", data, {
    attempts: 5,
    backoff: { type: "exponential", delay: 2_000 },
    removeOnComplete: { age: 3600, count: 1000 },
    removeOnFail: { age: 24 * 3600 },
  });
}

Los reintentos exponenciales merecen comentario. La opción backoff: { type: "exponential", delay: 2000 } produce intentos separados por 2s, 4s, 8s, 16s y 32s. En la práctica, este patrón es óptimo para errores transitorios de red o rate limits suaves: das tiempo al sistema externo a recuperarse sin tirarte cinco minutos clavado en el primer fallo. Para fallos deterministas (un payload inválido, una clave que no existe) conviene lanzar un UnrecoverableError para que BullMQ no insista:

import { Worker, UnrecoverableError } from "bullmq";
import { sendWithSendgrid } from "./sendgrid";
 
const worker = new Worker<EmailJob>(
  "emails",
  async (job) => {
    if (!job.data.to.includes("@")) {
      throw new UnrecoverableError("destinatario inválido");
    }
 
    const result = await sendWithSendgrid(job.data);
    return { messageId: result.id };
  },
  {
    connection,
    concurrency: 8,
    limiter: { max: 50, duration: 1000 },
  },
);
 
worker.on("completed", (job, result) => {
  console.log(`[email] ${job.id} -> ${result.messageId}`);
});
 
worker.on("failed", (job, err) => {
  console.error(`[email] ${job?.id} falló: ${err.message}`);
});

Dos parámetros aquí son críticos: concurrency: 8 significa que este worker procesa hasta ocho trabajos en paralelo (cada uno una promesa, no un thread), y limiter: { max: 50, duration: 1000 } impone un techo global de cincuenta trabajos por segundo entre todos los workers conectados a esa cola. Este último es, retrospectivamente, lo que me habría salvado del incidente de SendGrid.

Para trabajos retrasados, BullMQ acepta una opción delay en milisegundos:

await emailsQueue.add(
  "reminder",
  { to: user.email, template: "receipt", payload: { invoice: id } },
  { delay: 24 * 60 * 60 * 1000 },
);

Y para trabajos repetibles, la API moderna sustituye al antiguo QueueScheduler (que era un componente independiente en BullMQ v1 y desapareció en v3+, dado que los workers asumieron sus responsabilidades) por upsertJobScheduler:

await emailsQueue.upsertJobScheduler(
  "daily-digest",
  { pattern: "0 9 * * *", tz: "Europe/Madrid" },
  {
    name: "digest",
    data: { kind: "daily" },
    opts: { attempts: 3 },
  },
);

Para observabilidad, bull-board ofrece una interfaz web autoalojable que muestra el estado de cada cola, permite reintentar trabajos manualmente, inspeccionar payloads y purgar colas. Integrarlo en una app Express ocupa siete líneas:

import { createBullBoard } from "@bull-board/api";
import { BullMQAdapter } from "@bull-board/api/bullMQAdapter";
import { ExpressAdapter } from "@bull-board/express";
 
const serverAdapter = new ExpressAdapter();
serverAdapter.setBasePath("/admin/queues");
 
createBullBoard({
  queues: [new BullMQAdapter(emailsQueue)],
  serverAdapter,
});
 
app.use("/admin/queues", serverAdapter.getRouter());

Hallazgos: tres patrones, tres antipatrones

He acumulado, a base de tropezar, una lista corta de heurísticas que aplico por defecto.

Patrón 1: idempotencia obligatoria. Todo process debe ser idempotente. BullMQ garantiza ejecución at least once, no exactly once: un worker que muere a mitad del proceso provoca que el trabajo se reencole. Usar un jobId determinista y verificar dentro del worker si ya se hizo la acción evita duplicados desagradables:

emailsQueue.add("send", data, { jobId: `welcome:${userId}` })

Patrón 2: jobs pequeños, payloads grandes en almacén externo. El payload del trabajo viaja por Redis y se serializa como JSON. Si necesitas pasar un fichero de cincuenta megabytes al worker, súbelo a S3 y pasa solamente la URL. No solo es más eficiente; también facilita reintentos.

Patrón 3: separación lógica de colas por SLO. Mezclar en la misma cola trabajos críticos (envío de OTP, ventana de tres segundos) con trabajos diferibles (informes semanales) es receta para incidentes. Crea colas distintas y workers distintos, dimensionados según prioridad.

Antipatrón 1: concurrencia sin límite. Justamente el error que pagué con SendGrid. concurrency por sí solo no basta si tienes diez réplicas de worker; necesitas limiter global o, mejor aún, rate limiting en el servicio destino.

Antipatrón 2: capturar excepciones dentro del processor. Si tu processor envuelve todo en try/catch y devuelve normalmente, BullMQ considera el trabajo completado. Los reintentos solo se disparan ante excepciones propagadas. No silencies errores que querrías reintentar.

Antipatrón 3: usar la cola como base de datos. He visto sistemas donde los trabajos completados se conservan eternamente para consultarlos. Mal. Redis no es para eso: usa removeOnComplete con TTL razonable y persiste el resultado en tu base de datos transaccional si lo necesitas.

Closure

Las colas son, posiblemente, la pieza menos glamurosa de la arquitectura de un servicio moderno, pero también una de las que más sufrimiento ahorran a medio plazo. BullMQ encaja con elegancia en el mundo Node, hereda la robustez de Redis y se mantiene activamente. Por consiguiente, cada vez que veo una nueva aplicación que ejecuta tareas pesadas dentro del request handler, sé que estoy mirando el incidente futuro de alguien. Asimismo, el incidente del que abrí este artículo no fue culpa de BullMQ, sino mía por configurarlo mal. Las herramientas no eximen del diseño cuidadoso; lo facilitan, eso sí, considerablemente.