Message Queues & Background Jobs

What are Message Queues?

A message queue is a middleware component that allows services to communicate asynchronously by passing messages between producers and consumers. Instead of a service calling another service directly and waiting for a response, it pushes a message onto a queue and continues processing. A consumer picks up the message when ready.

Why Use Queues?

Synchronous request-response chains create tight coupling and bottleneck at the slowest service. If your API calls a payment processor, sends an email, and generates a PDF synchronously, the user waits for all three. Queues decouple these concerns:

• Temporal decoupling: Producer and consumer don't need to be online simultaneously
• Load leveling: Absorb traffic spikes — the queue buffers requests and processes them at a manageable rate
• Fault isolation: If the email service is down, other operations continue unaffected
• Scalability: Add more consumers horizontally to increase throughput without changing the producer

// Synchronous (bad) — user waits for all operations
app.post('/api/orders', async (req, res) => {
  const order = await createOrder(req.body);
  await sendConfirmationEmail(order);    // 200ms
  await chargePayment(order);             // 500ms
  await generateInvoice(order);           // 300ms
  res.json(order); // User waits ~1000ms
});

// Asynchronous (good) — user gets immediate response
app.post('/api/orders', async (req, res) => {
  const order = await createOrder(req.body);
  await queue.add('send-email', { orderId: order.id });
  await queue.add('charge-payment', { orderId: order.id });
  await queue.add('generate-invoice', { orderId: order.id });
  res.json(order); // User waits ~50ms
});

Queue Topologies

BullMQ vs Bull

BullMQ is the successor to Bull, rewritten with a focus on TypeScript, Redis Cluster support, and better performance. Bull is deprecated but still widely used. For new projects, always use BullMQ.

// BullMQ setup
import { Queue, Worker } from 'bullmq';
import Redis from 'ioredis';

const connection = new Redis({
  host: process.env.REDIS_HOST || '127.0.0.1',
  port: 6379,
  maxRetriesPerRequest: null, // Required by BullMQ
});

const emailQueue = new Queue('emails', { connection });
const paymentQueue = new Queue('payments', { connection });

Setting Up Bull/BullMQ

BullMQ requires a Redis instance as its backing store. Every queue, job, and state is persisted in Redis, which means queues survive application restarts and work across multiple processes.

Job Types

BullMQ supports several job types to cover different scheduling needs:

• Regular jobs: Added and processed immediately in FIFO order
• Delayed jobs: Processed after a specified delay period
• Repeatable jobs: Automatically re-added on a cron schedule or fixed interval
• Rate-limited jobs: Process at most N jobs within a time window
• Priority jobs: Higher priority jobs are processed before lower ones

import { Queue, QueueScheduler } from 'bullmq';

const connection = new Redis({ host: '127.0.0.1', port: 6379, maxRetriesPerRequest: null });
const emailQueue = new Queue('emails', { connection });

// 1. Regular job — processed immediately
await emailQueue.add('send-welcome', {
  userId: 42,
  template: 'welcome',
  to: 'user@example.com',
});

// 2. Delayed job — processed after 5 minutes (in milliseconds)
await emailQueue.add('send-followup', {
  userId: 42,
  template: 'followup',
}, {
  delay: 5 * 60 * 1000, // 5 minutes
});

// 3. Repeatable job — runs every day at 9 AM
await emailQueue.add('send-daily-digest', {
  type: 'daily-digest',
}, {
  repeat: {
    cron: '0 9 * * *',
    tz: 'America/New_York',
  },
});

// 4. Rate-limited — at most 5 jobs per minute
await emailQueue.addBulk(
  recipients.map((email, i) => ({
    name: 'send-newsletter',
    data: { to: email, campaignId: 123 },
    opts: {
      jobId: `newsletter-${campaignId}-${i}`,
    },
  })),
);

// 5. Priority — urgent emails processed first
await emailQueue.add('send-notification', {
  to: 'admin@example.com',
  message: 'Server alert',
}, {
  priority: 1, // Lower number = higher priority
});
await emailQueue.add('send-receipt', {
  to: 'user@example.com',
  orderId: 456,
}, {
  priority: 10, // Lower priority
});

Global Rate Limiting with QueueScheduler

To enforce rate limits across all producers (not just per-add), use a QueueScheduler:

import { QueueScheduler } from 'bullmq';

const scheduler = new QueueScheduler('emails', { connection });
await scheduler.waitUntilReady();

const emailQueue = new Queue('emails', {
  connection,
  limiter: {
    max: 100,         // max jobs
    duration: 60_000,  // per 60 seconds
  },
});

// All adds are now rate-limited globally
for (let i = 0; i < 500; i++) {
  await emailQueue.add('send', { to: `user${i}@example.com` });
}

Job Options Summary

Workers: The Consumer Side

A Worker connects to Redis, pulls jobs from the queue, and executes them. Workers run in separate processes or threads, keeping the main application responsive.

Basic Worker Setup

import { Worker } from 'bullmq';

const emailWorker = new Worker('emails', async (job) => {
  console.log(`Processing email job ${job.id}: ${job.name}`);
  console.log(`To: ${job.data.to}, Template: ${job.data.template}`);

  // Simulate email sending
  await sendEmail({
    to: job.data.to,
    template: job.data.template,
    data: job.data.payload,
  });

  // Return value is stored as job.returnvalue
  return { sent: true, messageId: 'abc-123' };
}, {
  connection: new Redis({ host: '127.0.0.1', port: 6379, maxRetriesPerRequest: null }),
  concurrency: 5,       // Process 5 jobs simultaneously
  limiter: {
    max: 10,            // Max 10 jobs...
    duration: 10_000,   // ...per 10 seconds
  },
});

// Event handlers
emailWorker.on('completed', (job, result) => {
  console.log(`Job ${job.id} completed:`, result);
});

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

emailWorker.on('stalled', (jobId) => {
  console.warn(`Job ${jobId} stalled — possibly crashed during processing`);
});

Retries and Backoff

Transient failures (network timeouts, rate limits) are expected. BullMQ retries jobs with configurable backoff strategies:

const orderWorker = new Worker('orders', async (job) => {
  const result = await chargePayment(job.data.paymentMethodId, job.data.amount);
  return result;
}, {
  connection,
  concurrency: 3,
});

// Jobs are added with retry configuration
await ordersQueue.add('charge', {
  paymentMethodId: 'pm_card_visa',
  amount: 9999,
}, {
  attempts: 4, // Retry up to 4 times total (original + 3 retries)
  backoff: {
    type: 'exponential', // Doubling delay: 2s, 4s, 8s
    delay: 2000,         // Initial delay in ms
  },
});

// Alternative backoff types:
// { type: 'fixed', delay: 5000 }      — Always wait 5 seconds
// { type: 'exponential', delay: 1000 } — 1s, 2s, 4s, 8s...
// Custom function for advanced scenarios:
backoff: {
  type: 'custom',
  delay: (attemptsMade, err) => {
    // Custom logic: longer delay for payment errors
    if (err.code === 'RATE_LIMITED') return 30_000;
    return Math.min(1000 * 2 ** attemptsMade, 60_000);
  },
}

Error Handling and Dead Letter Queues

When a job fails all retry attempts, it moves to the failed state. A Dead Letter Queue (DLQ) captures these failed jobs for later inspection:

import { QueueEvents } from 'bullmq';

const dlq = new Queue('dead-letter', { connection });
const queueEvents = new QueueEvents('orders', { connection });

queueEvents.on('failed', async ({ jobId, failedReason, prev }) => {
  // 'prev' is the state before failure (e.g., 'active', 'waiting')
  const job = await ordersQueue.getJob(jobId);
  if (job && job.attemptsMade >= (job.opts.attempts || 1)) {
    // All retries exhausted — move to DLQ
    await dlq.add('failed-order', {
      originalQueue: 'orders',
      jobId,
      data: job.data,
      failedReason,
      attemptsMade: job.attemptsMade,
      timestamp: Date.now(),
    });
    console.error(`Job ${jobId} moved to DLQ after ${job.attemptsMade} attempts`);
  }
});

// DLQ consumer for manual inspection
const dlqWorker = new Worker('dead-letter', async (job) => {
  const { originalQueue, jobId, failedReason, data } = job.data;
  
  // Log to monitoring system
  await logToDatadog('job.dlq', {
    queue: originalQueue,
    jobId,
    error: failedReason,
    data,
  });

  // Optionally alert the team
  if (job.data.data.amount > 10000) {
    await sendSlackAlert(`High-value order failed: ${jobId}`, data);
  }
}, { connection });

Stalled Job Detection

A job is "stalled" when a worker picks it up but crashes before completing. BullMQ's stalledInterval checks for stalled jobs and re-enqueues them:

const worker = new Worker('orders', processor, {
  connection,
  lockDuration: 30_000,       // Max time to hold a lock on a job
  stalledInterval: 15_000,    // Check for stalled jobs every 15s
  maxStalledCount: 3,         // Fail after 3 stalls (prevents infinite loops)
});

Graceful Shutdown

Always close workers and queues cleanly to avoid orphaned jobs:

const gracefulShutdown = async () => {
  console.log('Shutting down gracefully...');
  await emailWorker.close();   // Stop accepting new jobs, finish current
  await emailQueue.close();    // Disconnect from Redis
  await connection.quit();     // Close Redis connection
  process.exit(0);
};

process.on('SIGTERM', gracefulShutdown);
process.on('SIGINT', gracefulShutdown);

Advanced Queue Patterns

Job Flow (Chaining Jobs)

BullMQ's FlowProducer lets you chain jobs into pipelines where each step depends on the previous:

import { FlowProducer } from 'bullmq';

const flow = new FlowProducer({ connection });

await flow.addBulk([
  {
    name: 'process-order',
    queueName: 'orders',
    data: { orderId: 123 },
    children: [
      {
        name: 'charge-payment',
        queueName: 'payments',
        data: { orderId: 123, amount: 9999 },
        children: [
          {
            name: 'send-receipt',
            queueName: 'emails',
            data: { orderId: 123, type: 'receipt' },
          },
        ],
      },
      {
        name: 'update-inventory',
        queueName: 'inventory',
        data: { orderId: 123 },
      },
    ],
  },
]);

// The flow tree:
// process-order
//   ├── charge-payment
//   │     └── send-receipt
//   └── update-inventory

Concurrency Strategies

Different job types need different concurrency levels. Use separate workers with distinct settings:

// High concurrency for lightweight tasks
const emailWorker = new Worker('emails', emailProcessor, {
  connection,
  concurrency: 20,  // Handle 20 emails simultaneously
});

// Low concurrency for resource-heavy tasks
const reportWorker = new Worker('reports', reportProcessor, {
  connection,
  concurrency: 3,   // Only 3 reports at a time
});

// CPU-intensive with rate limiting
const exportWorker = new Worker('exports', exportProcessor, {
  connection,
  concurrency: 2,
  limiter: {
    max: 5,
    duration: 60_000, // 5 exports per minute
  },
});

Job Progress and Logging

For long-running jobs, report progress to give visibility:

const bulkEmailWorker = new Worker('bulk-emails', async (job) => {
  const recipients = job.data.recipients; // e.g., 10,000 emails
  const batchSize = 100;

  for (let i = 0; i < recipients.length; i += batchSize) {
    const batch = recipients.slice(i, i + batchSize);
    await Promise.all(batch.map(email => sendEmail(email)));

    // Update progress (0-100)
    const progress = Math.round((i + batch.length) / recipients.length * 100);
    await job.updateProgress(progress);

    // Log progress details
    job.log(`Sent batch ${i / batchSize + 1}: ${batch.length} emails`);
  }

  return { totalSent: recipients.length };
}, { connection });

// Monitor progress from outside
bulkEmailWorker.on('progress', (job, progress) => {
  console.log(`Job ${job.id} is ${progress}% complete`);
});

Priority Queues

Priority is set per-job. Lower numbers are processed first:

const taskQueue = new Queue('tasks', { connection });

// Regular tasks
await taskQueue.add('process-data', { dataset: 'user-events' }, { priority: 10 });
await taskQueue.add('process-data', { dataset: 'page-views' }, { priority: 10 });

// Urgent task — jumps ahead in the queue
await taskQueue.add('process-data', { dataset: 'security-alert' }, { priority: 1 });

// Background task — processed last
await taskQueue.add('process-data', { dataset: 'analytics-historical' }, { priority: 100 });

Sandbox Worker for CPU-Intensive Jobs

BullMQ supports sandboxed workers using worker-fork to isolate CPU-heavy processing from the main process:

// processor.js — runs in a child process
module.exports = async function(job) {
  const result = await heavyComputation(job.data);
  return result;
};

// main.js
const worker = new Worker('cpu-jobs', './processor.js', {
  connection,
  concurrency: 4,
  // Use child_process.fork for isolation
});

Production Monitoring

Queues are infrastructure. Without monitoring, failures are invisible until users complain.

Bull Board Dashboard

Bull Board provides a web UI to inspect queues, retry failed jobs, and monitor job status:

import express from 'express';
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(emailQueue, { readOnlyMode: false }),
    new BullMQAdapter(ordersQueue, { readOnlyMode: false }),
    new BullMQAdapter(reportsQueue, { readOnlyMode: false }),
  ],
  serverAdapter,
});

const app = express();
app.use('/admin/queues', serverAdapter.getRouter());
// Protect with authentication
app.use('/admin/queues', authenticateAdmin);
app.listen(3000);

Metrics Collection

Track queue health with Prometheus or similar metrics systems:

import { collectDefaultMetrics, Counter, Histogram } from 'prom-client';

const jobsProcessed = new Counter({
  name: 'queue_jobs_processed_total',
  help: 'Total number of jobs processed',
  labelNames: ['queue', 'status'],
});

const jobDuration = new Histogram({
  name: 'queue_job_duration_seconds',
  help: 'Job processing duration in seconds',
  labelNames: ['queue', 'job_name'],
  buckets: [0.1, 0.5, 1, 2, 5, 10, 30],
});

const worker = new Worker('orders', async (job) => {
  const end = jobDuration.startTimer({ queue: 'orders', job_name: job.name });
  try {
    const result = await processOrder(job.data);
    jobsProcessed.inc({ queue: 'orders', status: 'success' });
    end();
    return result;
  } catch (err) {
    jobsProcessed.inc({ queue: 'orders', status: 'failure' });
    end();
    throw err;
  }
}, { connection });

// Redis metrics for queue depth
async function getQueueMetrics() {
  const waiting = await emailQueue.getWaitingCount();
  const active = await emailQueue.getActiveCount();
  const completed = await emailQueue.getCompletedCount();
  const failed = await emailQueue.getFailedCount();
  const delayed = await emailQueue.getDelayedCount();

  return { waiting, active, completed, failed, delayed };
}

Production Checklist

Idempotent Job Design

Jobs may be processed more than once (after a stall or crash). Design for idempotency:

const paymentWorker = new Worker('payments', async (job) => {
  const { orderId, paymentId } = job.data;

  // Check if already processed (idempotency key)
  const existing = await db.paymentResult.findOne({ paymentId });n  if (existing) {
    console.log(`Payment ${paymentId} already processed, skipping`);
    return existing;
  }

  const result = await chargePayment(job.data);
  await db.paymentResult.create({ paymentId, orderId, result });
  return result;
}, { connection });