Events, Scheduling & Queues

This content was generated with the assistance of AI and should be verified against the official documentation before being relied on in production.

The reference material consulted while preparing these pages is listed in the section bibliography.

This page is written against NestJS 12.1 (npm 12.1.2), TypeScript 6 (6.0.3), Node.js 20.19+ or 22.12+, Express 5 and Fastify 5, and was checked on 2026-10-02. Package versions used: @nestjs/event-emitter 12.0.1 (on eventemitter2 6.4), @nestjs/schedule 12.0.2 (on cron 4.4), @nestjs/bullmq 12.0.0 and bullmq 6.3. The examples use the Nest 12 CLI project layout (ESM, Vitest). The event and scheduler examples run as shown with the real packages; the BullMQ examples are shown with in-memory stand-ins for Queue, Worker and FlowProducer, and behaviour that only a Redis server can show is described from the official and BullMQ documentation (see Testing events, schedules and queues).

Almost every real application has work that must not happen inside the request that triggered it: sending an email, resizing an uploaded cover, rebuilding a search index, or running a nightly report. Nest answers with three official packages that look similar and are not interchangeable:

Package Where the work runs Survives a restart? Use it for

@nestjs/event-emitter

The same process, usually the same tick

No: an event not yet handled is gone

Decoupling modules of one application: "a book was created, someone may care"

@nestjs/schedule

Every instance of the application, at a time you declare

No: no record of past runs, missed ticks are not replayed

Lightweight recurring housekeeping that is safe to run on every instance or is guarded by a lock

@nestjs/bullmq

Any worker process connected to the same Redis

Yes: jobs are persisted in Redis, with retries, delays and priorities

Durable background work, work that must run exactly where you want it, and work that needs back-pressure

None of the three runs inside an HTTP request. Event listeners, scheduled methods and queue consumers are plain providers invoked outside the request lifecycle, so guards, interceptors, pipes and exception filters do not wrap them: validate, authorise and handle errors inside the method itself.

Events inside one process: @nestjs/event-emitter

The Event Emitter package (@nestjs/event-emitter) is a thin Nest layer over the eventemitter2 library. It gives you an observer pattern between providers: one provider emits a named event with a payload, any number of other providers declare listeners with a decorator, and neither side knows about the other.

npm i --save @nestjs/event-emitter

Source: NestJS documentation: Events.

Registering the module

Import EventEmitterModule once, in the root module, and call forRoot(). The module is global by default, so EventEmitter2 can be injected anywhere without importing the module again; set global: false to opt out. forRoot() takes the options of EventEmitter2 plus two Nest-specific ones.

Table 1. Options of EventEmitterModule.forRoot()
Option Default Meaning

wildcard

false

Enable namespaces and wildcards in event names.

delimiter

'.'

Separator of the namespace segments.

newListener, removeListener

false

Emit the newListener / removeListener meta events.

maxListeners

10

Maximum listeners per event before EventEmitter2 warns about a leak.

verboseMemoryLeak

false

Include the event name in that warning.

ignoreErrors

false

Do not throw an uncaught exception when an error event is emitted and has no listener.

global

true

Nest option: register the module as a global module.

inheritRequestContextId

false

Nest option: when the emitted payload is the incoming request (or carries its context id), let request-scoped listeners reuse that request’s dependency-injection sub-tree instead of creating a new one.

The listener-registration work happens in the onApplicationBootstrap hook (see Lifecycle events and shutdown), after every module has been initialised and declared its listeners. That timing matters for the pitfall in Events emitted too early.

events.module.ts
import { Module } from '@nestjs/common';
import { EventEmitterModule } from '@nestjs/event-emitter';
import { BooksService } from './books.service.js';
import { AuditListener, SearchIndexListener } from './listeners.js';

@Module({
  imports: [
    EventEmitterModule.forRoot({
      wildcard: true, // enable namespaces and wildcards
      delimiter: '.', // segment separator for event names
      maxListeners: 20,
      verboseMemoryLeak: true,
    }),
  ],
  providers: [BooksService, SearchIndexListener, AuditListener],
  exports: [BooksService],
})
export class EventsModule {}

Source: NestJS documentation: Events, "Getting started".

Emitting events

Inject EventEmitter2 (imported from @nestjs/event-emitter, not from eventemitter2) and call emit() with an event name and a payload. A small class per event gives the payload a type and a place to evolve.

book-created.event.ts
export class BookCreatedEvent {
  constructor(
    readonly bookId: number,
    readonly title: string,
  ) {}
}
books.service.ts
import { Injectable } from '@nestjs/common';
import { EventEmitter2 } from '@nestjs/event-emitter';
import { BookCreatedEvent } from './book-created.event.js';

@Injectable()
export class BooksService {
  private nextId = 1;

  constructor(private readonly eventEmitter: EventEmitter2) {}

  create(title: string): { id: number; title: string } {
    const book = { id: this.nextId++, title };
    // emit() runs the listeners synchronously and returns true if there was at least one
    this.eventEmitter.emit('book.created', new BookCreatedEvent(book.id, book.title));
    return book;
  }

  async createAndWait(title: string): Promise<unknown[]> {
    const book = { id: this.nextId++, title };
    // emitAsync() returns a promise that resolves with every listener's result
    return this.eventEmitter.emitAsync('book.created', new BookCreatedEvent(book.id, book.title));
  }
}

Source: NestJS documentation: Events, "Dispatching events".

emit() returns true if the event had listeners and false otherwise; emitAsync() returns a promise for the array of listener results. Two behaviours are worth knowing:

  • emit() calls the listeners one after another before it returns. A listener that is an async method has started when emit() returns, but emit() does not wait for it. The calling code, for example the HTTP handler, therefore pays for synchronous listeners and not for the awaited part of asynchronous ones.

  • Nest wraps every @OnEvent() method in an async function, so emitAsync() does wait for the returned promises. The exception is a listener registered with async: true (below): emitAsync() then collects a timer handle instead of the result.

Listening with @OnEvent

Decorate a provider method with @OnEvent(event, options?). The first argument is a string or symbol; with wildcards enabled it may also be an array of them. The optional second argument is OnEventOptions: the eventemitter2 listener options (async, promisify, nextTick, objectify) plus two Nest options.

Table 2. Nest-specific listener options
Option Default Meaning

prependListener

false

Put this listener before the ones already registered for the event instead of after them.

suppressErrors

true

Log an error thrown by the listener instead of rethrowing it. With false the error is rethrown from the (async) wrapper.
listeners.ts
import { Injectable } from '@nestjs/common';
import { OnEvent } from '@nestjs/event-emitter';
import { BookCreatedEvent } from './book-created.event.js';

export const trace: string[] = [];

@Injectable()
export class SearchIndexListener {
  // A plain listener: runs synchronously, inside emit()
  @OnEvent('book.created')
  index(event: BookCreatedEvent) {
    trace.push(`index:${event.bookId}`);
  }

  // prependListener puts this listener before the others registered for the event
  @OnEvent('book.created', { prependListener: true })
  validate(event: BookCreatedEvent) {
    trace.push(`validate:${event.bookId}`);
  }

  // async: true schedules the handler on a later tick, so emit() does not wait for it
  @OnEvent('book.created', { async: true })
  async sendNotification(event: BookCreatedEvent) {
    await new Promise((resolve) => setTimeout(resolve, 10));
    trace.push(`notify:${event.bookId}`);
    return `notified ${event.bookId}`;
  }

  // suppressErrors defaults to true: a throwing listener is logged, not rethrown
  @OnEvent('book.created')
  failing() {
    throw new Error('listener failed');
  }
}

@Injectable()
export class AuditListener {
  // "book.*" matches exactly one more segment: book.created, book.deleted, ...
  @OnEvent('book.*')
  oneLevel(event: unknown) {
    trace.push(`audit:one:${event instanceof BookCreatedEvent ? event.title : String(event)}`);
  }

  // "**" matches everything, at any depth
  @OnEvent('**')
  everything(payload: unknown) {
    trace.push(`audit:all:${typeof payload}`);
  }
}

Running create('Dune') against this module produced the trace validate:1, index:1, audit:one:Dune, audit:all:object before create() returned: the prepended listener first, then the others in registration order, with the failing listener logged by Nest (listener failed) and not breaking the call. notify:1 appeared roughly ten milliseconds later.

Source: NestJS documentation: Events, "Listening to events".

Wildcards and namespaces

With wildcard: true, an event name is a delimiter-separated path and a listener may use wildcards. A single * matches exactly one segment: order.* matches order.created and order.shipped, but not order.delayed.out_of_stock. The multi-level wildcard ** matches any number of segments, so @OnEvent('**') receives every event in the application, which is a convenient hook for audit logging and a trap for performance. Without wildcard: true, the same strings are matched literally.

Errors, rethrowing and emitAsync

A throwing listener should not usually break the code that emitted the event, which is why suppressErrors defaults to true. If you set it to false, remember that the wrapper Nest registers is asynchronous: the rethrown error is a rejected promise. emitAsync() hands that rejection to the caller; a plain emit() does not, and the rejection becomes an unhandled promise rejection, which terminates a Node.js process by default.

strict.listeners.ts
import { Injectable } from '@nestjs/common';
import { OnEvent } from '@nestjs/event-emitter';

@Injectable()
export class StrictListeners {
  // Every listener is wrapped by Nest in an async function, so emitAsync() awaits it
  @OnEvent('invoice.issued')
  async total(amount: number) {
    await new Promise((resolve) => setTimeout(resolve, 5));
    return amount * 2;
  }

  // suppressErrors: false rethrows. The rethrow is a rejected promise, which only
  // emitAsync() hands back to the caller; with emit() it becomes an unhandled rejection
  @OnEvent('invoice.rejected', { suppressErrors: false })
  async reject() {
    throw new Error('invoice rejected');
  }

  // async: true is an EventEmitter2 option; the result emitAsync() collects is a timer handle
  @OnEvent('invoice.deferred', { async: true })
  deferred(amount: number) {
    return amount;
  }
}

With that provider registered, await eventEmitter.emitAsync('invoice.issued', 21) resolves with [42], emitAsync('invoice.rejected') rejects with invoice rejected, and emitAsync('invoice.deferred', 1) resolves with a timer handle instead of 1. Use emitAsync() when the caller needs the results or must observe failures, and a plain emit() for fire-and-forget notifications.

Source: NestJS documentation: Events, "Listening to events" and the EventEmitter2 documentation.

Events emitted too early

Listeners are registered in onApplicationBootstrap. An event emitted from a constructor or from onModuleInit() can therefore be emitted before its listener exists, and it is lost without any error. EventEmitterReadinessWatcher, exported by the package, exposes waitUntilReady(), a promise that resolves once every listener is registered. Call it in your own onApplicationBootstrap() before emitting.

readiness.service.ts
import { Injectable, OnApplicationBootstrap, OnModuleInit } from '@nestjs/common';
import { EventEmitter2, EventEmitterReadinessWatcher } from '@nestjs/event-emitter';

// Emitting from onModuleInit() can be lost: @OnEvent listeners are registered in onApplicationBootstrap.
@Injectable()
export class EarlyEmitter implements OnModuleInit {
  constructor(private readonly eventEmitter: EventEmitter2) {}

  onModuleInit() {
    this.eventEmitter.emit('app.warm', 'from onModuleInit');
  }
}

// Waiting for readiness first makes the emit safe.
@Injectable()
export class SafeEmitter implements OnApplicationBootstrap {
  constructor(
    private readonly eventEmitter: EventEmitter2,
    private readonly readiness: EventEmitterReadinessWatcher,
  ) {}

  async onApplicationBootstrap() {
    await this.readiness.waitUntilReady();
    this.eventEmitter.emit('app.warm', 'from onApplicationBootstrap');
  }
}

A listener for app.warm receives only from onApplicationBootstrap; the event from onModuleInit() is dropped.

Source: NestJS documentation: Events, "Preventing event loss".

Request-scoped listeners

If a listener belongs to a request-scoped provider (see Injection scopes), Nest creates a new instance of that provider for every event it handles, and the event payload takes the place of the request object: injecting REQUEST into the provider gives you the payload. Set inheritRequestContextId: true in forRoot() when the payload is the incoming request itself, so the listener reuses that request’s dependency-injection sub-tree.

It is in-process only

@nestjs/event-emitter is an in-memory observer. Events never leave the Node.js process: there is no network, no persistence, no retry and no delivery guarantee. Two instances of the same application, or two applications in a monorepo, each have their own emitter, and an event emitted in one is invisible to the other. If the process stops between emit() and the end of a listener, the work is simply lost.

This is the most common misuse found in tutorials, including the reference book (see Where the book is outdated): using the event emitter to "notify another microservice". For communication between services use a transport that crosses the process boundary: a Nest microservice transport (Messaging Reference explains brokers, and Kafka vs. RabbitMQ helps choose one) or a Redis-backed queue as described below. Inside one process, events are excellent: they keep a BooksService from importing the search, notification and audit modules it would otherwise have to know about.

Express and Fastify

Events have no HTTP dependency, so there is no difference between the Express and Fastify adapters, and the same module works in a standalone application context without an HTTP server (Platform agnosticism and standalone applications).

Time-based jobs: @nestjs/schedule

Task scheduling runs a method at a fixed date and time, at recurring intervals, or once after a delay. In the Linux world this is cron; in Node.js, @nestjs/schedule integrates the cron package and the built-in setInterval() and setTimeout() into Nest’s lifecycle.

npm install --save @nestjs/schedule

Source: NestJS documentation: Task scheduling.

Activating the scheduler

Import ScheduleModule and call forRoot() in the root module. The call initialises the scheduler and registers every declarative @Cron(), @Interval() and @Timeout() found on providers; the jobs start in onApplicationBootstrap, after every module has loaded. The module is global, so do not import it again elsewhere.

schedule-app.module.ts
import { Module } from '@nestjs/common';
import { ScheduleModule } from '@nestjs/schedule';
import { TasksService } from './tasks.service.js';

@Module({
  imports: [ScheduleModule.forRoot()], // call forRoot() in ONE module only
  providers: [TasksService],
})
export class ScheduleAppModule {}

forRoot() takes optional cronJobs, intervals and timeouts flags (per the package typings) to switch a whole category off, and forRootAsync() takes the same options from a factory. Running a module with forRoot({ cronJobs: false }) registered no cron jobs but kept the interval.

The official documentation carries a warning worth taking literally: call forRoot() in one module only. A second forRoot() registers every job again, and a @Cron('* * * * * *') job then runs twice per tick.

Source: NestJS documentation: Task scheduling, "Installation".

Cron jobs, CronExpression and options

Decorate a method with @Cron(pattern, options?). The pattern has six fields, the first (seconds) optional:

* * * * * *
| | | | | |
| | | | | day of week
| | | | months
| | | day of month
| | hours
| minutes
seconds (optional)

The supported syntax is asterisks, ranges and lists (1-3,5) and steps (*/2). 0 30 11 * * 1-5 is Monday to Friday at 11:30 and 0 */30 9-17 * * * is every 30 minutes between 9 and 17. The pattern can instead be a Date, which runs the job exactly once at that moment. The package ships a CronExpression enum of ready-made patterns (EVERY_30_SECONDS is */30 * * * * *, EVERY_DAY_AT_MIDNIGHT, EVERY_5_MINUTES, and many more).

Table 3. Options of @Cron()
Option Meaning

name

Names the job so the SchedulerRegistry can find it.

timeZone

IANA time zone for the pattern. An invalid zone throws when the application starts.

utcOffset

Offset instead of a zone name. Do not combine with timeZone; the typings make the two mutually exclusive.

waitForCompletion

If true, ticks that fire while the previous run is still going are skipped entirely.

disabled

If true, the job is registered but never runs on its own; start it later with the registry.

The package typings list a few more options (unrefTimeout, threshold, initialDelay) that the documentation page does not describe; check the package README before relying on them. A duplicate job name makes the scheduler throw at start-up.

Every decorated method is wrapped in a try/catch, so an exception is logged instead of crashing the process.

tasks.service.ts
import { Injectable, Logger } from '@nestjs/common';
import { Cron, CronExpression, Interval, SchedulerRegistry, Timeout } from '@nestjs/schedule';
import { CronJob } from 'cron';

export const runs = { everySecond: 0, slow: 0, interval: 0, timeout: 0, dynamic: 0 };

@Injectable()
export class TasksService {
  private readonly logger = new Logger(TasksService.name);

  constructor(private readonly schedulerRegistry: SchedulerRegistry) {}

  // Six fields: seconds (optional), minutes, hours, day of month, month, day of week
  @Cron('* * * * * *', { name: 'every-second' })
  everySecond() {
    runs.everySecond++;
  }

  // waitForCompletion: ticks that fire while the previous run is still going are skipped
  @Cron('* * * * * *', { name: 'slow-report', waitForCompletion: true })
  async slowReport() {
    runs.slow++;
    await new Promise((resolve) => setTimeout(resolve, 2500));
  }

  // Ready-made patterns live in the CronExpression enum
  @Cron(CronExpression.EVERY_DAY_AT_MIDNIGHT, { name: 'nightly-cleanup', timeZone: 'Europe/Madrid' })
  nightlyCleanup() {
    this.logger.log('midnight in Madrid');
  }

  // disabled: the job is registered (so it can be started later) but never runs on its own
  @Cron(CronExpression.EVERY_5_MINUTES, { name: 'maintenance', disabled: true })
  maintenance() {
    this.logger.log('maintenance');
  }

  @Interval('poll-feed', 50)
  pollFeed() {
    runs.interval++;
  }

  @Timeout('warm-cache', 30)
  warmCache() {
    runs.timeout++;
  }

  // Dynamic API: create a job at runtime and register it by name
  addCronJob(name: string, seconds: string) {
    const job = new CronJob(`${seconds} * * * * *`, () => {
      runs.dynamic++;
    });
    this.schedulerRegistry.addCronJob(name, job);
    job.start();
  }

  describeJobs() {
    const out: string[] = [];
    this.schedulerRegistry.getCronJobs().forEach((job, name) => {
      let next: Date | string;
      try {
        next = job.nextDate().toJSDate();
      } catch {
        next = 'no future date';
      }
      out.push(`${name}: next=${typeof next === 'string' ? next : next.toISOString()} running=${job.isActive}`);
    });
    return out;
  }
}

Over three seconds the slow-report job, which takes 2.5 seconds, runs once thanks to waitForCompletion while every-second runs three times; maintenance reports isActive as false.

Source: NestJS documentation: Task scheduling, "Declarative cron jobs" and the cron package.

Intervals and timeouts

@Interval(ms) runs a method repeatedly and @Timeout(ms) runs it once, that many milliseconds after the application starts. Both are thin wrappers over setInterval() and setTimeout(). Pass a name as the first argument (@Interval('poll-feed', 50)) to reach the job through the registry. As with cron jobs, an exception is caught and logged. Because a timer starts when the scheduler starts, an interval has no notion of wall-clock time: @Interval(3_600_000) fires one hour after each process start, not on the hour. Use @Cron() for that.

SchedulerRegistry, the dynamic API

Inject SchedulerRegistry to manage jobs at runtime. A named declarative job can be fetched and controlled; new jobs can be created from values read at runtime (a cron pattern from a settings table, say).

Table 4. SchedulerRegistry methods
Method Purpose

getCronJob(name), getCronJobs()

Fetch a CronJob (it has start(), stop(), setTime(), lastDate(), nextDate(), nextDates(count) and the isActive flag), or the Map of all of them.

addCronJob(name, job), deleteCronJob(name)

Register a CronJob created with new CronJob(pattern, callback) from the cron package, or remove one. Call job.start() after adding.

getInterval(name), addInterval(name, id), deleteInterval(name), getIntervals()

Work with the timer handle returned by setInterval(); getIntervals() returns the names.

getTimeout(name), addTimeout(name, id), deleteTimeout(name), getTimeouts()

The same for setTimeout() handles.

doesExist(type, name)

type is 'cron', 'timeout' or 'interval' (per the package typings).

nextDate() throws when a job has no future date (a one-shot job that has already fired), which is why describeJobs() above guards it with try/catch. After the checks above, getCronJob('maintenance').start() brought the disabled job to isActive: true and deleteInterval('poll-feed') emptied getIntervals().

Source: NestJS documentation: Task scheduling, "Dynamic schedule module API".

Running on several instances

The scheduler runs every job in every process. With three instances behind a load balancer, @Cron('0 2 * * *') runs three times a night: three invoice exports, three batches of reminder emails. This is the single most important fact about @nestjs/schedule in production, and the options are:

  • A distributed lock. The official page recommends @OnOneInstance() from @nestjs/locks (a pre-1.0 package) next to the scheduling decorator. The instances coordinate through a lock store they share, such as your database or Redis, the first instance to fire the job takes a lease and keeps renewing it, and the others skip their ticks. @nestjs/locks ships an in-memory store that only excludes callers within one process, so for several instances you provide a shared LockStore; see the Distributed locks chapter. Missed ticks are not caught up, so write jobs that do everything that is due, not "yesterday’s" work.

  • A queue job scheduler. Add the repeating job to a BullMQ queue (below): the schedule lives in Redis once, and whichever worker is free takes each job, so it runs once regardless of the number of instances.

  • One dedicated instance. Run the scheduler in a single worker deployment and keep it out of the API replicas, for example by importing the module only in the worker’s root module.

  • An external trigger. A Kubernetes CronJob or a platform scheduler that calls an endpoint or runs a command.

invoices.job.ts
import { Injectable } from '@nestjs/common';
import { OnOneInstance, WithoutOverlapping } from '@nestjs/locks';
import { Cron } from '@nestjs/schedule';

export const exported: string[] = [];

@Injectable()
export class InvoicesJob {
  // Only the instance that owns the lease runs the tick; the others skip it
  @Cron('0 2 * * *', { name: 'invoice-export' })
  @OnOneInstance({ key: 'invoices:nightly-export' })
  @WithoutOverlapping() // skip a tick while the previous run is still going
  async exportInvoices() {
    exported.push('run');
    await new Promise((resolve) => setTimeout(resolve, 50));
  }
}

This class was compiled, and with LocksModule.forRoot({ allowInMemoryStorage: true }) two concurrent calls to exportInvoices() produced one run. That proves the decorators compose and the overlap guard works; it does not prove cross-instance exclusion, which depends on the shared store you register. In production (NODE_ENV=production) @nestjs/locks refuses to start on the in-memory store unless you allow it explicitly, because every instance would take the same lock.

Source: NestJS documentation: Task scheduling, "Running on several instances" and Distributed locks.

Pitfalls of the scheduler

  • A job that crashes with the process is simply not run. There is no run history and no catch-up. If "did the nightly job run?" matters, record each run yourself or use a queue.

  • A slow job overlaps itself. A plain @Cron('* * * * * *') starts a new run every second even when the last has not finished. Use waitForCompletion: true for one process or @WithoutOverlapping() from @nestjs/locks.

  • Time zones are explicit. Without timeZone, a pattern runs in the server’s zone, which in containers is usually UTC.

  • Heavy work belongs in a queue. A @Cron() method runs on the application’s event loop. Let the cron method enqueue a job and have workers do the heavy part.

  • Silent failure is the real risk. A scheduled job that throws is logged; a job that silently stops being scheduled is not. The official documentation describes job-silence alerts in NestJS Observe for this case (see Observability overview).

  • Duplicate names fail. Two jobs with the same name of the same kind make the scheduler throw at start-up, so give every named job a unique name.

  • Shutdown. The scheduler clears its timeouts and intervals and stops its cron jobs in beforeApplicationShutdown, so they stop when the application closes; enable shutdown hooks so that SIGTERM triggers this (Lifecycle events and shutdown).

Express and Fastify

The scheduler is independent of the HTTP adapter. It behaves identically with Express 5 and Fastify 5, and in a standalone application context.

Durable work: BullMQ and Redis

A queue decouples producing work from doing it. A producer, typically an HTTP handler, adds a job to a queue and returns immediately; a worker, typically another process, takes jobs from the queue and runs them. Nest ships @nestjs/bullmq for BullMQ and @nestjs/bull for its predecessor Bull; both wrap libraries from the same team. Bull is in maintenance mode (bug fixes only) and BullMQ is the actively developed TypeScript implementation, so new code should use BullMQ.

Queues help in three ways, which the official page lists: they smooth out processing peaks, they keep CPU-heavy or blocking work off the process serving requests, and they provide a reliable channel between services, because jobs persisted in Redis survive the crash of a producer or a consumer and are picked up again when nodes restart.

flowchart LR subgraph api["API process (producer)"] ctrl["Controller / service"] -->|"queue.add('resize', data)"| q end subgraph redis["Redis"] q[("covers queue
wait / delayed / active /
completed / failed")] end subgraph w1["Worker process 1"] c1["@Processor('covers')
process(job)"] end subgraph w2["Worker process 2"] c2["@Processor('covers')
process(job)"] end q -->|"job"| c1 q -->|"job"| c2 c1 -.->|"completed / failed"| q c2 -.->|"completed / failed"| q q -.->|"QueueEvents stream"| ev["@QueueEventsListener
(any process)"]

Source: NestJS documentation: Queues.

Installing and connecting

npm install --save @nestjs/bullmq bullmq ioredis

The official page lists only @nestjs/bullmq bullmq. BullMQ 6 treats its Redis driver as an optional peer dependency (ioredis, redis or pg for its PostgreSQL backend), and creating a Queue without ioredis installed throws BullMQ could not load the optional 'ioredis' package. Installing ioredis fixes it, and a plain Node.js ESM script using bullmq then reaches the connection stage normally. You also need a running Redis server for anything real. BullMQ keeps jobs in Redis, so Redis must not evict keys under memory pressure; the BullMQ documentation recommends the noeviction memory policy for production.

BullModule.forRoot() registers a connection configuration shared by every queue in the application; forRootAsync() accepts a factory so the host and port come from ConfigService rather than from source code. The options are the BullMQ QueueOptions, plus extraOptions for Nest.

queue-infra.module.ts
import { Module } from '@nestjs/common';
import { ConfigModule, ConfigService } from '@nestjs/config';
import { BullModule } from '@nestjs/bullmq';

export const COVERS_QUEUE = 'covers';
export const IMPORT_FLOW = 'bookImport';

const bullInfra = [
  BullModule.forRootAsync({
    imports: [ConfigModule],
    inject: [ConfigService],
    useFactory: (config: ConfigService) => ({
      connection: {
        host: config.get<string>('REDIS_HOST', 'localhost'),
        port: Number(config.get('REDIS_PORT', 6379)), // env values are strings
      },
      prefix: 'bookshelf', // every queue key starts with bookshelf: instead of bull:
      defaultJobOptions: {
        attempts: 3,
        backoff: { type: 'exponential', delay: 1000 },
        removeOnComplete: { count: 1000 },
        removeOnFail: { age: 7 * 24 * 3600 },
      },
    }),
  }),
  BullModule.registerQueue({ name: COVERS_QUEUE }),
  BullModule.registerFlowProducer({ name: IMPORT_FLOW }),
];

@Module({
  imports: [ConfigModule.forRoot({ isGlobal: true }), ...bullInfra],
  exports: bullInfra,
})
export class QueueInfraModule {}

Several forRoot() options are worth knowing:

  • connection configures Redis (host, port, credentials, TLS; or an already-constructed client). Credentials come from ConfigService or the environment, never from source code.

  • prefix prefixes every key (default bull), which lets several applications or environments share a Redis server.

  • defaultJobOptions sets the default for new jobs. It does not apply to jobs added through a FlowProducer.

  • settings holds advanced options for repeatable jobs and should usually be left alone.

  • extraOptions carries Nest’s own options, currently manualRegistration (see Manual registration).

Two details from the run: ConfigService.get() returns strings for environment variables even when you give it a numeric type argument, so convert the port with Number(); and the queue name is both the injection token and the argument of the consumer decorators.

Beyond forRootAsync() with a factory, useClass (with SharedBullConfigurationFactory) and useExisting are supported, and registerQueueAsync() configures an individual queue asynchronously. To connect different queues to different Redis servers, register named configurations: BullModule.forRoot('alternative-config', { …​ }) and point a queue at it with configKey in registerQueue().

Source: NestJS documentation: Queues, "BullMQ installation" and BullMQ: Connections.

Producers

Register queues with BullModule.registerQueue({ name }) (several configuration objects may be passed at once), inject one with @InjectQueue(name), and call add(jobName, data, options?). Queues are shared by every module and process that connects to the same Redis with the same credentials and queue name. Job data must be serialisable, because it is stored in Redis.

covers.service.ts
import { Injectable } from '@nestjs/common';
import { InjectFlowProducer, InjectQueue } from '@nestjs/bullmq';
import { FlowProducer, Queue } from 'bullmq';
import { COVERS_QUEUE, IMPORT_FLOW } from './queue-infra.module.js';

export interface CoverJob {
  bookId: number;
  source: string;
}

@Injectable()
export class CoversService {
  constructor(
    @InjectQueue(COVERS_QUEUE) private readonly queue: Queue<CoverJob>,
    @InjectFlowProducer(IMPORT_FLOW) private readonly flows: FlowProducer,
  ) {}

  async enqueueResize(bookId: number, source: string) {
    const job = await this.queue.add(
      'resize',
      { bookId, source },
      {
        jobId: `resize-${bookId}`, // idempotent: a second add with the same id is ignored
        priority: 2,
        delay: 3000,
        attempts: 5, // overrides the default of 3
      },
    );
    return job.id;
  }

  // Parent/child flow: "publish" runs only after both children completed
  async enqueueImport(bookId: number) {
    const tree = await this.flows.add({
      name: 'publish',
      queueName: COVERS_QUEUE,
      data: { bookId },
      children: [
        { name: 'resize', queueName: COVERS_QUEUE, data: { bookId, source: 'large' } },
        { name: 'thumbnail', queueName: COVERS_QUEUE, data: { bookId, source: 'small' } },
      ],
    });
    return tree.job.id;
  }

  // A repeatable job is a job scheduler (the repeat option of add() is gone from BullMQ)
  async scheduleNightlyThumbnails() {
    await this.queue.upsertJobScheduler(
      'nightly-thumbnails',
      { pattern: '0 3 * * *' },
      { name: 'thumbnail', data: { bookId: 0, source: 'all' } },
    );
  }

  pause() {
    return this.queue.pause();
  }

  resume() {
    return this.queue.resume();
  }
}

Source: NestJS documentation: Queues, "Producers".

Table 5. Common job options (third argument of add())
Option Meaning

priority

0 to 2,097,151. 0 (default) means no priority; jobs without one run before prioritised ones, and among prioritised jobs lower numbers run first. Priorities have a small performance cost.

delay

Milliseconds to wait before the job can be processed. Keep the clocks of producers and workers synchronised for accurate delays.

attempts

Total number of attempts before the job is failed (default 1).

backoff

A number or { type, delay } for the wait between retries ('exponential' and 'fixed' are the built-in types).

lifo

Add to the other end of the queue so the job is processed last-in, first-out.

jobId

Overrides the job id. You must keep it unique; adding a job whose id exists does not add it, which makes jobId a simple deduplication key.

removeOnComplete, removeOnFail

true, a count, or { age, count }. By default finished jobs stay in Redis forever, so set these in defaultJobOptions.

stackTraceLimit

Limit the stack-trace lines recorded for a failed job.

BullMQ no longer accepts a repeat option in Queue.add(): use job schedulers (below). Queue.pause() and resume() stop and restart a queue: a paused queue starts nothing new, and jobs already running finish.

Consumers: @Processor and WorkerHost

A consumer is a provider class decorated with @Processor(queueName) that extends WorkerHost and implements process(job). process() is called whenever the worker is idle and there is a job; its return value is stored on the job and passed to completed listeners. A thrown error fails the attempt. The class must be registered in a module’s providers so that @nestjs/bullmq can discover it.

The second argument of @Processor() takes BullMQ worker options, for example concurrency (how many jobs one worker runs in parallel), limiter (rate limiting), lockDuration and stalledInterval. The decorator also accepts an object form with name, scope and configKey.

BullMQ has one process() per class, unlike Bull’s @Process('name'). To handle several job kinds, switch on job.name.

covers.consumer.ts
import { Logger } from '@nestjs/common';
import { OnWorkerEvent, Processor, WorkerHost } from '@nestjs/bullmq';
import { Job } from 'bullmq';
import { COVERS_QUEUE } from './queue-infra.module.js';
import type { CoverJob } from './covers.service.js';

export const processed: string[] = [];

// The second argument takes BullMQ WorkerOptions (concurrency, limiter, lockDuration, ...)
@Processor(COVERS_QUEUE, { concurrency: 5 })
export class CoversConsumer extends WorkerHost {
  private readonly logger = new Logger(CoversConsumer.name);

  // One process() per class: branch on job.name
  async process(job: Job<CoverJob, unknown, string>): Promise<unknown> {
    switch (job.name) {
      case 'resize': {
        for (let pct = 25; pct <= 100; pct += 25) {
          await job.updateProgress(pct);
        }
        processed.push(`resize:${job.data.bookId}`);
        return { width: 800 };
      }
      case 'thumbnail':
        processed.push(`thumbnail:${job.data.bookId}`);
        return { width: 120 };
      case 'publish': {
        // parent of a flow: the children's return values, keyed by job key
        const children = await job.getChildrenValues();
        processed.push(`publish:${job.data.bookId}:${Object.keys(children).length}`);
        return children;
      }
      default:
        throw new Error(`Unknown job ${job.name}`); // marks the attempt as failed
    }
  }

  // Worker-level events: only for jobs this worker processed
  @OnWorkerEvent('completed')
  onCompleted(job: Job) {
    this.logger.log(`job ${job.id} (${job.name}) completed`);
  }

  @OnWorkerEvent('failed')
  onFailed(job: Job | undefined, error: Error) {
    this.logger.warn(`job ${job?.id} failed on attempt ${job?.attemptsMade}: ${error.message}`);
  }

  // Always handle 'error': an unhandled worker error event can crash the process
  @OnWorkerEvent('error')
  onError(error: Error) {
    this.logger.error(error.message);
  }
}

The worker created for this class receives concurrency: 5, the bookshelf prefix and the connection built from ConfigService, and driving its process() with a resize job returns { width: 800 } while an unknown job name fails with Unknown job nope.

Source: NestJS documentation: Queues, "Consumers".

Request-scoped consumers

A consumer flagged request-scoped (@Processor({ name: 'covers', scope: Scope.REQUEST })) is instantiated once per job and released afterwards; inject the current job with @Inject(JOB_REF) (JOB_REF is exported by @nestjs/bullmq). Request-scoped classes cannot register event listeners: Nest logs a warning and ignores @OnWorkerEvent() handlers on them. See Injection scopes for the cost of scoped providers.

Worker events and queue events

BullMQ reports state changes at two levels, and Nest has a decorator for each.

  • @OnWorkerEvent(event) goes inside the @Processor() class and is bound to that worker. It sees only the jobs that worker processed (active, completed, failed, progress, error, stalled, and the others of BullMQ’s WorkerListener interface).

  • @QueueEventsListener(queueName) marks a class that extends QueueEventsHost, and @OnQueueEvent(event) marks its handlers. It listens to the queue’s event stream in Redis, so it sees events from every worker in every process. Its handlers receive small payloads (jobId, returnvalue, failedReason) and not Job objects. Register the class in providers; Nest closes its connection on application shutdown.

covers.events.ts
import { Logger } from '@nestjs/common';
import { OnQueueEvent, QueueEventsHost, QueueEventsListener } from '@nestjs/bullmq';
import { COVERS_QUEUE } from './queue-infra.module.js';

// Queue-level events: every worker in every process, delivered through a Redis stream
@QueueEventsListener(COVERS_QUEUE)
export class CoversEvents extends QueueEventsHost {
  private readonly logger = new Logger(CoversEvents.name);

  @OnQueueEvent('completed')
  onCompleted({ jobId, returnvalue }: { jobId: string; returnvalue: unknown }) {
    this.logger.log(`job ${jobId} completed with ${JSON.stringify(returnvalue)}`);
  }

  @OnQueueEvent('failed')
  onFailed({ jobId, failedReason }: { jobId: string; failedReason: string }) {
    this.logger.warn(`job ${jobId} failed: ${failedReason}`);
  }
}

This class was compiled; it was not started, because QueueEvents is created internally with the real BullMQ class and needs a live Redis connection.

Source: NestJS documentation: Queues, "Event listeners".

Flows and scheduled jobs

BullMQ supports parent-child relationships between jobs. A flow is a tree: the children run first, and the parent moves to the active state only when they have all completed. Register a producer with BullModule.registerFlowProducer({ name }), inject it with @InjectFlowProducer(name) and call add() with a tree of { name, queueName, data, children } nodes, as enqueueImport() does above. In the parent’s process(), job.getChildrenValues() returns the children’s return values. Note that defaultJobOptions from forRoot() do not apply to jobs added through a FlowProducer, so set the options on the nodes.

Repeating work is expressed with a job scheduler, created with queue.upsertJobScheduler(id, { pattern }, { name, data, opts }). upsert means calling it at every start-up is safe: it creates the scheduler or updates it in place. Because the schedule is stored in Redis, one job is produced per tick however many instances start the application, which solves the several-instance problem of @nestjs/schedule for durable jobs.

Source: BullMQ: Flows and BullMQ: Job schedulers.

Separate processes

Two different things are called separate processes.

Producer and workers in different deployments. A Nest process starts a worker for every @Processor() provider in its module graph. If the API module imports the consumer, every API replica also becomes a worker and competes with the HTTP traffic for the event loop. Keep producers and consumers in different modules, give each deployment its own root module, and start the worker as an application context without an HTTP listener:

modules.ts
import { Module } from '@nestjs/common';
import { CoversConsumer } from './covers.consumer.js';
import { CoversEvents } from './covers.events.js';
import { CoversService } from './covers.service.js';
import { QueueInfraModule } from './queue-infra.module.js';

// HTTP process: producers only
@Module({
  imports: [QueueInfraModule],
  providers: [CoversService],
  exports: [CoversService],
})
export class CoversApiModule {}

// Worker process: consumers and listeners, no HTTP server
@Module({
  imports: [QueueInfraModule],
  providers: [CoversConsumer, CoversEvents],
})
export class CoversWorkerModule {}
worker.ts
import { NestFactory } from '@nestjs/core';
import { CoversWorkerModule } from './modules.js';

async function bootstrap() {
  // An application context: DI container and lifecycle hooks, but no HTTP listener
  const app = await NestFactory.createApplicationContext(CoversWorkerModule);
  app.enableShutdownHooks(); // on SIGTERM the workers are closed after their active jobs finish
}
void bootstrap();

Both files compiled with the project build. Worker.close() waits for the jobs being processed unless it is called with force, and the explorer calls it on application shutdown, so with shutdown hooks enabled a SIGTERM lets running jobs finish (give the container a termination grace period longer than your slowest job; see Lifecycle events and shutdown). The application-context pattern is the one described in Platform agnosticism and standalone applications. Producers and consumers also do not have to be written in Nest or in the same language, since Redis is the only shared contract.

Sandboxed processors. BullMQ can run the handler itself in a forked child process (see BullMQ: Sandboxed processors). The advantages are isolation from crashes, blocking code that does not stall the worker, better multi-core use and fewer Redis connections. In Nest, pass the file to registerQueue():

BullModule.registerQueue({
  name: 'audio',
  processors: [join(import.meta.dirname, 'processor.js')],
});

The child runs outside the Nest container, so dependency injection is not available in it: the processor file must create the dependencies it needs. In a CommonJS project use __dirname instead of import.meta.dirname.

Source: NestJS documentation: Queues, "Separate processes".

Manual registration

By default BullModule registers queues, processors and event listeners during onModuleInit. To prevent that, set extraOptions: { manualRegistration: true } in forRoot() and inject BullRegistrar, calling register() yourself, ideally in onModuleInit() or onApplicationBootstrap(). Until register() runs, no BullMQ components work and no job is processed. This is how one codebase can start in "API only" or "worker only" mode from an environment variable without two root modules.

import { Injectable, OnModuleInit } from '@nestjs/common';
import { BullRegistrar } from '@nestjs/bullmq';

@Injectable()
export class WorkerSwitch implements OnModuleInit {
  constructor(private readonly registrar: BullRegistrar) {}

  onModuleInit() {
    if (process.env.RUN_WORKERS === 'true') {
      this.registrar.register();
    }
  }
}

Source: NestJS documentation: Queues, "Manual registration".

Testing queue code

Unit tests do not need Redis. Provide the queue under its injection token with getQueueToken(name) and a fake, and call the consumer’s process() directly with a stand-in job. getFlowProducerToken(name) does the same for flow producers.

import { getQueueToken } from '@nestjs/bullmq';
import { Test } from '@nestjs/testing';
import { CoversService } from './covers.service.js';

const queue = { add: vi.fn().mockResolvedValue({ id: '1' }) };
const mod = await Test.createTestingModule({
  providers: [CoversService, { provide: getQueueToken('covers'), useValue: queue }],
}).compile();
await mod.get(CoversService).enqueueResize(1, 's');
expect(queue.add).toHaveBeenCalledWith('resize', { bookId: 1, source: 's' }, expect.objectContaining({ jobId: 'resize-1' }));

The full test also supplies the flow-producer token, which CoversService injects. Tests of retries, delays and flows need a real Redis; run one in a container in your CI.

Source: NestJS documentation: Testing (overriding providers).

Legacy @nestjs/bull

@nestjs/bull wraps Bull, the maintenance-mode predecessor. Skip this section if you use BullMQ. The module API is similar, with different names, and the consumer style differs:

Table 6. @nestjs/bull compared with @nestjs/bullmq
Concern @nestjs/bull (Bull) @nestjs/bullmq (BullMQ)

Install

npm i @nestjs/bull bull

npm i @nestjs/bullmq bullmq ioredis

Connection option

BullModule.forRoot({ redis: { …​ } })

BullModule.forRoot({ connection: { …​ } })

Consumer class

@Processor('audio') on a plain class

@Processor('audio') on a class that extends WorkerHost

Job handlers

Methods marked @Process() or @Process('name'), several per class

One process(job) method; switch on job.name

Worker events

@OnQueueActive(), @OnQueueCompleted(), @OnQueueFailed(), @OnQueueError(), and so on; @OnGlobalQueue…​() for events from all workers

@OnWorkerEvent(event) in the processor; @QueueEventsListener() + @OnQueueEvent(event) for queue-level events

Repeating jobs

repeat job option

Job schedulers (upsertJobScheduler())

Status

Maintenance mode, bug fixes only

Actively developed

When you maintain a Bull application, keep it working and use BullMQ for new queues. They are different libraries with different job APIs, so do not point a Bull producer and a BullMQ worker at the same queue name.

Source: NestJS documentation: Queues, "Bull installation" and the Bull reference.

Pitfalls of queues

  • Importing the consumer into the API module. Every replica becomes a worker. Split the modules, or use manualRegistration.

  • Forgetting ioredis. BullMQ 6 does not install a Redis driver for you.

  • Unbounded Redis growth. Finished jobs are kept by default. Set removeOnComplete and removeOnFail (counts or ages).

  • Non-idempotent jobs. A job can run more than once (retries, stalled jobs that are moved back to the wait list). Make handlers idempotent, and use jobId to deduplicate producers.

  • Large or non-serialisable data. Store an id or a storage key in the job, not the file.

  • Unhandled error events. Handle @OnWorkerEvent('error'), as in the example.

  • Throwing non-Error values from process(). Throw Error objects so a message and stack trace are recorded on the failed job.

  • Treating a queue like a database. Queues move work; keep results in your database and the job’s return value small.

Express and Fastify

Consumers and producers are ordinary providers, so the adapter makes no difference to them. A worker started with createApplicationContext() has no HTTP adapter at all. The only place the adapter appears is in an API that enqueues from a controller, and there the controller code is the same under Express 5 and Fastify 5.

When to move to a broker

Redis queues cover a large part of "background jobs", and they sit between in-process events and a full message broker. Move to a broker such as Kafka or RabbitMQ when you need event streams with replay, many independent consumer groups, routing by topic or key, or delivery guarantees across teams and languages. The Messaging Reference explains the models; Messaging Models and Patterns, Delivery Guarantees and Kafka vs. RabbitMQ help you choose, and Messaging Patterns in Practice covers patterns such as the outbox. Nest’s microservice transports are the matching client side. An event you emit "to another service" with @nestjs/event-emitter never leaves the process, so the broker or the queue, not the emitter, is the tool for that job.

Testing events, schedules and queues

What each part needs in a test, with Vitest (or Jest) and Test.createTestingModule:

  • Events: everything runs in process against the real @nestjs/event-emitter: listener order, the prepended listener, wildcard (book. and *) delivery, the logged (suppressed) listener error, the asynchronous listener, emitAsync() results and rejections, and the onModuleInit() versus waitUntilReady() behaviour.

  • Scheduler: real timers work for @Cron(), @Interval(), @Timeout(), waitForCompletion, disabled, SchedulerRegistry (add, start, stop, delete, list), cronJobs: false, the double-forRoot() duplication and the invalid time-zone error. The @nestjs/locks decorators run on the in-memory store.

  • BullMQ: without a Redis server you can still check the build, the dependency-injection wiring, the worker options passed from @Processor(), the @OnWorkerEvent() registrations, producer calls, the consumer’s process() and the shutdown close, by registering in-memory replacements for Queue, Worker and FlowProducer, which @nestjs/bullmq allows through static class setters on BullModule. Behaviour that only Redis can show (retries, delays, priorities, flow ordering, queue events, rate limiting) is described on this page from the official and BullMQ documentation. Run your own integration tests against Redis before depending on it.

What changed recently

  • NestJS 12 packages are ESM-only. The @nestjs/event-emitter, @nestjs/schedule and @nestjs/bullmq 12.x releases all declare Nest 11 or 12 as peers, and @nestjs/bullmq accepts BullMQ 3 to 6. The examples use the .js import extensions the ESM template needs.

  • BullMQ 6 makes the Redis driver optional: install ioredis (or pass a redis client). Its package metadata also lists pg as an optional peer for a PostgreSQL backend, which neither the Nest documentation nor this page covers.

  • Queue.add() no longer takes repeat; use upsertJobScheduler().

  • The current queues documentation covers @Processor() with worker options, @QueueEventsListener(), BullRegistrar with manual registration, and extraOptions on forRoot().

  • The current scheduling documentation covers waitForCompletion and disabled, running on several instances with @OnOneInstance() from the pre-1.0 @nestjs/locks, cron-silence monitoring with NestJS Observe, and a warning against calling ScheduleModule.forRoot() in more than one module.

  • The current event-emitter documentation covers EventEmitterReadinessWatcher, the inheritRequestContextId option and suppressErrors.

Source: NestJS documentation: Migration guide.

Where the book is outdated

The reference book, published in January 2025 (see the section bibliography), uses the event emitter to notify other microservices. That is a misuse: @nestjs/event-emitter is in-process only, so an event never reaches another service; use a microservice transport, a broker or a queue. It also predates several features described above: BullMQ job schedulers replacing the repeat option, the @Processor() worker options, @QueueEventsListener(), EventEmitterReadinessWatcher, waitForCompletion and @OnOneInstance(). Verify any import path or decorator from the book against the package typings before copying it.

References