Posted inProgramming

Building a Distributed Task Queue with BullMQ, Redis, and TypeScript

Programming tutorial - IT technology blog
Programming tutorial - IT technology blog

Comparing Background Processing Approaches in Node.js

When an API endpoint handles heavy work—such as generating 50MB PDFs, transcoding 4K video clips, or syncing thousands of records with external webhooks—HTTP response times quickly degrade. Offloading these tasks from the request-response cycle is critical. In production Node.js systems, teams typically choose from three main patterns:

  • In-Memory Micro-Queues (e.g., fastq, EventEmitter): Quickest to implement with zero external dependencies. However, if your Node.js process restarts during a deployment, every queued job in memory vanishes. It also cannot distribute work across multiple server instances.
  • Heavy Enterprise Brokers (e.g., Apache Kafka, RabbitMQ): Excellent for streaming millions of events per second and managing complex message topologies. The downside is significant operational overhead, a steep learning curve, and complex setups for simple task scheduling or delayed retries.
  • Redis-Backed Distributed Queues (BullMQ): The sweet spot for modern TypeScript and Node.js microservices. BullMQ uses Redis streams, sorted sets, and atomic Lua scripts to handle scheduling, retries, and rate limiting across distributed workers without heavy infrastructure baggage.

Pros and Cons of the BullMQ + Redis Architecture

Before standardizing on BullMQ, evaluate how its architecture aligns with your infrastructure constraints:

The Advantages

  • Native Delayed & Cron Jobs: Schedule a task to run after 15 minutes or recur every night at midnight UTC without maintaining external cron daemons.
  • Atomic State Transitions: BullMQ runs Lua scripts directly inside Redis. This guarantees atomic status updates from waiting to active, completed, or failed with zero race conditions.
  • Strict TypeScript Typing: Generics for job data and return types catch schema mismatches between producers and consumers at compile time.
  • Granular Concurrency & Rate Limiting: Cap worker concurrency and throttle overall throughput (e.g., 50 requests per second across 10 pods) to avoid hitting third-party API rate limits.

The Tradeoffs

  • Redis Memory Footprint: Job payloads and execution metadata live in RAM. Processing 500,000 jobs daily without strict retention policies can exhaust Redis memory in a few days.
  • Dedicated Connection Overhead: BullMQ workers use blocking Redis commands to pull jobs instantly. Each queue and worker instance maintains multiple dedicated TCP connections.

Recommended Architecture and Production Setup

A reliable task processing system splits work into two distinct roles: Producers (API servers pushing tasks) and Consumers (dedicated worker processes executing tasks). Both communicate through a shared Redis cluster.

Here is the baseline pattern for a production pipeline:

  1. Primary Queue: Receives standard jobs with configured exponential backoff (e.g., initial 2s delay doubling on each attempt).
  2. Dead-Letter Queue (DLQ): Captures payloads that fail all retry attempts, allowing engineers to inspect stack traces and replay jobs without blocking normal traffic.
  3. Connection Isolation: Configures ioredis with maxRetriesPerRequest: null as required by BullMQ to allow long-lived blocking commands.

Implementation Guide

1. Project Setup and Connection Configuration

Initialize a TypeScript project and install the required dependencies:

npm install bullmq ioredis dotenv
npm install -D typescript @types/node tsx

Configure the shared Redis connection in src/config/redis.ts:

import { ConnectionOptions } from 'bullmq';

export const redisConnection: ConnectionOptions = {
  host: process.env.REDIS_HOST || '127.0.0.1',
  port: Number(process.env.REDIS_PORT) || 6379,
  password: process.env.REDIS_PASSWORD || undefined,
  maxRetriesPerRequest: null,
  enableReadyCheck: false,
};

2. Defining Job Types and the Producer

Define clear payload interfaces in src/types/queue.ts to enforce contract safety:

export interface EmailJobData {
  to: string;
  subject: string;
  templateId: string;
  variables: Record<string, string>;
}

export interface EmailJobResult {
  messageId: string;
  sentAt: string;
}

export const EMAIL_QUEUE_NAME = 'email-delivery';
export const EMAIL_DLQ_NAME = 'email-delivery-dlq';

Instantiate the queue producer in src/queues/emailQueue.ts:

import { Queue } from 'bullmq';
import { redisConnection } from '../config/redis';
import { EMAIL_QUEUE_NAME, EmailJobData, EmailJobResult } from '../types/queue';

export const emailQueue = new Queue<EmailJobData, EmailJobResult>(
  EMAIL_QUEUE_NAME,
  {
    connection: redisConnection,
    defaultJobOptions: {
      attempts: 5,
      backoff: {
        type: 'exponential',
        delay: 2000, // Retries at 2s, 4s, 8s, 16s, 32s
      },
      removeOnComplete: {
        age: 3600, // Keep completed job metadata for 1 hour
        count: 1000,
      },
      removeOnFail: false, // Retain failed jobs for inspection
    },
  }
);

export async function scheduleEmail(data: EmailJobData, delayMs: number = 0) {
  const job = await emailQueue.add('send-transactional-email', data, {
    delay: delayMs,
  });
  return job.id;
}

3. Building the Worker with Dead-Letter Routing

When external APIs encounter temporary network hiccups, BullMQ retries the task automatically using exponential backoff. If all 5 attempts fail, we capture the failure context and push it directly into a Dead-Letter Queue for inspection and alerting.

Create the consumer worker in src/workers/emailWorker.ts:

import { Worker, Queue, Job } from 'bullmq';
import { redisConnection } from '../config/redis';
import {
  EMAIL_QUEUE_NAME,
  EMAIL_DLQ_NAME,
  EmailJobData,
  EmailJobResult,
} from '../types/queue';

const deadLetterQueue = new Queue(EMAIL_DLQ_NAME, { connection: redisConnection });

export const emailWorker = new Worker<EmailJobData, EmailJobResult>(
  EMAIL_QUEUE_NAME,
  async (job: Job<EmailJobData, EmailJobResult>) => {
    // Simulate downstream API failure
    if (job.data.to.includes('fail')) {
      throw new Error(`Upstream SMTP provider timeout for ${job.data.to}`);
    }

    return {
      messageId: `msg_${job.id}_${Date.now()}`,
      sentAt: new Date().toISOString(),
    };
  },
  {
    connection: redisConnection,
    concurrency: 10, // Process 10 jobs concurrently per worker instance
    limiter: {
      max: 100,
      duration: 1000, // Cap throughput at 100 jobs per second per worker
    },
  }
);

emailWorker.on('completed', (job: Job) => {
  console.log(`[Job ${job.id}] Successfully delivered to ${job.data.to}`);
});

emailWorker.on('failed', async (job: Job | undefined, error: Error) => {
  if (!job) return;

  console.warn(`[Job ${job.id}] Attempt ${job.attemptsMade}/${job.opts.attempts} failed: ${error.message}`);

  // Route to DLQ once all attempts are exhausted
  if (job.attemptsMade >= (job.opts.attempts || 1)) {
    console.error(`[Job ${job.id}] Retries exhausted. Routing to DLQ: ${EMAIL_DLQ_NAME}`);
    await deadLetterQueue.add('failed-email-poison-pill', {
      originalJobId: job.id,
      originalData: job.data,
      failedReason: error.message,
      failedAt: new Date().toISOString(),
      stacktrace: error.stack,
    });
  }
});

4. Graceful Shutdown & Signal Handling

Stopping a worker abruptly while processing tasks can leave jobs orphaned or corrupt data. Always intercept system termination signals (SIGTERM, SIGINT) in your entry file (src/index.ts) to allow active jobs to finish:

import { emailWorker } from './workers/emailWorker';
import { emailQueue } from './queues/emailQueue';

async function shutdown(signal: string) {
  console.log(`Received ${signal}. Draining active tasks and closing connections...`);
  
  // Finish active jobs and reject new tasks
  await emailWorker.close();
  await emailQueue.close();
  
  console.log('Task queue shutdown complete. Exiting cleanly.');
  process.exit(0);
}

process.on('SIGTERM', () => shutdown('SIGTERM'));
process.on('SIGINT', () => shutdown('SIGINT'));

Best Practices for High-Volume Production Systems

  • Keep Payloads Lightweight: Pass IDs or object storage URLs (e.g., S3 paths) rather than embedding 10MB file buffers in the job data.
  • Ensure Idempotent Handlers: Transient network errors trigger retries, meaning a job might execute twice. Use database transaction IDs or UPSERT operations to avoid duplicate charges or duplicate emails.
  • Automate Redis Memory Trimming: Always specify removeOnComplete and removeOnFail limits. Set up cluster alerts whenever Redis RAM usage crosses 75% capacity.
  • Monitor Queue Depth and Latency: Track waiting and active job counts via queue.getJobCounts() in Prometheus or Datadog. Scale worker pods dynamically if wait times exceed 5 seconds.
Share:
Tags: