← Volver al listado de tecnologías ← Índice de NestJS: Buenas Prácticas

Async, colas y eventos

Por: Artiko
nestjsbuenas-practicasbullmqeventosmicroservicioscolasasync

Async, colas y eventos

Hasta acá el request entra, hace su trabajo y responde. Funciona mientras ese trabajo sea barato. En cuanto el handler tiene que redimensionar una imagen, llamar a un proveedor externo lento, mandar mil mails o recalcular un reporte, el modelo request-response se rompe: el cliente espera segundos, el connection pool se llena de requests colgados esperando I/O, y un pico de tráfico tumba el proceso.

Este capítulo trata de una sola idea con muchas caras: el request path solo debería hacer lo que el cliente necesita para recibir su respuesta; todo lo demás se saca afuera. “Afuera” puede ser un evento en memoria, una cola en Redis, un microservicio o un job programado. Vamos a ver cuándo alcanza cada herramienta y cuánto cuesta cada salto.

Por qué sacar trabajo del request path

La latencia que percibe el usuario es el tiempo hasta que recibe la respuesta, no el tiempo hasta que todo terminó. Si al crear un pedido tenés que:

  1. Validar y persistir el pedido (30 ms, el cliente lo necesita).
  2. Mandar el mail de confirmación (400 ms, llamada SMTP externa).
  3. Notificar al sistema de facturación (200 ms).
  4. Actualizar métricas de analytics (150 ms).

…hacer los cuatro pasos síncronos le cobra al cliente 780 ms para algo que a él solo le importa el paso 1. Peor: si el SMTP está caído, el POST /orders falla y el pedido no se crea, aunque el pedido en sí era válido. Acoplaste la creación del pedido a la disponibilidad del mail.

Sacando 2, 3 y 4 del request path, el cliente recibe su respuesta en ~30 ms y los efectos secundarios ocurren de forma asíncrona, con reintentos propios y sin poder tumbar la operación principal.

flowchart LR
  C[Cliente] -->|POST /orders| H[Handler]
  H -->|30 ms: persistir| DB[(DB)]
  H -->|201 Created| C
  H -.->|emitir OrderCreated| Q[Cola / Bus de eventos]
  Q --> W1[Enviar mail]
  Q --> W2[Notificar facturación]
  Q --> W3[Actualizar analytics]

La regla operativa: si el resultado de una tarea no cambia la respuesta HTTP, no debería bloquear la respuesta HTTP.

Colas con BullMQ

@nestjs/bullmq es la integración oficial sobre BullMQ, que usa Redis como backend. A diferencia de un evento en memoria, un job en una cola sobrevive a un reinicio del proceso, se puede reintentar con política propia, se distribuye entre varios workers y te da visibilidad (jobs pendientes, activos, fallidos).

Setup: conexión y registro de cola

// app.module.ts
import { BullModule } from '@nestjs/bullmq';

@Module({
  imports: [
    BullModule.forRoot({
      connection: { host: process.env.REDIS_HOST, port: 6379 },
    }),
    BullModule.registerQueue({
      name: 'emails',
      defaultJobOptions: {
        attempts: 3,
        backoff: { type: 'exponential', delay: 2000 },
        removeOnComplete: 1000,   // conservá los últimos 1000 completados
        removeOnFail: 5000,       // y los últimos 5000 fallidos, para inspección
      },
    }),
  ],
})
export class AppModule {}

defaultJobOptions es donde vive la política de robustez: attempts (cuántas veces reintentar antes de darlo por fallido) y backoff (cuánto esperar entre intentos). Con backoff exponential y delay: 2000, los reintentos ocurren a ~2 s, ~4 s, ~8 s: le das tiempo al servicio externo a recuperarse en vez de martillarlo.

Si no ponés removeOnComplete/removeOnFail, los jobs completados se acumulan en Redis para siempre: es una fuga de memoria lenta que nadie nota hasta que Redis se queda sin RAM.

Productor: encolar sin esperar el resultado

import { InjectQueue } from '@nestjs/bullmq';
import { Queue } from 'bullmq';

@Injectable()
export class OrdersService {
  constructor(@InjectQueue('emails') private readonly emailsQueue: Queue) {}

  async create(dto: CreateOrderDto) {
    const order = await this.repo.save(dto); // esto sí bloquea la respuesta
    // esto no: solo encola y sigue
    await this.emailsQueue.add('order-confirmation', { orderId: order.id });
    return order;
  }
}

queue.add(nombre, data) escribe el job en Redis y vuelve enseguida. La data debe ser serializable (se guarda como JSON): pasá IDs, no entidades ni instancias de clase. Un antipatrón común es meter el objeto entero con relaciones cargadas; pasá orderId y que el worker vuelva a leer lo que necesite.

Consumidor: Processor + WorkerHost

El consumidor extiende WorkerHost e implementa process. En BullMQ moderno (y @nestjs/bullmq) el decorador @Process() por nombre está deprecado: se usa un único process y se ramifica por job.name.

import { Processor, WorkerHost, OnWorkerEvent } from '@nestjs/bullmq';
import { Job } from 'bullmq';

@Processor('emails', { concurrency: 5, limiter: { max: 10, duration: 1000 } })
export class EmailsConsumer extends WorkerHost {
  private readonly logger = new Logger(EmailsConsumer.name);

  async process(job: Job): Promise<void> {
    switch (job.name) {
      case 'order-confirmation':
        await this.sendConfirmation(job.data.orderId);
        break;
      default:
        throw new Error(`Job desconocido: ${job.name}`);
    }
  }

  @OnWorkerEvent('failed')
  onFailed(job: Job, err: Error) {
    this.logger.warn(`Job ${job.id} falló (intento ${job.attemptsMade}): ${err.message}`);
  }
}

Dos parámetros del @Processor que definen el comportamiento bajo carga:

Reintentos y backoff en detalle

Cuando process lanza una excepción, BullMQ reintenta según attempts/backoff. Un detalle clave: no todos los errores merecen reintento. Reintentar un 404 el recurso no existe o un 422 payload inválido es tirar recursos; esos fallos son permanentes. Reintentar un 503 servicio no disponible o un timeout sí tiene sentido.

async process(job: Job) {
  try {
    await this.callExternalApi(job.data);
  } catch (err) {
    if (err instanceof PermanentError) {
      // no reintentar: descartá los intentos restantes
      await job.discard();
      throw err;
    }
    throw err; // error transitorio: dejá que BullMQ reintente
  }
}

Idempotencia de jobs

Un job puede ejecutarse más de una vez: reintento tras un fallo parcial, o el worker que se cae después de hacer el trabajo pero antes de marcar el job como completado. Si tu handler no es idempotente, “enviar mail” se convierte en “enviar tres mails”.

Dos capas de defensa:

  1. Deduplicación al encolar con jobId. BullMQ ignora un add si ya existe un job con ese mismo jobId en la cola. Sirve para que dos requests concurrentes no encolen el mismo trabajo dos veces.

    await this.emailsQueue.add(
  'order-confirmation',
  { orderId: order.id },
  { jobId: `confirmation:${order.id}` },
);

  2. Idempotencia en el handler: antes de ejecutar el efecto, verificá si ya se hizo. Guardá una marca (email_sent_at, o una fila en una tabla de idempotencia con clave única) y salí temprano si ya existe.

    async sendConfirmation(orderId: string) {
  const order = await this.repo.findById(orderId);
  if (order.confirmationSentAt) return; // ya se envió, no repetir
  await this.mailer.send(order.email, ...);
  await this.repo.markConfirmationSent(orderId);
}

La regla: diseñá el handler para que ejecutarlo N veces tenga el mismo efecto que ejecutarlo una vez.

Jobs recurrentes (cron) con la cola

Para trabajo periódico distribuido, BullMQ ofrece repeatable jobs: la cola encola automáticamente un job según un patrón cron. A diferencia de @nestjs/schedule (que veremos abajo), esto corre una sola vez aunque tengas varias instancias, porque el scheduling vive en Redis, no en cada proceso.

await this.reportsQueue.add(
  'daily-summary',
  {},
  { repeat: { pattern: '0 0 * * *' } }, // todos los días a medianoche
);

Dead-letter: qué hacer con lo que falló definitivamente

BullMQ no tiene una dead-letter queue nativa como RabbitMQ, pero la construís vos: cuando un job agota sus attempts, queda en el set failed. Podés detectar el fallo final en @OnWorkerEvent('failed') (cuando job.attemptsMade >= job.opts.attempts) y moverlo a una cola aparte para inspección o reprocesamiento manual, en vez de perderlo.

@OnWorkerEvent('failed')
async onFailed(job: Job, err: Error) {
  if (job.attemptsMade >= (job.opts.attempts ?? 1)) {
    await this.deadLetterQueue.add('dead', { original: job.name, data: job.data, error: err.message });
  }
}

Flujo completo: producer → queue → worker con reintentos

flowchart TD
  P[Productor: queue.add] --> R[(Redis: cola)]
  R --> W{Worker: process}
  W -->|éxito| Done[completed<br/>removeOnComplete]
  W -->|error transitorio| Retry{attemptsMade < attempts?}
  Retry -->|sí| Backoff[esperar backoff exponencial] --> R
  Retry -->|no| Failed[failed]
  Failed --> DLQ[(Dead-letter queue)]
  W -->|error permanente<br/>job.discard| Failed

¿Cuándo una cola?

Meté una cola cuando se cumpla al menos una de estas condiciones:

No metas una cola si el trabajo es barato y en memoria: agregás Redis, latencia de red y una pieza más que puede fallar, para nada.

EventEmitter: desacoplar dentro del proceso

@nestjs/event-emitter (sobre eventemitter2) resuelve un problema distinto: desacoplar módulos dentro del mismo proceso sin que uno importe al otro. El módulo de pedidos no debería conocer al de mails, ni al de analytics, ni al de facturación. En vez de llamarlos, emite un evento de dominio y quien quiera reacciona.

// app.module.ts
import { EventEmitterModule } from '@nestjs/event-emitter';

@Module({ imports: [EventEmitterModule.forRoot()] })
export class AppModule {}

Emitir un evento de dominio

export class OrderCreatedEvent {
  constructor(public readonly orderId: string, public readonly total: number) {}
}

@Injectable()
export class OrdersService {
  constructor(private readonly events: EventEmitter2) {}

  async create(dto: CreateOrderDto) {
    const order = await this.repo.save(dto);
    this.events.emit('order.created', new OrderCreatedEvent(order.id, order.total));
    return order;
  }
}

OrdersService no sabe quién escucha. Podés agregar un cuarto listener mañana sin tocar esta clase: es el principio abierto/cerrado del capítulo de SOLID aplicado al flujo de datos.

Escuchar con @OnEvent

@Injectable()
export class BillingListener {
  @OnEvent('order.created')
  handle(event: OrderCreatedEvent) {
    // notificar a facturación
  }
}

@Injectable()
export class AnalyticsListener {
  @OnEvent('order.created', { async: true }) // no bloquea al emisor
  async handle(event: OrderCreatedEvent) {
    await this.track(event);
  }
}

La trampa: emit es síncrono por defecto

Este es el malentendido más peligroso del EventEmitter. Por defecto, emit() ejecuta los listeners de forma síncrona, en el mismo tick y en el mismo request. Si un listener tarda 400 ms o lanza una excepción, eso sí afecta al emisor. El desacople es de código (quién conoce a quién), no de ejecución.

EventEmitter vs cola vs broker: cuándo cada uno

flowchart TD
  Start[Necesito reaccionar a algo] --> Q1{¿Cruza procesos<br/>o servicios?}
  Q1 -->|Sí| Broker[Broker/microservicio<br/>Redis · NATS · Kafka]
  Q1 -->|No, mismo proceso| Q2{¿Debe sobrevivir<br/>a un reinicio o<br/>reintentarse?}
  Q2 -->|Sí| Queue[BullMQ]
  Q2 -->|No, best-effort<br/>en memoria| EE[EventEmitter]

Microservicios y transporte

Cuando el trabajo vive en otro servicio, @nestjs/microservices abstrae el transporte detrás de una interfaz común. Hay dos estilos de mensajería, y confundirlos es un error de diseño frecuente:

// Servicio que responde (request-response)
@MessagePattern({ cmd: 'get_stock' })
getStock(@Payload() data: { sku: string }): Promise<number> {
  return this.inventory.stockOf(data.sku);
}

// Servicio que reacciona a un hecho (event-based, no responde)
@EventPattern('order.created')
async onOrderCreated(@Payload() data: OrderCreatedEvent) {
  await this.reserveStock(data);
}
// Cliente
const stock = await firstValueFrom(this.client.send({ cmd: 'get_stock' }, { sku }));
this.client.emit('order.created', event); // fire-and-forget

Transportes disponibles

Nest soporta varios transporters con la misma API; elegís según la topología:

TransporteEstilo naturalCuándo
Redis (Pub/Sub)event-basedYa tenés Redis; eventos simples entre pocos servicios.
NATSambosMensajería liviana, alto throughput, patrones de subject.
Kafkaevent-based (log)Streaming, retención y replay de eventos, alto volumen.
RabbitMQambosColas robustas, routing complejo, DLQ nativa, ACKs.
gRPCrequest-responseContratos tipados (protobuf), baja latencia entre servicios.
TCPambosDefault simple, sin dependencias externas.

Cuándo pasar a microservicios (y el costo)

Pasar a microservicios no es una mejora automática: cambiás una llamada de función en memoria (nanosegundos, transaccional, con stack trace completo) por una llamada de red (milisegundos, que puede fallar, sin transacción distribuida gratis). Los costos concretos:

Justifican el salto: escalar partes del sistema de forma independiente, aislar fallos, equipos distintos con ciclos de release propios, o límites de dominio ya muy marcados. Un monolito modular bien hecho con colas y eventos internos resuelve la mayoría de los casos sin pagar esa cuenta.

Patrones de mensajería (a grandes rasgos)

Sobre estos transportes se construyen patrones que merecen cursos propios; acá te doy el mapa para que sepas cuándo buscar cuál.

sequenceDiagram
  participant S as Servicio
  participant DB as DB (misma tx)
  participant R as Relay
  participant B as Broker
  S->>DB: guardar entidad + fila en outbox (1 transacción)
  Note over S,DB: commit atómico: o van los dos o ninguno
  R->>DB: leer outbox pendientes
  R->>B: publicar evento
  R->>DB: marcar como publicado

Scheduling con @nestjs/schedule

Para tareas periódicas dentro del proceso, @nestjs/schedule da tres primitivas declarativas: @Cron, @Interval y @Timeout.

import { ScheduleModule } from '@nestjs/schedule';
// AppModule: imports: [ScheduleModule.forRoot()]

@Injectable()
export class TasksService {
  @Cron(CronExpression.EVERY_DAY_AT_MIDNIGHT)
  purgeExpiredSessions() { /* ... */ }

  @Interval(10_000)
  pollQueue() { /* cada 10 s */ }

  @Timeout(5_000)
  warmupCache() { /* una vez, 5 s después del arranque */ }
}

El límite en multi-instancia

Esta es la trampa que rompe en producción. @nestjs/schedule corre el cron en cada instancia del proceso. En desarrollo tenés una instancia y funciona. En producción, detrás de un balanceador con 3 réplicas (procesos stateless, como manda el modelo 12-factor), el @Cron de medianoche se dispara 3 veces: mandás el reporte por triplicado, cobrás tres veces, purgás mientras otro escribe.

flowchart TD
  Cron[Cron 00:00] --> I1[Instancia 1 ejecuta]
  Cron --> I2[Instancia 2 ejecuta]
  Cron --> I3[Instancia 3 ejecuta]
  I1 --> X[Efecto ejecutado 3 veces]
  I2 --> X
  I3 --> X

Opciones para arreglarlo, de menor a mayor robustez:

  1. BullMQ repeatable jobs: el scheduling vive en Redis, corre una sola vez y el worker que lo tome lo procesa. Es la opción recomendada cuando ya tenés Redis.
  2. Lock distribuido: antes de ejecutar, cada instancia intenta tomar un lock en Redis (SET key NX EX) o una fila con clave única; solo la que lo gana ejecuta.
  3. Proceso scheduler dedicado: una sola réplica corre los crons y encola trabajo para los workers.

Nunca dejes un @Cron con efectos externos corriendo tal cual en un despliegue de N réplicas sin uno de estos mecanismos.

Antipatrones

Checklist

Con el trabajo pesado ya fuera del request path y los flujos asíncronos bajo control, el próximo frente es blindar la superficie expuesta: autenticación, autorización e inyecciones. Seguí con Seguridad →.

← Volver al listado de tecnologías ← Índice de NestJS: Buenas Prácticas