Testing BullMQ Workers: From Unit Mocks to Production CI

Published: July 1, 2026 · Reading time: ~14 minutes
Tags: BullMQ, Testing, Node.js, TypeScript, CI/CD, Testcontainers
Author: QueueHub Team

---

Introduction — The Testing Gap in Queue-Based Systems

Queues are the backbone of asynchronous processing in modern Node.js applications. They handle everything — email delivery, report generation, image processing, order fulfillment — and yet they're often the least-tested part of the system.

We've all heard (or said) it: "It works in development, queues just process jobs." But queue failures are uniquely dangerous because they're silent. An HTTP endpoint returns a 500, and you know immediately. A queue worker that drops jobs? You might not notice until a customer complains three days later that their order never shipped.

Why are BullMQ workers harder to test than, say, HTTP handlers?

• Asynchronous execution — there's no request-response cycle. A job is added, the worker picks it up sometime later, and the result arrives via events or polling.
• Redis dependency — workers need a live Redis instance for internal operations: blocking list pops, streams, distributed locks, Lua scripts.
• Timing challenges — delays, retry backoffs, and stalled-job detection all involve real wall-clock time.
• Event-driven outcomes — completion and failure are observed through QueueEvents, not function return values.

The cost of untested queue logic: data loss, duplicate processing, silent production failures that erode customer trust. In this post, we'll build a layered testing strategy that covers unit tests (fast, no Redis), integration tests (real Redis via Testcontainers), and CI/CD pipelines that keep everything green.

---

Section 1: The Testing Pyramid for Queue Workers

A well-balanced test suite for BullMQ workers follows the classic testing pyramid — adapted for queue-specific concerns:

         ╱  E2E  ╲          Few — full pipeline, slow, high confidence
        ╱─────────╲
       ╱ Integration ╲      Medium — real Redis, retry/shutdown scenarios
      ╱───────────────╲
     ╱   Unit Tests     ╲   Many — pure processor logic, fast, no Redis
    ╱─────────────────────╲

• Unit tests test your processor functions in isolation — pure async functions with mocked dependencies. No Redis, no BullMQ Worker instantiation. These run in milliseconds.
• Integration tests spin up a real Redis instance (disposable Docker containers via testcontainers) and exercise the full Queue → Worker → QueueEvents lifecycle. They verify retries, stalled jobs, and graceful shutdown.
• E2E tests exercise the complete pipeline: an API endpoint adds a job, a worker processes it, and you verify side effects (database writes, file system artifacts, external API calls).

The directory structure mirrors this separation:

tests/
├── unit/
│   └── processors/
│       ├── orderProcessor.spec.ts
│       └── reportProcessor.spec.ts
├── integration/
│   ├── queues/
│   │   ├── orderWorker.spec.ts
│   │   └── retryScenarios.spec.ts
│   └── helpers/
│       └── createTestRedis.ts
└── e2e/
    └── orderPipeline.spec.ts

Pro tip: Don't skip integration tests just because your unit tests pass. Unit tests verify your logic; integration tests prove your workers actually work with a real Redis instance.

---

Section 2: Unit Testing Processor Functions in Isolation

The cardinal rule of testable BullMQ code: separate business logic from queue infrastructure.

Your processor function should be a pure async function that accepts typed data and injected dependencies, then returns a result. The Worker becomes a thin wrapper — and the wrapper itself is tested in integration tests.

The Dependency Injection Pattern

// processors/emailProcessor.ts — TESTABLE
export interface EmailDeps {
  sendMail: (to: string, subject: string, body: string) => Promise<void>;
  templateRenderer: (template: string, vars: Record<string, unknown>) => string;
}

export async function processEmailJob(
  data: { to: string; template: string; variables: Record<string, unknown> },
  deps: EmailDeps,
): Promise<{ sent: boolean; recipient: string }> {
  const body = deps.templateRenderer(data.template, data.variables);
  await deps.sendMail(data.to, 'Notification', body);
  return { sent: true, recipient: data.to };
}

Compare this with the untestable approach — a function that calls sendMail directly from a global import, mixed inside the Worker callback:

// ❌ NOT TESTABLE: Mixed concerns, implicit dependencies
import { sendMail } from './mailService';

export function createBadWorker(connection: IORedis) {
  return new Worker('email', async (job) => {
    const body = renderTemplate(job.data.template, job.data.variables);
    await sendMail(job.data.to, 'Notification', body);
    return { sent: true };
  }, { connection });
}

Writing Unit Tests (Vitest)

With the processor function separated, testing becomes straightforward:

import { describe, it, expect, vi, beforeEach } from 'vitest';
import { processEmailJob } from './processors/emailProcessor';

describe('processEmailJob', () => {
  const mockDeps = {
    sendMail: vi.fn().mockResolvedValue(undefined),
    templateRenderer: vi.fn().mockReturnValue('<h1>Hello</h1>'),
  };

  beforeEach(() => vi.clearAllMocks());

  it('sends email with rendered template', async () => {
    const result = await processEmailJob(
      { to: 'user@test.com', template: 'welcome', variables: { name: 'Alice' } },
      mockDeps,
    );

    expect(mockDeps.templateRenderer).toHaveBeenCalledWith('welcome', { name: 'Alice' });
    expect(mockDeps.sendMail).toHaveBeenCalledWith('user@test.com', 'Notification', '<h1>Hello</h1>');
    expect(result).toEqual({ sent: true, recipient: 'user@test.com' });
  });

  it('rejects when sendMail fails and propagates the error', async () => {
    mockDeps.sendMail.mockRejectedValueOnce(new Error('SMTP timeout'));

    await expect(
      processEmailJob({ to: 'user@test.com', template: 'welcome', variables: {} }, mockDeps),
    ).rejects.toThrow('SMTP timeout');
  });
});

When Processors Need the Job Object

Some processors call methods on the full Job object — updateProgress(), log(), extendLock(). When that's the case, create a mock-job helper factory:

// tests/helpers/mockJob.ts
import { vi } from 'vitest';
import type { Job } from 'bullmq';

export function createMockJob<T>(overrides: Partial<Job<T>> = {}): Job<T> {
  return {
    id: `mock-${Date.now()}`,
    name: 'default',
    data: {} as T,
    opts: { attempts: 1 },
    attemptsMade: 0,
    timestamp: Date.now(),
    queueName: 'test-queue',
    updateProgress: vi.fn().mockResolvedValue(undefined),
    log: vi.fn().mockResolvedValue(undefined),
    extendLock: vi.fn().mockResolvedValue(undefined),
    ...overrides,
  } as unknown as Job<T>;
}

This gives you full control over what each Job method returns without spinning up Redis.

---

Section 3: Simulating Redis for Structural Worker Tests

For pure unit tests of processor functions, you don't need Redis at all — Section 2 covers that completely. But what about testing the Worker instantiation and event wiring?

You have three options:

1. Mock IORedis directly — a hand-rolled mock or ioredis-mock for testing worker creation and construction options.
2. Lightweight embedded Redis — use the redis-server npm package to spawn a temporary Redis process.
3. Testcontainers — Docker-based disposable Redis, covered in depth in Section 4.

The Case for Mocking Redis

Mocking Redis is appropriate when you're testing structural concerns: does the worker construct with the right options? Do custom event handlers fire correctly? Does the worker start polling?

// A minimal IORedis mock for worker construction tests
import { Worker, Queue } from 'bullmq';
import type { IORedis } from 'ioredis';

function createMockConnection(): IORedis {
  return {
    on: vi.fn().mockReturnThis(),
    redisVersion: '7.2.0',
    status: 'connected',
    connect: vi.fn().mockResolvedValue(undefined),
    disconnect: vi.fn().mockResolvedValue(undefined),
    quit: vi.fn().mockResolvedValue(undefined),
    duplicate: vi.fn().mockReturnThis(),
  } as unknown as IORedis;
}

it('constructs a worker with custom options', () => {
  const connection = createMockConnection();
  const worker = new Worker('test', async () => ({ ok: true }), {
    connection,
    concurrency: 5,
    stalledInterval: 5_000,
  });
  expect(worker).toBeInstanceOf(Worker);
  worker.close();
});

When NOT to Mock Redis

⚠️ Critical warning: BullMQ's internal mechanics — lock renewal, stalled-job detection, delayed job scheduling, rate limiting — all rely on specific Redis data structures and Lua scripts. A mock cannot replicate BRPOPLPUSH, stream group operations, or Lua EVALSHA. Use mocking only for structural/construction tests. For behavioral tests, use a real Redis instance (see Section 4).

---

Section 4: Integration Testing with Real Redis (Testcontainers)

The gold standard for BullMQ integration tests: spin up a disposable Redis container per test suite using @testcontainers/redis. This gives you total isolation, zero test pollution, and production-like Redis behavior.

Testcontainers Setup

// tests/helpers/createTestRedis.ts
import { RedisContainer, StartedRedisContainer } from '@testcontainers/redis';
import { Redis } from 'ioredis';
import { Queue, Worker, QueueEvents } from 'bullmq';

export interface TestRedisContext {
  connection: Redis;
  container: StartedRedisContainer;
  createQueue: (name: string) => Queue;
  createWorker: (
    name: string,
    processor: (job: any) => Promise<any>,
  ) => Worker;
  createQueueEvents: (name: string) => QueueEvents;
  cleanup: () => Promise<void>;
}

export async function createTestRedis(): Promise<TestRedisContext> {
  const container = await new RedisContainer('redis:7-alpine')
    .withExposedPorts(6379)
    .start();

  const connection = new Redis({
    host: container.getHost(),
    port: container.getMappedPort(6379),
    maxRetriesPerRequest: null,
  });

  const queues: Queue[] = [];
  const workers: Worker[] = [];
  const eventsBus: QueueEvents[] = [];

  return {
    connection,
    container,
    createQueue: (name: string) => {
      const q = new Queue(name, { connection });
      queues.push(q);
      return q;
    },
    createWorker: (name, processor) => {
      const w = new Worker(name, processor, { connection });
      workers.push(w);
      return w;
    },
    createQueueEvents: (name) => {
      const e = new QueueEvents(name, { connection });
      eventsBus.push(e);
      return e;
    },
    cleanup: async () => {
      await Promise.all(workers.map((w) => w.close()));
      await Promise.all(queues.map((q) => q.close()));
      await Promise.all(eventsBus.map((e) => e.close()));
      await connection.quit();
      await container.stop();
    },
  };
}

Using It in a Test Suite

import { describe, it, beforeAll, afterAll, expect } from 'vitest';

describe('Order Worker Integration', () => {
  let ctx: TestRedisContext;

  beforeAll(async () => {
    ctx = await createTestRedis();
  }, 30_000); // Container startup can take time

  afterAll(async () => {
    await ctx.cleanup();
  });

  it('processes a job and returns the expected result', async () => {
    const queue = ctx.createQueue('orders');
    const worker = ctx.createWorker('orders', async (job) => {
      return { processed: true, orderId: job.data.id };
    });
    const events = ctx.createQueueEvents('orders');

    const job = await queue.add('process-order', { id: 'ORD-001', items: ['A', 'B'] });
    const result = await job.waitUntilFinished(events, 5_000);

    expect(result).toEqual({ processed: true, orderId: 'ORD-001' });
  });

  it('marks a job as failed when the processor throws', async () => {
    const queue = ctx.createQueue('orders');
    const worker = ctx.createWorker('orders', async () => {
      throw new Error('Processing error');
    });
    const events = ctx.createQueueEvents('orders');

    const job = await queue.add('failing-job', { id: 'ORD-002' });

    await expect(job.waitUntilFinished(events, 5_000)).rejects.toThrow();
    const state = await job.getState();
    expect(state).toBe('failed');
  });
});

Key details to get right:

• maxRetriesPerRequest: null — required for BullMQ compatibility with IORedis v5+.
• Container startup timeouts — beforeAll needs a generous timeout (30s) for the first Docker pull.
• Always call cleanup() — unclosed BullMQ connections cause Vitest/Jest to hang with open handles.
• Run integration tests serially — vitest --run or threads: false avoids Redis port contention.

---

Section 5: Testing Retry and Failure Scenarios

BullMQ's retry mechanism is one of its most powerful features — and one of the hardest to get right without tests. You need to verify that retries actually happen, that backoff delays are respected, and that unrecoverable errors skip retries.

Testing Retry Attempts

describe('Job Retry Scenarios', () => {
  let ctx: TestRedisContext;

  it('retries the configured number of times before succeeding', async () => {
    const queue = ctx.createQueue('retry-queue');
    let attempts = 0;

    const worker = ctx.createWorker('retry-queue', async (job) => {
      attempts++;
      if (attempts < 3) throw new Error(`Attempt ${attempts} failed`);
      return { success: true, attempts };
    }, { concurrency: 1 });

    const events = ctx.createQueueEvents('retry-queue');
    const job = await queue.add('retry-job', { x: 1 }, { attempts: 3 });

    const result = await job.waitUntilFinished(events, 10_000);
    expect(result).toEqual({ success: true, attempts: 3 });
  });

  it('moves to failed state when max attempts are exceeded', async () => {
    const queue = ctx.createQueue('retry-queue');
    const worker = ctx.createWorker('retry-queue', async () => {
      throw new Error('Always fails');
    });
    const events = ctx.createQueueEvents('retry-queue');

    const job = await queue.add('fail-job', {}, { attempts: 2 });

    const failedEvent = await new Promise<any>((resolve) => {
      events.on('failed', ({ jobId, failedReason }) => resolve({ jobId, failedReason }));
    });

    expect(failedEvent.jobId).toBe(job.id);
    expect(failedEvent.failedReason).toContain('Always fails');
  });
});

Testing UnrecoverableError

BullMQ's UnrecoverableError tells the queue: "Don't bother retrying — this job will never succeed." Testing this is critical for jobs where retries would be wasteful or harmful:

import { UnrecoverableError } from 'bullmq';

it('skips retries for UnrecoverableError', async () => {
  const queue = ctx.createQueue('retry-queue');
  const worker = ctx.createWorker('retry-queue', async () => {
    throw new UnrecoverableError('Permanent failure — invalid data');
  });
  const events = ctx.createQueueEvents('retry-queue');

  const job = await queue.add('no-retry-job', {}, { attempts: 5 });
  const failedEvent = await new Promise<any>((resolve) => {
    events.on('failed', ({ jobId, failedReason }) => resolve({ jobId, failedReason }));
  });

  expect(failedEvent.jobId).toBe(job.id);

  // Verify it only attempted once despite attempts: 5
  const finalJob = await queue.getJob(job.id!);
  expect(finalJob?.attemptsMade).toBe(1);
});

Testing Backoff Delays

For testing exponential backoff, use timing assertions with generous tolerances:

it('respects exponential backoff delay between retries', async () => {
  const queue = ctx.createQueue('backoff-queue');
  const timestamps: number[] = [];

  const worker = ctx.createWorker('backoff-queue', async () => {
    timestamps.push(Date.now());
    throw new Error('Retry me');
  });

  const events = ctx.createQueueEvents('backoff-queue');

  await queue.add('backoff-job', {}, {
    attempts: 3,
    backoff: { type: 'exponential', delay: 500 },
  });

  await new Promise<void>((resolve) => {
    events.on('failed', () => resolve());
  });

  // Allow some time for retries to execute
  await new Promise((r) => setTimeout(r, 3_000));

  expect(timestamps.length).toBe(3);
  // Delay between attempt 1 and 2 should be ~500ms, between 2 and 3 ~1000ms
  const gaps: number[] = [];
  for (let i = 1; i < timestamps.length; i++) {
    gaps.push(timestamps[i] - timestamps[i - 1]);
  }
  expect(gaps[0]).toBeGreaterThanOrEqual(400);
  expect(gaps[1]).toBeGreaterThanOrEqual(800);
});

---

Section 6: Testing Stalled Job Detection and Recovery

Stalled jobs are one of the trickiest failure modes in BullMQ. A job stalls when the worker processing it stops renewing its lock — usually because the event loop is blocked by CPU-intensive work, or the worker process crashed. BullMQ detects these after stalledInterval (default: 30s) and re-queues them.

Simulating a Stall in Tests

The challenge is creating a stalled job without actually killing a process. You can block the event loop intentionally, combined with a very short stalledInterval:

describe('Stalled Job Detection', () => {
  let ctx: TestRedisContext;

  it('detects stalled jobs and re-queues them with short stalledInterval', async () => {
    const queue = ctx.createQueue('stall-queue');
    const events = ctx.createQueueEvents('stall-queue');

    // Worker that blocks the event loop to simulate a stall
    const worker = ctx.createWorker('stall-queue', async (job) => {
      if (job.data.block) {
        // Block event loop so the worker can't renew its lock
        const start = Date.now();
        while (Date.now() - start < 4_000) {
          // Busy-wait — prevents lock renewal
        }
      }
      return { done: true };
    }, {
      stalledInterval: 2_000, // Short enough for tests
      lockDuration: 2_000,    // Must match stalledInterval closely
    });

    // Add a blocking job, then a quick one
    await queue.add('blocking-job', { block: true });
    const quickJob = await queue.add('quick-job', { block: false });

    // The quick job should complete even though the first stalled
    const result = await quickJob.waitUntilFinished(events, 10_000);
    expect(result).toEqual({ done: true });
  }, 20_000);
});

Testing maxStalledCount

When a job stalls repeatedly and exceeds maxStalledCount, BullMQ moves it to the failed state. Here's how to simulate and verify this:

it('fails a job after exceeding maxStalledCount', async () => {
  const queue = ctx.createQueue('stall-queue');
  const events = ctx.createQueueEvents('stall-queue');

  // Use a custom worker that we can close mid-processing
  let worker = new Worker('stall-queue', async () => {
    // Long processing that lets us close the worker before it finishes
    await new Promise((resolve) => setTimeout(resolve, 10_000));
    return { ok: true };
  }, {
    connection: ctx.connection,
    stalledInterval: 2_000,
    lockDuration: 2_000,
    maxStalledCount: 2,
  });

  const job = await queue.add('stall-me', {});

  // Close the worker while job is being processed — simulates crash
  await new Promise((r) => setTimeout(r, 500)); // Let job go active
  await worker.close(true); // Force close mimics an unclean shutdown

  // Restart worker — it should pick up the stalled job
  worker = new Worker('stall-queue', async () => {
    await new Promise((resolve) => setTimeout(resolve, 10_000));
    return { ok: true };
  }, {
    connection: ctx.connection,
    stalledInterval: 2_000,
    lockDuration: 2_000,
    maxStalledCount: 2,
  });

  // Let it process and stall again
  await new Promise((r) => setTimeout(r, 6_000));
  await worker.close(true);

  // Restart one more time — third attempt should exceed maxStalledCount
  const finalWorker = new Worker('stall-queue', async () => {
    return { ok: true };
  }, {
    connection: ctx.connection,
    stalledInterval: 2_000,
    lockDuration: 2_000,
  });

  const stalledEvent = await new Promise<any>((resolve) => {
    events.on('failed', ({ jobId, failedReason }) => resolve({ jobId, failedReason }));
  });

  expect(stalledEvent.jobId).toBe(job.id);
  expect(stalledEvent.failedReason).toContain('exceeded');

  await finalWorker.close();
}, 30_000);

Pro tip: The 'stalled' event on QueueEvents lets you observe stalled jobs deterministically instead of relying on arbitrary timeouts.

---

Section 7: Testing Graceful Shutdown

A worker that stops processing mid-job can cause duplicate processing or stalled jobs. BullMQ provides worker.close() (graceful — waits for in-flight jobs) and worker.close(true) (force — immediate stop).

Graceful vs. Force Close

describe('Graceful Shutdown', () => {
  let ctx: TestRedisContext;

  it('graceful close waits for in-flight jobs to finish', async () => {
    const queue = ctx.createQueue('shutdown-queue');
    let jobCompleted = false;

    const worker = ctx.createWorker('shutdown-queue', async () => {
      await new Promise((resolve) => setTimeout(resolve, 1_000));
      jobCompleted = true;
      return { ok: true };
    });

    const job = await queue.add('slow-job', {});
    // Give it a moment to start processing
    await new Promise((resolve) => setTimeout(resolve, 100));

    // Graceful close — should wait for job
    await worker.close();
    expect(jobCompleted).toBe(true);
  }, 10_000);

  it('force close does NOT wait for in-flight jobs', async () => {
    const queue = ctx.createQueue('shutdown-queue');
    let jobCompleted = false;

    const worker = ctx.createWorker('shutdown-queue', async () => {
      await new Promise((resolve) => setTimeout(resolve, 3_000));
      jobCompleted = true;
    });

    await queue.add('slow-job', {});
    await new Promise((resolve) => setTimeout(resolve, 100));

    // Force close — should NOT wait
    await worker.close(true);
    expect(jobCompleted).toBe(false);
  }, 10_000);

  it('closed worker does not pick up new jobs', async () => {
    const queue = ctx.createQueue('shutdown-queue');
    const processedJobs: string[] = [];

    const worker = ctx.createWorker('shutdown-queue', async (job) => {
      processedJobs.push(job.id!);
    });

    await worker.close();

    const orphanJob = await queue.add('orphan-job', {});
    await new Promise((resolve) => setTimeout(resolve, 1_000));

    expect(processedJobs).toHaveLength(0);

    const state = await orphanJob.getState();
    expect(state).toBe('waiting');
  });
});

Always verify that worker.isRunning() returns false after close, and run your test suite with --detectOpenHandles in CI to catch leaked connections.

---

Section 8: Deterministic Testing Patterns for Flaky-Proof Suites

Flaky tests are the #1 complaint in queue testing. Here are proven patterns to eliminate them.

Event-Based Waits (Avoid setTimeout)

// ✅ BETTER: Wait for specific event with timeout
async function waitForJobCompletion(
  events: QueueEvents,
  jobId: string,
  timeoutMs: number = 5_000,
): Promise<any> {
  return new Promise((resolve, reject) => {
    const timer = setTimeout(() => {
      reject(new Error(`Timeout waiting for job ${jobId}`));
    }, timeoutMs);

    events.on('completed', ({ jobId: completedId, returnvalue }) => {
      if (completedId === jobId) {
        clearTimeout(timer);
        resolve(returnvalue);
      }
    });

    events.on('failed', ({ jobId: failedId, failedReason }) => {
      if (failedId === jobId) {
        clearTimeout(timer);
        reject(new Error(failedReason));
      }
    });
  });
}

Using removeOnComplete / removeOnFail for Assertions

By default, BullMQ removes completed jobs. For tests that need to inspect job state after processing, disable this:

const job = await queue.add('inspect-me', { x: 1 }, {
  removeOnComplete: false,
  removeOnFail: false,
});

await job.waitUntilFinished(events, 5_000);

const state = await job.getState();
expect(state).toBe('completed');

const completed = await queue.getJobs('completed');
expect(completed).toHaveLength(1);

Snapshot Testing Job State

Capture the full job state for regression detection:

it('produces expected job state after processing', async () => {
  const queue = ctx.createQueue('snapshot-queue');
  const worker = ctx.createWorker('snapshot-queue', async (job) => ({
    result: job.data.value * 2,
  }));
  const events = ctx.createQueueEvents('snapshot-queue');

  const job = await queue.add('snapshot-job', { value: 21 }, {
    removeOnComplete: false,
  });

  await job.waitUntilFinished(events, 5_000);
  const freshJob = await queue.getJob(job.id!);

  // Snapshot the job state for regression detection
  expect(freshJob!.toJSON()).toMatchSnapshot();
});

---

Section 9: CI/CD Pipelines for Queue-Heavy Codebases

Getting queue tests to run reliably in CI requires careful orchestration. Here are three proven approaches.

Option A: GitHub Actions Service Containers

# .github/workflows/test.yml
jobs:
  test:
    runs-on: ubuntu-latest
    services:
      redis:
        image: redis:7-alpine
        ports:
          - 6379:6379
        options: >-
          --health-cmd "redis-cli ping"
          --health-interval 5s
          --health-timeout 5s
          --health-retries 5
    env:
      TEST_REDIS_HOST: localhost
      TEST_REDIS_PORT: 6379
    steps:
      - uses: actions/checkout@v4
      - uses: actions/setup-node@v4
        with:
          node-version: 20
      - run: npm ci
      - run: npm run test:unit
      - run: npm run test:integration

Option B: Testcontainers Inside CI

Testcontainers works inside CI too — but requires Docker-in-Docker support. This gives full isolation at the cost of slightly slower startup.

Option C: Separate Unit and Integration Commands

Always split your test commands so unit tests (parallel, fast) don't block on integration tests (serial, slower):

// vitest.config.ts
import { defineConfig } from 'vitest/config';

export default defineConfig({
  test: {
    include: ['tests/unit/**/*.spec.ts'],
    threads: true, // parallel — fastest possible
  },
});

// vitest.integration.config.ts
export default defineConfig({
  test: {
    include: ['tests/integration/**/*.spec.ts'],
    threads: false, // serial — avoid Redis contention
    testTimeout: 30_000,
    hookTimeout: 30_000,
  },
});

Then in package.json:

{
  "scripts": {
    "test:unit": "vitest run",
    "test:integration": "vitest run --config vitest.integration.config.ts",
    "test": "npm run test:unit && npm run test:integration"
  }
}

---

Section 10: Testing BullMQ Flows (Parent-Child Jobs)

When using FlowProducer to create parent-child job trees, you need to verify that children complete before the parent, and that the parent receives children's results:

it('processes a parent-child flow correctly', async () => {
  const flowProducer = new FlowProducer({ connection: ctx.connection });
  const parentQueue = ctx.createQueue('parent');
  const childQueue = ctx.createQueue('child');

  const childWorker = ctx.createWorker('child', async (job) => ({
    processed: true,
    index: job.data.index,
  }));

  const parentWorker = ctx.createWorker('parent', async (job) => {
    return {
      complete: true,
      childCount: Object.keys(job.data.children?.results || {}).length,
    };
  });

  const parentEvents = ctx.createQueueEvents('parent');

  const flow = await flowProducer.add({
    name: 'parent-job',
    queueName: 'parent',
    data: {},
    children: [
      { name: 'child-1', queueName: 'child', data: { index: 1 } },
      { name: 'child-2', queueName: 'child', data: { index: 2 } },
    ],
  });

  const result = await flow.job.waitUntilFinished(parentEvents, 10_000);

  expect(result).toMatchObject({
    complete: true,
    childCount: 2,
  });

  await flowProducer.close();
});

This ensures your flow orchestration is correct — one of the most common sources of bugs in multi-step queue pipelines.

---

Conclusion: Test Your Workers Like You Test Your HTTP Routes

We've covered the full testing spectrum for BullMQ workers:

• Unit tests (layer 1) — pure processor functions with mocked dependencies, running in milliseconds. The foundation of your test pyramid.
• Integration tests (layer 2) — real Redis via Testcontainers, verifying retries, stalls, shutdown, and event wiring. Where most of your queue-specific confidence comes from.
• E2E tests (layer 3) — full pipeline exercises from API to queue to side effects. Fewer tests, but maximum confidence.

The key takeaways:

1. Separate business logic from queue infrastructure — it's the single highest-leverage change you can make for testability.
2. Use Testcontainers for integration tests — disposable, isolated Redis instances eliminate test pollution and flakiness.
3. Prefer event-based waits over setTimeout — deterministic patterns eliminate your #1 source of flaky tests.
4. Run integration tests serially in CI — parallel Redis access is a recipe for contention failures.
5. Test the edge cases — UnrecoverableError, stalled jobs, graceful shutdown, maxStalledCount. These are the failures that bite you in production.

Your queue workers deserve the same testing rigor as your HTTP handlers. An untested queue is a time bomb waiting for a production incident. Build your testing pyramid, start with unit tests, add integration coverage for behavioral scenarios, and wire it all into CI.

---

Try QueueHub

Monitoring your tested queues in production is just as important. QueueHub gives you real-time visibility into job states, worker health, retry counts, and stalled-job metrics — so when a test fails in CI or a queue backs up in production, you know exactly what's happening.

What testing patterns are you using for your BullMQ workers? Drop into our community and share your war stories — the best queue-testing insights come from production experience.