BullMQ scheduled tasks: queues, retries, and Job Schedulers

BullMQ moves work from a Node.js request into a queue that workers can process independently. Queue adds and manages jobs, Worker executes them, and QueueEvents observes queue-wide events through Redis. Scheduled jobs, delayed execution, retries, and concurrency controls build on that model.

Redis stores queue state, but durability still depends on persistence, replication, failover, and recovery configuration. A worker crash or a lost completion acknowledgment can cause the same work to execute again. Design business side effects to tolerate that possibility.

Start with a queue and worker

This guide uses the BullMQ 5.x API, including Job Schedulers introduced in 5.16. Use a compatible Node.js and Redis release and lock the dependencies for your application:

npm install bullmq@5 ioredis

With an isolated local Redis instance listening on port 6379, save this as email-demo.mjs. It processes a demonstration job; it does not send an email:

import { Queue, QueueEvents, Worker } from 'bullmq';

const connection = { host: '127.0.0.1', port: 6379 };
const queue = new Queue('email-demo', { connection });
const events = new QueueEvents('email-demo', { connection });
const worker = new Worker('email-demo', async job => {
  console.log('Processing welcome job', { jobId: job.id, userId: job.data.userId });
  return { processedUserId: job.data.userId };
}, { connection, concurrency: 5 });

for (const component of [queue, events, worker]) {
  component.on('error', error => console.error('Queue component error', error));
}

try {
  await events.waitUntilReady();
  const job = await queue.add('welcome', { userId: 123 }, {
    removeOnComplete: 100,
    removeOnFail: 500,
  });
  const result = await job.waitUntilFinished(events, 10_000);
  console.log('Completed', result);
} finally {
  await worker.close();
  await events.close();
  await queue.close();
}

The producer and worker share one process only to keep the demonstration small. In a service deployment they can run separately and coordinate through the same queue. Configure Redis authentication, transport, network access, and connection retry behavior for that environment.

A real email processor could perform the source application's work like this:

// Application integration fragment: db and sendWelcomeEmail are your services.
const user = await db.user.find(job.data.userId);
if (!user) throw new Error('User no longer exists');
await sendWelcomeEmail(user.email);

Pass stable IDs and JSON-serializable data in job.data; a database model instance, open socket, or function will not retain its runtime behavior through serialization. Use an idempotent operation key or a provider-supported deduplication contract for the actual send. Marking a queue job complete and an external provider accepting a message are not automatically one transaction.

Register recurring and delayed work

Create the queue before registering its scheduler:

import { Queue } from 'bullmq';

const backupQueue = new Queue('backup', {
  connection: { host: '127.0.0.1', port: 6379 },
});
try {
  await backupQueue.upsertJobScheduler(
    'daily-backup',
    { pattern: '0 2 * * *', tz: 'Asia/Shanghai' },
    { name: 'daily-backup', data: {}, opts: {
      removeOnComplete: 100,
      removeOnFail: 500,
    } },
  );
} finally {
  await backupQueue.close();
}

This defines a daily 02:00 schedule in Asia/Shanghai. A running worker for the backup queue still needs to implement the backup operation. Using the same scheduler ID updates its definition without a separate remove/add gap. Follow Job Scheduler documentation for the deployed version, especially when migrating from older repeatable-job APIs.

A scheduler produces jobs as earlier scheduled jobs begin processing. Busy or unavailable workers can therefore affect the observed cadence; do not interpret the definition as a strict real-time execution guarantee or assume every missed interval will become a catch-up job.

For a one-time delay, use a queue instance that is still open:

await queue.add('reminder', { userId: 123 }, {
  delay: 10 * 60 * 1000,
});

// runAt is a validated Date supplied by the application.
const runAtMs = runAt.getTime();
if (!Number.isFinite(runAtMs)) throw new Error('Invalid scheduled time');
await queue.add('report', { reportId: 'report-42' }, {
  delay: Math.max(0, runAtMs - Date.now()),
});

This policy makes a past target time eligible immediately. Choose a different explicit policy if an overdue report should be rejected. Delay establishes eligibility, not an exact start time; queue load and worker availability still matter.

Configure retries, priority, and job identity

The following fragment adds a retryable API synchronization job:

await queue.add('api-sync', { accountId: 'acct-42' }, {
  attempts: 5,
  backoff: { type: 'exponential', delay: 3000 },
  priority: 10,
  jobId: 'sync-acct-42',
  removeOnComplete: 100,
  removeOnFail: 500,
});

attempts: 5 allows up to five processing attempts including the initial one. Exponential backoff spaces retries for transient failures; it does not repair invalid input or make a non-idempotent operation safe to repeat.

For positive priorities, a lower number means higher priority. Jobs without a positive priority are treated differently from the prioritized set and can run first, so assigning 10 does not necessarily outrank all ordinary jobs. Verify ordering with the combination used by your application. See prioritized jobs.

A custom jobId prevents another job with the same ID from being added to the same queue while the original record exists. Once automatic cleanup or an explicit deletion removes that record, the ID can be added again. Custom job IDs must not contain :. This is queue-record deduplication, not durable business idempotency. See job IDs.

Completed and failed history consume Redis memory. Choose count/age retention and monitor actual growth, including jobs still waiting or delayed. Automatic history cleanup is not an audit archive; export the necessary business or operational records separately.

Control concurrency, rate, and shutdown

A worker's concurrency controls how many jobs that worker can process concurrently. Several workers increase aggregate concurrency. A BullMQ rate limiter controls the queue's job-start rate across participating workers, rather than giving every instance an independent allowance:

const worker = new Worker('api-sync', processApiSync, {
  connection,
  concurrency: 5,
  limiter: { max: 10, duration: 1000 },
});

Here processApiSync and connection are application-provided. Ten job starts per second is not necessarily ten HTTP requests per second if a job makes several requests or retries inside its processor. Align the limit with the downstream provider's contract. See rate limiting.

For a long-running worker process, close resources during graceful shutdown:

let stopping = false;
async function shutdown() {
  if (stopping) return;
  stopping = true;
  try {
    await worker.close();
    await events.close();
    await queue.close();
  } catch (error) {
    console.error('Queue shutdown failed', error);
    process.exitCode = 1;
  }
}
process.on('SIGTERM', shutdown);
process.on('SIGINT', shutdown);

This fragment assumes that this process owns the three objects; close only the resources it actually creates. worker.close() stops fetching new work and waits for active jobs. It does not impose an application deadline or guarantee completion before a container's termination grace period expires. Give processors bounded I/O and enough shutdown time, and close any separately owned database or Redis clients too.

A job becomes stalled when the worker cannot maintain its processing lock, for example after a crash or a blocked event loop. This is not determined by whether application code calls job.updateProgress(). Stalled jobs may return to waiting or fail after the configured limit. CPU-heavy work may need sandboxed processors or another execution design. See stalled jobs.

Protect the queue dashboard

Bull Board can expose queue status and administrative actions. Install the API and adapter packages alongside Express:

npm install @bull-board/api @bull-board/express express

An ES module integration can mount it behind existing administrative authorization:

import { createBullBoard } from '@bull-board/api';
import { BullMQAdapter } from '@bull-board/api/bullMQAdapter';
import { ExpressAdapter } from '@bull-board/express';

// app, requireAdmin, and emailQueue belong to the host application.
const serverAdapter = new ExpressAdapter();
serverAdapter.setBasePath('/admin/queues');
createBullBoard({
  queues: [new BullMQAdapter(emailQueue)],
  serverAdapter,
});
app.use('/admin/queues', requireAdmin, serverAdapter.getRouter());

The imported components do not implement requireAdmin for you. Apply authentication, authorization, and appropriate network access before the dashboard router. Consider read-only access for operators who should inspect jobs but not retry or delete them. See the Bull Board project.

Measure waiting separately from execution

A healthy Redis connection does not prove jobs start promptly. Report queue depth, oldest waiting age, enqueue-to-start delay, execution duration, failures, and retries. Include queue name, job_id, attempt, and outcome in application logs, while limiting sensitive job payloads.

Guance's Redis integration can show connection and memory pressure. BullMQ-specific backlog and wait metrics need BullMQ APIs, its supported telemetry, or your own instrumentation. Collect worker records through DataKit logging and define alerts around the queue's service objective.

For distributed tracing, propagate the enqueue context into the job and restore or link it during processing through compatible instrumentation. A Node.js HTTP tracing setup alone does not establish queue-span continuity. See the Node.js tracing guide.

Common questions

Should I choose BullMQ or Agenda?

An existing, well-operated Redis deployment can make BullMQ a natural fit. Agenda may suit an application that already uses MongoDB and wants that persistence model. Compare scheduler behavior, retries, operations, and integration requirements; neither an existing database nor a popularity claim alone decides the choice.

What happens if a worker crashes mid-job?

Its lock can expire and the stalled-job mechanism can make the work eligible again, subject to the stall limit. External side effects may already have happened. Test that retrying a partially completed job produces the intended business result.

Where should recurring jobs be registered?

Use a controlled startup or deployment step with stable scheduler IDs. Idempotent upserts avoid separate remove/add races, but replicas applying different definitions can still overwrite the desired schedule. Keep one agreed configuration and migrate legacy repeat APIs deliberately.

The Express and Pug guide and bun install guide are currently in Chinese. This guide describes a small queue workflow, not a production durability or throughput benchmark.