·QueueHub Team·20 min read

BullMQ Worker Operations: Resource Sizing, Idempotency, Quarantine & Observability

BullMQWorker OperationsResource SizingIdempotencyPoison PillDead Letter QueueQuarantineObservabilityPrometheusContainer Isolation

You know how to spin up a BullMQ Worker, configure concurrency, attach event listeners, and handle retries. But production worker operations bring a different set of questions:

  • How many workers should I run, and at what concurrency, given my job duration, hardware, and queue depth target?
  • How do I guarantee a job is processed exactly once when BullMQ only promises at-least-once?
  • What happens when a corrupt job keeps failing — how do I detect and quarantine it automatically?
  • What per-worker metrics should I track to spot degradation before it becomes a fleet-wide problem?
  • Should all workers share one process, or do I need separate containers?

This post answers those questions with production-ready TypeScript patterns. Each section is self-contained — adopt only what your system needs.

Worker Resource Profiles and Sizing

The default concurrency: 1 is safe but wasteful. Crank it to 50 and your worker might OOM. The right number depends on three resources: CPU, memory, and event loop lag.

Resource Anatomy of a Running Worker

Every concurrent job slot in a worker consumes:

  • One async task slot in the Node.js microtask queue
  • Heap memory for the job payload, processor closure, and any data it allocates
  • RSS for the V8 heap, event loop machinery, and Redis connection buffers
  • Event loop time — the synchronous portion of the processor runs on the main thread

Use this diagnostic helper to profile your worker's resource usage:

import { Worker, Job } from 'bullmq';
import { Redis } from 'ioredis';
import * as v8 from 'v8';

const connection = new Redis({
  host: process.env.REDIS_HOST,
  port: Number(process.env.REDIS_PORT),
  maxRetriesPerRequest: null,
});

function logWorkerResources(label: string): void {
  const heap = v8.getHeapStatistics();
  const usage = process.memoryUsage();
  console.log(JSON.stringify({
    label,
    timestamp: new Date().toISOString(),
    heapUsedMb: Math.round(heap.used_heap_size / 1024 / 1024),
    heapTotalMb: Math.round(heap.total_heap_size / 1024 / 1024),
    rssMb: Math.round(usage.rss / 1024 / 1024),
    externalMb: Math.round(usage.external / 1024 / 1024),
  }));
}

Sizing for I/O-Bound Processors

I/O-bound workers (HTTP calls, database queries, file reads) spend most of their time waiting. High concurrency is safe because the event loop stays responsive.

Rule of thumb: Start at concurrency = 25 per CPU core, then increase until event loop lag exceeds 50ms.

// I/O-bound worker — high concurrency is safe
const ioBoundWorker = new Worker(
  'email-notifications',
  async (job: Job) => {
    await fetch('https://api.sendgrid.com/v3/mail/send', {
      method: 'POST',
      headers: { Authorization: `Bearer ${process.env.SENDGRID_KEY}` },
      body: JSON.stringify(job.data),
    });
  },
  {
    connection,
    concurrency: Number(process.env.WORKER_CONCURRENCY ?? 50),
  },
);

Sizing for CPU-Bound Processors

CPU-bound workers (image processing, PDF generation, JSON serialization) block the event loop. High concurrency just makes them compete for the same core.

Rule of thumb: concurrency = os.cpus().length maximum. Use sandboxed workers with worker threads to offload processing.

import { Worker } from 'bullmq';
import { cpus } from 'os';

const cpuBoundWorker = new Worker(
  'image-processing',
  `${__dirname}/processors/image-processor.js`,
  {
    connection,
    concurrency: cpus().length - 1, // leave one core for the event loop
    useWorkerThreads: true,          // sandboxed via worker_threads
  },
);

Worker Pool Sizing Formula

Given target metrics, compute the required worker pool with this formula:

Throughput target:      T jobs/sec
Average job duration:   D seconds (p50)
Concurrency per worker: C
Jobs per worker/sec:    C / D
Workers needed:         ceil(T / (C / D))

Example: Target 200 jobs/sec, average duration 0.5s, concurrency 20:

  • Jobs per worker = 20 / 0.5 = 40 jobs/sec
  • Workers needed = ceil(200 / 40) = 5 workers

Add 20% headroom for bursts and 1 extra worker per deployment zone for redundancy:

export function computeWorkerCount(config: {
  throughputTarget: number;
  averageJobDurationMs: number;
  concurrencyPerWorker: number;
  headroomPercent?: number;
  redundancyPerZone?: number;
}): number {
  const jobsPerWorkerPerSec =
    config.concurrencyPerWorker / (config.averageJobDurationMs / 1000);

  let workers = Math.ceil(config.throughputTarget / jobsPerWorkerPerSec);

  const headroom = config.headroomPercent ?? 20;
  workers = Math.ceil(workers * (1 + headroom / 100));

  const redundancy = config.redundancyPerZone ?? 1;
  workers += redundancy;

  return workers;
}

Memory Budget Per Worker

Safe limit = (container memory limit x 0.8) - Redis connection overhead
Per-job budget = safe limit / concurrency

Validate at startup with a budget checker:

export function validateMemoryBudget(
  memoryLimitMb: number,
  concurrency: number,
  averagePayloadKb: number,
): { safe: boolean; perJobBudgetMb: number; recommendation?: string } {
  const overheadMb = 50; // Redis connection, V8 baseline, event loop
  const safeLimitMb = memoryLimitMb * 0.8;
  const availableMb = safeLimitMb - overheadMb;
  const perJobBudgetMb = availableMb / concurrency;

  if (perJobBudgetMb < averagePayloadKb / 1024) {
    return {
      safe: false,
      perJobBudgetMb,
      recommendation: `Reduce concurrency to ${Math.floor(availableMb / (averagePayloadKb / 1024))} or increase memory limit`,
    };
  }

  return { safe: true, perJobBudgetMb };
}

Worker Execution Guarantees and Idempotency Patterns

BullMQ guarantees at-least-once delivery. A job may be processed twice if the worker crashes after completing the work but before BullMQ acknowledges it. To achieve exactly-once semantics, the worker must enforce idempotency.

Worker-Level Idempotency with an Idempotency Key

The most reliable pattern: the producer generates an idempotency key, and the worker checks if it has already been processed.

import { Job } from 'bullmq';
import { Redis } from 'ioredis';

export interface IdempotencyOptions {
  redis: Redis;
  namespace?: string;
  ttlSeconds?: number;
}

export class WorkerIdempotencyGuard {
  private redis: Redis;
  private namespace: string;
  private ttlSeconds: number;

  constructor(opts: IdempotencyOptions) {
    this.redis = opts.redis;
    this.namespace = opts.namespace ?? 'idempotency';
    this.ttlSeconds = opts.ttlSeconds ?? 86_400; // 24 hours default
  }

  async wasProcessed(idempotencyKey: string): Promise<boolean> {
    const key = `${this.namespace}:${idempotencyKey}`;
    const result = await this.redis.get(key);
    return result !== null;
  }

  async markProcessed(idempotencyKey: string): Promise<void> {
    const key = `${this.namespace}:${idempotencyKey}`;
    await this.redis.set(key, '1', 'EX', this.ttlSeconds);
  }

  async processIfNotDone(
    job: Job,
    processor: (job: Job) => Promise<void>,
  ): Promise<{ processed: boolean; skipped: boolean }> {
    const idempotencyKey = job.data.idempotencyKey ?? job.id;
    if (await this.wasProcessed(idempotencyKey)) {
      return { processed: false, skipped: true };
    }
    await processor(job);
    await this.markProcessed(idempotencyKey);
    return { processed: true, skipped: false };
  }
}

Using it in a worker:

import { Worker } from 'bullmq';
import { Redis } from 'ioredis';

const idempotencyRedis = new Redis(process.env.REDIS_URL);
const guard = new WorkerIdempotencyGuard({ redis: idempotencyRedis });

const worker = new Worker(
  'payment-webhooks',
  async (job: Job) => {
    const result = await guard.processIfNotDone(job, async (j) => {
      await fetch(job.data.callbackUrl, {
        method: 'POST',
        body: JSON.stringify(job.data.payload),
        headers: { 'Idempotency-Key': job.data.idempotencyKey },
      });
    });

    if (result.skipped) {
      console.log(`Skipped duplicate job ${job.id}`);
    }
  },
  { connection, concurrency: 10 },
);

Scalable Deduplication with a Bloom Filter

At extreme scale (millions of idempotency keys), Redis SETs consume too much memory. A Bloom filter offers a space-efficient probabilistic alternative with a configurable false-positive rate:

import { Job } from 'bullmq';
import { Redis } from 'ioredis';

interface BloomOptions {
  redis: Redis;
  key: string;
  expectedInsertions: number;
  falsePositiveRate?: number;
}

export class BloomDeduplicationFilter {
  private redis: Redis;
  private key: string;
  private bitCount: number;
  private hashCount: number;

  constructor(opts: BloomOptions) {
    this.redis = opts.redis;
    this.key = opts.key;

    const fpr = opts.falsePositiveRate ?? 0.01;
    const insertions = opts.expectedInsertions;
    const bits = Math.ceil(-(insertions * Math.log(fpr)) / (Math.log(2) ** 2));
    const hashes = Math.ceil((bits / insertions) * Math.log(2));

    this.bitCount = bits;
    this.hashCount = hashes;
  }

  private hash(value: string, seed: number): number {
    let h = 0;
    for (let i = 0; i < value.length; i++) {
      h = (h * seed + value.charCodeAt(i)) & 0x7fffffff;
    }
    return h % this.bitCount;
  }

  async mightContain(idempotencyKey: string): Promise<boolean> {
    for (let i = 0; i < this.hashCount; i++) {
      const offset = this.hash(idempotencyKey, i + 1);
      const bit = await this.redis.getbit(this.key, offset);
      if (bit === 0) return false;
    }
    return true; // false positive possible
  }

  async add(idempotencyKey: string): Promise<void> {
    const pipeline = this.redis.pipeline();
    for (let i = 0; i < this.hashCount; i++) {
      const offset = this.hash(idempotencyKey, i + 1);
      pipeline.setbit(this.key, offset, 1);
    }
    await pipeline.exec();
  }

  async checkAndAdd(idempotencyKey: string): Promise<boolean> {
    const guardKey = `${this.key}:seen:${idempotencyKey}`;
    const acquired = await this.redis.setnx(guardKey, '1');
    if (acquired === 0) return false;

    await this.add(idempotencyKey);
    await this.redis.expire(guardKey, 86_400);
    return true;
  }
}

When to choose which approach:

  • Redis SET (exact): Up to ~100K unique keys per day, need zero false positives
  • Bloom filter (probabilistic): Millions of keys, can tolerate 0.1-1% false positives

At-Least-Once vs. Exactly-Once: Where Duplicates Come From

Scenario Duplicate? Mitigation
Worker crashes after processing but before ack One duplicate Idempotency key
Stalled job recovery (worker heartbeat expires) Possible duplicate Idempotency key + short lockDuration
Redis failover with unacked commands Possible duplicate Idempotency key
Network partition during completion Possible duplicate Idempotency key
UnrecoverableError thrown intentionally No duplicate Deterministic failure

Achieving exactly-once requires three layers:

  1. Producer side: Generate deterministic jobId or include a deduplication.id field
  2. Worker side: Idempotency guard or Bloom filter (as shown above)
  3. Downstream side: Idempotency keys on external API calls so POST retries are safe

Deduplication Window Management

How long should an idempotency key live? Match your maximum job processing window:

export function computeDeduplicationTTL(config: {
  maxJobDurationMinutes: number;
  maxRetryAttempts: number;
  maxBackoffMinutes: number;
  safetyMarginMinutes: number;
}): number {
  const worstCaseMinutes =
    config.maxJobDurationMinutes
    + (config.maxRetryAttempts * config.maxBackoffMinutes)
    + config.safetyMarginMinutes;

  const ttlSeconds = Math.ceil(worstCaseMinutes / 60) * 3600;
  return Math.max(ttlSeconds, 3600);
}

// Example: job takes 5 min, retries 5 times with 10 min backoff, 30 min safety margin
const ttl = computeDeduplicationTTL({
  maxJobDurationMinutes: 5,
  maxRetryAttempts: 5,
  maxBackoffMinutes: 10,
  safetyMarginMinutes: 30,
});
console.log(ttl); // 14400 seconds = 4 hours

Poison Pill Management and Quarantine Patterns

A poison pill is a job that will never succeed — corrupt data, a processor bug, or an upstream dependency that permanently rejects it. Without automatic detection, poison pills waste retries, degrade throughput, and pollute your failed-job set.

Automatic Poison Pill Detection

Track each job's failure signature and detect patterns:

import { Job } from 'bullmq';
import { Redis } from 'ioredis';
import crypto from 'crypto';

interface PoisonDetectionConfig {
  redis: Redis;
  maxFailuresPerSignature?: number;
  windowSeconds?: number;
  namespace?: string;
}

export class PoisonPillDetector {
  private redis: Redis;
  private maxFailures: number;
  private windowSeconds: number;
  private namespace: string;

  constructor(config: PoisonDetectionConfig) {
    this.redis = config.redis;
    this.maxFailures = config.maxFailuresPerSignature ?? 3;
    this.windowSeconds = config.windowSeconds ?? 3600;
    this.namespace = config.namespace ?? 'poison';
  }

  computeSignature(job: Job, error: Error): string {
    const hash = crypto.createHash('sha256');
    hash.update(JSON.stringify(job.data));
    hash.update(error.message);
    hash.update(error.name);
    return hash.digest('hex').slice(0, 16);
  }

  async recordAndCheck(job: Job, error: Error): Promise<{
    isPoison: boolean;
    signature: string;
    failureCount: number;
  }> {
    const signature = this.computeSignature(job, error);
    const key = `${this.namespace}:sig:${signature}`;

    const count = await this.redis.incr(key);
    if (count === 1) {
      await this.redis.expire(key, this.windowSeconds);
    }

    return {
      isPoison: count >= this.maxFailures,
      signature,
      failureCount: count,
    };
  }

  async clearSignature(signature: string): Promise<void> {
    await this.redis.del(`${this.namespace}:sig:${signature}`);
  }
}

The Quarantine Queue Worker Pattern

When a poison pill is detected, move it to a quarantine (dead letter) queue instead of exhausting retries:

import { Worker, Queue, Job, UnrecoverableError } from 'bullmq';

const quarantineQueue = new Queue('quarantine', { connection });
const poisonDetector = new PoisonPillDetector({ redis: connection });

const mainWorker = new Worker(
  'orders',
  async (job: Job): Promise<void> => {
    try {
      await processOrder(job.data);
    } catch (error) {
      const err = error instanceof Error ? error : new Error(String(error));
      const result = await poisonDetector.recordAndCheck(job, err);

      if (result.isPoison) {
        await quarantineQueue.add(
          job.name,
          {
            originalJobId: job.id,
            originalQueue: 'orders',
            data: job.data,
            errorMessage: err.message,
            errorStack: err.stack,
            failureSignature: result.signature,
            failureCount: result.failureCount,
          },
          { priority: 1 },
        );

        throw new UnrecoverableError(
          `Poison pill detected: ${err.message} (signature: ${result.signature})`,
        );
      }

      throw err;
    }
  },
  {
    connection,
    concurrency: 10,
    attempts: 3, // fewer attempts since poison detection catches early
    backoff: { type: 'exponential', delay: 2000 },
  },
);

Configurable Failure Thresholds Per Job Type

Not all job types are equally resilient. Allow per-type configuration:

import crypto from 'crypto';

interface JobTypeThresholds {
  [jobName: string]: {
    maxFailuresBeforeQuarantine: number;
    windowSeconds: number;
  };
}

const defaultThresholds: JobTypeThresholds = {
  'send-email': { maxFailuresBeforeQuarantine: 5, windowSeconds: 3600 },
  'process-payment': { maxFailuresBeforeQuarantine: 2, windowSeconds: 600 },
  'generate-report': { maxFailuresBeforeQuarantine: 10, windowSeconds: 7200 },
};

class PerTypePoisonDetector {
  private redis: Redis;
  private thresholds: JobTypeThresholds;

  constructor(redis: Redis, thresholds: JobTypeThresholds) {
    this.redis = redis;
    this.thresholds = thresholds;
  }

  async check(job: Job, error: Error): Promise<{
    isPoison: boolean;
    threshold: number;
  }> {
    const config = this.thresholds[job.name] ?? {
      maxFailuresBeforeQuarantine: 3,
      windowSeconds: 3600,
    };

    const hash = crypto.createHash('sha256');
    hash.update(JSON.stringify(job.data));
    hash.update(error.message);
    const signature = hash.digest('hex').slice(0, 16);

    const key = `poison:${job.name}:${signature}`;
    const count = await this.redis.incr(key);
    if (count === 1) {
      await this.redis.expire(key, config.windowSeconds);
    }

    return {
      isPoison: count >= config.maxFailuresBeforeQuarantine,
      threshold: config.maxFailuresBeforeQuarantine,
    };
  }
}

Quarantine Worker — Review, Retry, or Discard

A dedicated quarantine worker processes quarantined jobs for manual or automated review:

const quarantineWorker = new Worker(
  'quarantine',
  async (job: Job) => {
    const payload = job.data as {
      originalJobId: string;
      originalQueue: string;
      data: unknown;
      errorMessage: string;
      failureSignature: string;
      failureCount: number;
    };

    console.log(JSON.stringify({
      level: 'WARN',
      message: 'Job quarantined',
      originalQueue: payload.originalQueue,
      originalJobId: payload.originalJobId,
      error: payload.errorMessage,
      failureCount: payload.failureCount,
      signature: payload.failureSignature,
    }));

    // In production, this would:
    // 1. Send a Slack / PagerDuty alert
    // 2. Store in a database for a review dashboard
    // 3. Optionally re-enqueue after a manual data fix
    // 4. Discard if deterministically unrecoverable

    return { quarantined: true };
  },
  {
    connection,
    concurrency: 1,
    attempts: 1, // never retry quarantine processing
  },
);

Alerting on Quarantine Events

Use QueueEvents to monitor quarantine queue activity:

import { QueueEvents } from 'bullmq';

const quarantineEvents = new QueueEvents('quarantine', { connection });

quarantineEvents.on('completed', ({ jobId, returnvalue }) => {
  console.log(`Job ${jobId} entered quarantine`);
});

// Track quarantine rate for a dashboard
let quarantineRate = 0;
quarantineEvents.on('completed', () => {
  quarantineRate++;
});

setInterval(() => {
  console.log(`Quarantine rate: ${quarantineRate} jobs/min`);
  quarantineRate = 0;
}, 60_000);

Worker Observability

Aggregate queue dashboards (throughput, queue depth, error rate) are essential but insufficient. You also need per-worker observability to detect single-worker degradation before it becomes a fleet-wide problem.

Per-Worker Metrics Collector

import { Worker, Job } from 'bullmq';
import { performance } from 'perf_hooks';

interface WorkerMetrics {
  workerId: string;
  queueName: string;
  jobsProcessed: number;
  jobsFailed: number;
  totalProcessingTimeMs: number;
  maxProcessingTimeMs: number;
  minProcessingTimeMs: number;
  histogram: Map<string, number>;
  saturationRatio: number;
  lastErrorByType: Map<string, number>;
}

export class WorkerMetricsCollector {
  private metrics: WorkerMetrics;
  private startTime: number;
  private busyMs = 0;
  private lastJobStart = 0;

  constructor(workerId: string, queueName: string) {
    this.metrics = {
      workerId,
      queueName,
      jobsProcessed: 0,
      jobsFailed: 0,
      totalProcessingTimeMs: 0,
      maxProcessingTimeMs: 0,
      minProcessingTimeMs: Infinity,
      histogram: new Map(),
      saturationRatio: 0,
      lastErrorByType: new Map(),
    };
    this.startTime = Date.now();
  }

  recordJobStart(): void {
    this.lastJobStart = performance.now();
  }

  recordJobEnd(job: Job, durationMs: number): void {
    this.metrics.jobsProcessed++;
    this.metrics.totalProcessingTimeMs += durationMs;
    this.metrics.maxProcessingTimeMs = Math.max(this.metrics.maxProcessingTimeMs, durationMs);
    this.metrics.minProcessingTimeMs = Math.min(this.metrics.minProcessingTimeMs, durationMs);
    this.busyMs += durationMs;

    const bucket = this.bucketFor(durationMs);
    this.metrics.histogram.set(bucket, (this.metrics.histogram.get(bucket) ?? 0) + 1);
  }

  recordJobFailure(job: Job, error: Error): void {
    this.metrics.jobsFailed++;
    const errorType = error.name;
    this.metrics.lastErrorByType.set(
      errorType,
      (this.metrics.lastErrorByType.get(errorType) ?? 0) + 1,
    );
  }

  snapshot(): WorkerMetrics & { uptimeMs: number; saturationRatio: number } {
    const uptimeMs = Date.now() - this.startTime;
    const saturationRatio = uptimeMs > 0 ? this.busyMs / uptimeMs : 0;
    return {
      ...this.metrics,
      uptimeMs,
      saturationRatio: Math.min(saturationRatio, 1),
    };
  }

  private bucketFor(ms: number): string {
    if (ms < 10) return '0-10ms';
    if (ms < 50) return '10-50ms';
    if (ms < 100) return '50-100ms';
    if (ms < 500) return '100-500ms';
    if (ms < 1000) return '500-1000ms';
    if (ms < 5000) return '1-5s';
    return '>5s';
  }
}

Wiring Metrics into the Worker

function createObservableWorker(
  queueName: string,
  processor: (job: Job) => Promise<void>,
  workerId: string = `worker-${process.pid}`,
): { worker: Worker; collector: WorkerMetricsCollector } {
  const collector = new WorkerMetricsCollector(workerId, queueName);

  const worker = new Worker(
    queueName,
    async (job: Job) => {
      collector.recordJobStart();
      const start = performance.now();
      try {
        await processor(job);
        collector.recordJobEnd(job, performance.now() - start);
      } catch (error) {
        collector.recordJobFailure(
          job,
          error instanceof Error ? error : new Error(String(error)),
        );
        throw error;
      }
    },
    { connection, concurrency: 10 },
  );

  // Export metrics as structured JSON every 15 seconds (for log aggregation)
  setInterval(() => {
    console.log(JSON.stringify({
      type: 'worker_metrics',
      ...collector.snapshot(),
    }));
  }, 15_000);

  return { worker, collector };
}

Feeding Metrics to Prometheus (Worker-Local)

This supplements queue-level Prometheus instrumentation with worker-local gauges:

import promClient from 'prom-client';

const workerJobsProcessed = new promClient.Counter({
  name: 'bullmq_worker_jobs_processed_total',
  help: 'Total jobs processed by this worker instance',
  labelNames: ['worker_id', 'queue_name', 'job_name'],
});

const workerJobDurationHistogram = new promClient.Histogram({
  name: 'bullmq_worker_job_duration_seconds',
  help: 'Job processing duration in seconds',
  labelNames: ['worker_id', 'queue_name', 'job_name'],
  buckets: [0.01, 0.05, 0.1, 0.5, 1, 5, 10, 30],
});

const workerErrorsByType = new promClient.Counter({
  name: 'bullmq_worker_errors_total',
  help: 'Errors by type in this worker instance',
  labelNames: ['worker_id', 'queue_name', 'error_type'],
});

const workerSaturation = new promClient.Gauge({
  name: 'bullmq_worker_saturation_ratio',
  help: 'Ratio of time this worker spent processing jobs vs idle',
  labelNames: ['worker_id', 'queue_name'],
});

export function recordWorkerMetric(job: Job, durationMs: number): void {
  const labels = {
    worker_id: `worker-${process.pid}`,
    queue_name: job.queueName ?? 'unknown',
    job_name: job.name,
  };
  workerJobsProcessed.inc(labels);
  workerJobDurationHistogram.observe(labels, durationMs / 1000);
}

export function recordWorkerError(job: Job, errorType: string): void {
  workerErrorsByType.inc({
    worker_id: `worker-${process.pid}`,
    queue_name: job.queueName ?? 'unknown',
    error_type: errorType,
  });
}

export function updateSaturation(ratio: number): void {
  workerSaturation.set(
    { worker_id: `worker-${process.pid}`, queue_name: 'orders' },
    ratio,
  );
}

Interpreting Worker Saturation

Worker saturation (busy / (busy + idle)) is a leading indicator:

  • > 80% — Worker is near capacity, queue depth will grow
  • > 95% — Worker is saturated, latency will spike
  • < 20% — Overprovisioned, consider reducing worker count

Saturation is more actionable than queue depth alone because it accounts for variable job duration.

Worker Resource Isolation in Production

Running multiple workers requires decisions about isolation boundaries, resource limits, and failure modes.

Multi-Worker Deployment Patterns

Pattern Isolation Resource Efficiency Complexity Best For
Single process, multiple Worker instances None (shared heap) High Low Low-throughput queues with similar job types
Separate processes (PM2 cluster mode) Process-level (separate RSS) Medium Medium Medium-throughput queues with different resource profiles
Separate containers Container-level (cgroups) Lower Higher High-throughput queues needing strict SLAs
Separate pods (Kubernetes) Pod-level (node isolation) Lowest Highest Multi-tenant systems with independent scaling needs

Single Process / Multiple Workers Pattern

When job types share a resource profile and total throughput is under ~500 jobs/sec:

import { Worker } from 'bullmq';

function createSharedWorker(
  queueName: string,
  processor: (job: any) => Promise<void>,
): Worker {
  return new Worker(queueName, processor, {
    connection, // same Redis connection shared across all workers
    concurrency: Number(process.env.WORKER_CONCURRENCY ?? 10),
  });
}

// All workers share the same Redis connection
const emailWorker = createSharedWorker('emails', emailProcessor);
const reportWorker = createSharedWorker('reports', reportProcessor);
const webhookWorker = createSharedWorker('webhooks', webhookProcessor);

Risk: A CPU-intensive report job can starve email processing. Mitigate by using sandboxed workers (useWorkerThreads: true) for CPU-bound queues.

Separate Containers with CPU/Memory Limits

For strict resource isolation, each worker type gets its own container with explicit cgroup limits:

# docker-compose.yml — resource isolation per worker type
services:
  worker-emails:
    build:
      context: .
      dockerfile: Dockerfile.worker
    command: ["node", "workers/email-worker.js"]
    deploy:
      resources:
        limits:
          cpus: '0.5'
          memory: 256M
        reservations:
          cpus: '0.25'
          memory: 128M
    environment:
      - WORKER_CONCURRENCY=20
      - REDIS_HOST=redis

  worker-reports:
    build:
      context: .
      dockerfile: Dockerfile.worker
    command: ["node", "workers/report-worker.js"]
    deploy:
      resources:
        limits:
          cpus: '2.0'
          memory: 1G
        reservations:
          cpus: '1.0'
          memory: 512M
    environment:
      - WORKER_CONCURRENCY=4
      - REDIS_HOST=redis

Graceful OOM Handling

When a worker hits a memory limit, the kernel sends SIGKILL immediately — no graceful shutdown. Mitigate with proactive memory checks:

import { Worker } from 'bullmq';

const MEMORY_WARNING_PERCENT = 0.85;
const MEMORY_CRITICAL_PERCENT = 0.95;

async function startMemoryAwareWorker(
  queueName: string,
  processor: (job: any) => Promise<void>,
  memoryLimitBytes: number,
): Promise<Worker> {
  const worker = new Worker(queueName, processor, {
    connection,
    concurrency: Number(process.env.WORKER_CONCURRENCY ?? 10),
  });

  const monitor = setInterval(async () => {
    const usage = process.memoryUsage();
    const rssRatio = usage.rss / memoryLimitBytes;

    if (rssRatio >= MEMORY_CRITICAL_PERCENT) {
      console.error(
        `CRITICAL: RSS at ${Math.round(rssRatio * 100)}% — draining and shutting down`,
      );
      clearInterval(monitor);
      await worker.pause();       // stop accepting new jobs
      await worker.drain();       // wait for active jobs to finish
      await worker.close();
      process.exit(1);            // orchestrator restarts
    } else if (rssRatio >= MEMORY_WARNING_PERCENT) {
      console.warn(
        `WARNING: RSS at ${Math.round(rssRatio * 100)}% — reducing concurrency`,
      );
      worker.concurrency = Math.max(1, Math.floor(worker.concurrency / 2));
    }
  }, 5000);

  return worker;
}

Worker-Level Retry Budget and Slow-Consumer Backpressure

When a worker processes jobs slower than the queue can deliver them, you need backpressure.

Strategy A: Retry budget per worker instance

class WorkerRetryBudget {
  private maxRetriesPerWindow: number;
  private windowMs: number;
  private attempts: number[] = [];

  constructor(maxRetriesPerWindow: number, windowMs: number = 60_000) {
    this.maxRetriesPerWindow = maxRetriesPerWindow;
    this.windowMs = windowMs;
  }

  allowRetry(): boolean {
    const now = Date.now();
    this.attempts = this.attempts.filter(t => now - t < this.windowMs);
    if (this.attempts.length >= this.maxRetriesPerWindow) {
      return false; // budget exhausted — fail fast
    }
    this.attempts.push(now);
    return true;
  }
}

Strategy B: Dynamic backpressure via queue depth

import { Queue, Worker } from 'bullmq';

async function createBackpressureAwareWorker(
  queue: Queue,
  processor: (job: any) => Promise<void>,
  baseConcurrency: number,
): Promise<Worker> {
  const worker = new Worker(queue.name, processor, {
    connection,
    concurrency: baseConcurrency,
  });

  setInterval(async () => {
    const counts = await queue.getJobCounts();
    const waiting = counts.wait + counts.delayed;

    if (waiting > 10_000 && worker.concurrency < baseConcurrency * 2) {
      worker.concurrency = Math.min(worker.concurrency + 5, baseConcurrency * 2);
      console.log(`Backpressure: concurrency increased to ${worker.concurrency}`);
    } else if (waiting < 100 && worker.concurrency > baseConcurrency) {
      worker.concurrency = Math.max(baseConcurrency, worker.concurrency - 5);
      console.log(`Backpressure: concurrency reduced to ${worker.concurrency}`);
    }
  }, 10_000);

  return worker;
}

Deployment Checklist

Before deploying worker isolation patterns in production:

  • Each worker class has an explicit memoryLimitBytes set via environment variable
  • --max-old-space-size is configured in the Node.js runtime args for each container
  • OOM monitoring is wired to the orchestrator's restart policy
  • Each worker type has its own Prometheus metrics (separate labels)
  • Health probes include worker-specific checks (saturation, memory ratio, connection status)
  • Log aggregation tags each line with worker_type and worker_id
  • Retry budget is configured per worker process, not globally

QueueHub Integration Points

Every pattern in this post can be monitored through QueueHub:

  1. Resource profiles — QueueHub shows per-worker saturation and active job counts, helping you validate sizing formulas in real-time
  2. Idempotency — QueueHub's job detail view displays job IDs and metadata, making deduplication debugging straightforward
  3. Quarantine — View quarantined jobs in QueueHub's failed job view; re-enqueue them after fixing upstream data
  4. Observability — QueueHub's live worker view shows per-worker completed, failed, and active counts alongside your custom metrics
  5. Resource isolation — QueueHub's multi-backend support lets you monitor workers across multiple Redis instances, each with their own resource pool

Summary

Pattern Key Takeaway
Resource Sizing Start with the formula: workers = ceil(T / (C / D)). Validate memory budget per worker before deploying.
Idempotency BullMQ gives at-least-once. Guard with idempotency keys (Redis SET for precision, Bloom filter for scale).
Poison Pill Quarantine Detect repeatedly failing jobs by error signature. Move to a quarantine queue. Alert on quarantine events.
Worker Observability Track per-worker: jobs processed, duration histogram, error rate by type, saturation ratio.
Resource Isolation Match isolation level to throughput needs. Monitor memory proactively to avoid silent OOM kills.

QueueHub — The universal dashboard for BullMQ and BeeQueue. Monitor your workers, queues, and Redis health in one place. Get started free.

← All Articles
Share on XShare on LinkedIn

Related Articles

2026-07-12

Best BullMQ Dashboard Alternatives in 2026: A Comprehensive Comparison

Comparing every BullMQ UI option side by side: Bull Board, Arena, Taskforce, QueueHub, and raw redis-cli. Feature matrices, pricing, pros and cons, and recommendations for every team size.

2026-07-10

QueueHub vs pg-boss: Redis vs PostgreSQL as a Job Queue Backend

pg-boss uses PostgreSQL as a Node.js job queue, while BullMQ (monitored by QueueHub) uses Redis. We compare the two approaches across throughput, persistence, transactional queues, deployment, and when transactional enqueueing matters.

2026-07-09

QueueHub vs Temporal: Job Queues vs Workflow Orchestration

Temporal is a workflow orchestration platform — a different category from Redis-backed job queues. We compare Temporal.io against BullMQ + QueueHub, covering complexity, use cases, durability, observability, and when to choose each.