⚙️ Bull Queue Implementation

Setup: Redis & BullMQ

BullMQ stores its queues in Redis, so you need a Redis server running. In development the quickest path is Docker; in production you'll point at a managed Redis (or a hosted service). Then install the library.

// Start Redis locally with Docker (one line, no install):
//   docker run -d --name redis -p 6379:6379 redis:7

// Install BullMQ (v5 is the current major line):
//   npm install bullmq

BullMQ (the modern successor to the original "Bull") splits responsibilities into small, focused classes. Three are the heart of everything:

Because a connection config is needed everywhere, define it once and import it. This keeps every part of the system pointed at the same Redis.

// queues/connection.js
// One place to configure how everything reaches Redis.
export const connection = {
  host: process.env.REDIS_HOST || '127.0.0.1',
  port: Number(process.env.REDIS_PORT) || 6379,
  // password: process.env.REDIS_PASSWORD,  // for managed Redis
};

Creating a Queue & Adding Jobs

A Queue is your producer handle. You give it a name (a namespace in Redis) and the connection. Then you call add(jobName, data, options) to enqueue work. The data is any JSON-serializable object — keep it small.

// queues/emailQueue.js
import { Queue } from 'bullmq';
import { connection } from './connection.js';

// The queue name 'email' is shared by the producer and the worker.
export const emailQueue = new Queue('email', { connection });

Now enqueue from an Express route. Notice how little the handler does — create the record, add the job, respond. The email hasn't been sent yet, and that's the point.

// server.js  (the producer / API process)
import express from 'express';
import { emailQueue } from './queues/emailQueue.js';

const app = express();
app.use(express.json());

app.post('/signup', async (req, res) => {
  const user = await User.create(req.body);

  // Add a 'welcome' job. First arg = job name, second = payload.
  await emailQueue.add('welcome', {
    userId: user.id,
    to: user.email,
    template: 'welcome'
  });

  // Respond immediately — the worker will send the email later.
  res.status(201).json({ id: user.id });
});

app.listen(3000, () => console.log('API on http://localhost:3000'));

Job "welcome" (id: 1) → state: waiting
payload: { userId: 42, to: "ada@example.com", template: "welcome" }

The job now sits in the waiting list. Nothing runs until a worker connects and pulls it — which is exactly what we build next.

The Worker Process

A Worker connects to the same queue name and supplies a processor function — an async function that receives a job and does the work. When it returns, the job is completed; if it throws, the job is failed (and retried if attempts remain).

Critically, this runs as its own process, separate from the web server. That's what gives you independent scaling and stops a heavy job from blocking incoming HTTP requests.

// worker.js  (run with: node worker.js — a SEPARATE process)
import { Worker } from 'bullmq';
import { connection } from './queues/connection.js';
import { sendMail } from './services/mailer.js';

const worker = new Worker(
  'email',                          // must match the queue name
  async (job) => {                  // the processor function
    const { to, template } = job.data;
    console.log(`Sending "${template}" to ${to}…`);

    // The actual slow work. If it throws, BullMQ handles the retry.
    const info = await sendMail({ to, template });

    // Whatever you return is stored as the job's result.
    return { messageId: info.messageId };
  },
  { connection, concurrency: 5 }    // run up to 5 jobs at once
);

console.log('Email worker started, waiting for jobs…');

Concurrency

The concurrency option is how many jobs a single worker runs in parallel. Because email is I/O-bound (mostly waiting on the network), a concurrency of 5–50 is fine — one worker can juggle many in-flight sends. CPU-bound work (image processing) should use low concurrency and instead scale by running more worker processes.

Reporting progress

Long jobs can report a percentage so a UI can show a progress bar. Call job.updateProgress() inside the processor.

const worker = new Worker('image', async (job) => {
  const { sizes } = job.data;
  let done = 0;

  for (const size of sizes) {
    await resizeTo(size);
    done++;
    // Report progress as a percentage (0–100).
    await job.updateProgress(Math.round((done / sizes.length) * 100));
  }
  return { count: sizes.length };
}, { connection });

Job Options: Retries, Backoff & Delay

The third argument to add() is where the reliability magic lives. These options ride along with the job in Redis and control how BullMQ retries and schedules it.

await emailQueue.add('welcome', payload, {
  attempts: 5,                    // try up to 5 times before failing for good
  backoff: {
    type: 'exponential',          // wait 2s, 4s, 8s, 16s between attempts
    delay: 2000                   // base delay in ms
  },
  delay: 10_000,                  // don't run until 10s from now (delayed state)
  priority: 1,                    // lower number = higher priority
  removeOnComplete: 1000,         // keep only the last 1000 completed jobs
  removeOnFail: 5000              // keep the last 5000 failed jobs for inspection
});

Setting shared defaults

Rather than repeating options at every add(), set defaultJobOptions on the queue itself.

export const emailQueue = new Queue('email', {
  connection,
  defaultJobOptions: {
    attempts: 3,
    backoff: { type: 'exponential', delay: 1000 },
    removeOnComplete: 500,
    removeOnFail: 2000
  }
});

Events & Progress

You have two ways to observe what's happening, and they serve different purposes.

Worker-local listeners

The Worker emits events for the jobs it processed. Perfect for logging inside the worker process.

worker.on('completed', (job, result) => {
  console.log(`Job ${job.id} completed →`, result);
});

worker.on('failed', (job, err) => {
  console.error(`Job ${job?.id} failed:`, err.message);
});

worker.on('progress', (job, progress) => {
  console.log(`Job ${job.id} is ${progress}% done`);
});

QueueEvents — the global stream

QueueEvents subscribes to all lifecycle events for a queue, from any process. Use it in your API or a dashboard to react to jobs a different process ran — for example, to push a WebSocket update when an image finishes.

// Can live in the API process, watching work done by the worker process.
import { QueueEvents } from 'bullmq';
import { connection } from './queues/connection.js';

const queueEvents = new QueueEvents('email', { connection });

queueEvents.on('completed', ({ jobId, returnvalue }) => {
  console.log(`[events] job ${jobId} done →`, returnvalue);
});

queueEvents.on('failed', ({ jobId, failedReason }) => {
  console.log(`[events] job ${jobId} failed: ${failedReason}`);
});

queueEvents.on('progress', ({ jobId, data }) => {
  console.log(`[events] job ${jobId} progress: ${data}%`);
});

Sending "welcome" to ada@example.com…
Job 1 completed → { messageId: '<abc@mail>' }
[events] job 1 done → { messageId: '<abc@mail>' }

Handling Failures

Failures are normal, not exceptional — mail servers time out, APIs return 500s. BullMQ's job is to make failures recoverable. When your processor throws, the job is retried per its attempts/backoff until it either succeeds or exhausts its attempts and lands in the failed set.

const worker = new Worker('email', async (job) => {
  try {
    return await sendMail(job.data);
  } catch (err) {
    // Distinguish transient from permanent failures.
    if (err.code === 'EInvalidAddress') {
      // Permanent — don't waste retries. Throw a marker you can detect,
      // or discard by returning; here we rethrow so it's recorded as failed.
      throw new Error(`Undeliverable address: ${job.data.to}`);
    }
    // Transient (timeout, 5xx) — rethrow so BullMQ retries with backoff.
    throw err;
  }
}, { connection });

Inspecting failed jobs

Because you set removeOnFail to keep history, you can list and examine what went wrong — essential for debugging and for building a retry-by-hand admin action.

// A small script or admin endpoint to review the dead-letter set.
const failed = await emailQueue.getJobs(['failed'], 0, 20);

for (const job of failed) {
  console.log({
    id: job.id,
    name: job.name,
    reason: job.failedReason,
    attemptsMade: job.attemptsMade,
    data: job.data
  });
  // await job.retry();   // re-queue a job you believe will now succeed
}

Graceful shutdown

When you deploy, don't kill a worker mid-job. Close it cleanly so the in-flight job finishes (or is released back to waiting).

process.on('SIGTERM', async () => {
  console.log('Shutting down worker…');
  await worker.close();   // stop taking new jobs, finish current ones
  process.exit(0);
});

Practice & Quiz

import { Queue } from 'bullmq';
import { connection } from './queues/connection.js';

export const notificationsQueue = new Queue('notifications', {
  connection,
  defaultJobOptions: {
    attempts: 4,
    backoff: { type: 'exponential', delay: 3000 },
    removeOnComplete: 100,
    removeOnFail: 1000
  }
});

await notificationsQueue.add(
  'push',
  { userId: 42, message: 'Your report is ready' },
  { delay: 30_000 }   // held in "delayed" for 30 seconds
);

import { Worker } from 'bullmq';
import { connection } from './queues/connection.js';
import { deliver } from './services/notifier.js';

const worker = new Worker(
  'notifications',
  async (job) => {
    return await deliver(job.data);   // throw on failure → BullMQ retries
  },
  { connection, concurrency: 10 }
);

worker.on('completed', (job) => console.log(`✓ ${job.id} delivered`));
worker.on('failed', (job, err) =>
  console.error(`✗ ${job?.id} failed (attempt ${job?.attemptsMade}):`, err.message)
);

Best Practices & Pitfalls

✅ Do

• Share one connection config module across queue, worker, and events
• Set removeOnComplete and removeOnFail on every queue
• Match concurrency to the work: high for I/O-bound, low + more processes for CPU-bound
• Throw on transient errors so BullMQ retries; keep processors idempotent
• Close workers with worker.close() on SIGTERM for zero-downtime deploys

❌ Don't

• Don't create a new Queue per request — instantiate once and reuse
• Don't run the worker in the same process as your web server for real workloads
• Don't stuff large blobs into job.data — pass an id and fetch inside the job
• Don't ignore the failed set — monitor it and alert on growth

Summary

📚 Additional Resources

• BullMQ — Queues guide
• BullMQ — Workers & concurrency
• BullMQ — Jobs, options & retries
• BullMQ — QueueEvents API reference

🚀 What's Next?

You can now run work on demand. But a lot of background work runs on a clock — a nightly cleanup, a Monday-morning report. In the next lesson, Scheduled Tasks, you'll learn cron syntax and schedule recurring jobs with both node-cron and BullMQ's built-in repeatable jobs.