Resumable Jobs and Circuit Breakers: Advanced BullMQ Retry Patterns

You configure attempts: 5 with exponential backoff. Your job fails on attempt 3 — but it had already processed 80% of a 10,000-row CSV. Now it restarts from row 0. Your downstream API is actually down, not just slow, but each retry attempt burns a connection and a rate-limit slot. When all 5 attempts exhaust, the failed job sits in the completed set with no fallback, no compensation event, and no human notification.

Basic retry configuration — attempts, backoff, stalled job detection — answers when to retry and how fast. But it doesn't address what to retry (the work already done is lost), whether retrying is even safe (the dependency is down), or what happens next (exhaustion is a dead end).

This post covers six advanced patterns that go beyond the basics:

Job checkpointing — save progress mid-execution so retries resume where they left off
Circuit breaker integration — prevent retries when the upstream dependency is known to be down
Graceful degradation — return cached data, trigger compensation events, or route to human approval when retries exhaust
Timeout-aware retry — distinguish timeout failures from logical failures with separate handling
Retry deduplication with idempotency tokens — prevent duplicate side-effects when BullMQ can't confirm completion
Testing retry scenarios — a strategy pattern for simulating every failure mode in your test suite

Prerequisites: You know attempts, backoff { type: 'exponential' }, stalled job detection, UnrecoverableError and the dead letter queue pattern. If any of those are new, start with our BullMQ Job Retry Strategies post.

Job Checkpointing — Resumable Progress for Long-Running Jobs

A job that processes 10,000 rows, resizes 500 images, or paginates through 200 API pages fails on attempt 3 after 80% progress. Without checkpointing, it restarts entirely — wasting the work already done and increasing time-to-completion by up to 5× in a 5-attempt scenario.

The Checkpoint Token Pattern

BullMQ's job.updateProgress() API sets a progress field on the job's Redis hash. This value persists across retries — when the job is re-processed, job.progress returns the last checkpoint. Use this to store a checkpoint token (cursor ID, page number, file offset, or batch index) in the job data on each progress update.

Here's a paginated API processor that saves its cursor after every page:

import { Worker, Job } from 'bullmq';

interface ProcessPagesData {
  apiEndpoint: string;
  cursor?: string | null;
  accumulatedResults: unknown[];
}

async function processPaginatedApi(
  job: Job<ProcessPagesData>,
  apiClient: {
    get: (url: string, cursor?: string) => Promise<{
      data: unknown[];
      nextCursor: string | null;
    }>;
  },
): Promise<{ totalPages: number; totalRecords: number }> {
  const data = job.data;
  let cursor = data.cursor ?? null;
  let pageCount = 0;

  do {
    const response = await apiClient.get(data.apiEndpoint, cursor);

    data.accumulatedResults.push(...response.data);
    pageCount++;

    // Save checkpoint — restart from here if we fail
    await job.updateProgress({
      page: pageCount,
      cursor: response.nextCursor,
      recordsProcessed: data.accumulatedResults.length,
    });

    data.cursor = response.nextCursor;
    cursor = response.nextCursor;
  } while (cursor !== null);

  return { totalPages: pageCount, totalRecords: data.accumulatedResults.length };
}

On retry, the processor reads job.data.cursor — if it's non-null, it resumes from the last saved position instead of starting over.

The JobCheckpoint Helper Class

For reusable checkpointing, encapsulate the logic in a helper class that handles progress updates, state persistence, and lock extension:

import { Job } from 'bullmq';

export interface CheckpointState {
  offset: number;
  processedCount: number;
  cursor: string | null;
  lastUpdated: number;
}

export class JobCheckpoint {
  constructor(
    private readonly job: Job,
    private readonly lockDurationMs: number = 30_000,
  ) {}

  async save(state: Partial<CheckpointState>): Promise<void> {
    const current = this.job.data.checkpoint ?? {};
    const updated = { ...current, ...state, lastUpdated: Date.now() };

    // Merge checkpoint into job data for automatic persistence on retry
    Object.assign(this.job.data, { checkpoint: updated });

    // Update BullMQ's progress field for monitoring visibility
    await this.job.updateProgress(updated);

    // Extend lock if we've been running near the lock window
    if (this.shouldExtendLock()) {
      await this.job.extendLock(this.lockDurationMs);
    }
  }

  getState(): CheckpointState {
    return (this.job.data.checkpoint ?? {}) as CheckpointState;
  }

  hasProgress(): boolean {
    return this.getState().lastUpdated > 0;
  }

  private shouldExtendLock(): boolean {
    const elapsed = Date.now() - this.job.timestamp;
    return elapsed > this.lockDurationMs * 0.7;
  }
}

Usage in a worker:

const worker = new Worker('data-import', async (job) => {
  const checkpoint = new JobCheckpoint(job);
  const state = checkpoint.getState();

  let offset = state.offset ?? 0;
  // Resume processing from the saved offset...
}, { connection });

Checkpointing Best Practices

Checkpoint every 1–5 seconds or every 1–5% of total work. Too frequent writes add Redis overhead. Too infrequent wastes work on retry.
Keep checkpoint data lean. Store a cursor/offset, not accumulated data blobs. Write final results to a database.
Extend the lock when your checkpoint interval approaches the lockDuration. Use the shouldExtendLock() helper above.
Checkpointing naturally deduplicates retries at the batch level. If a job completes a batch, saves the checkpoint, then crashes, the retry picks up exactly where it left off — no duplicates, no gaps.

Circuit Breaker Integration — Fail Fast When Dependencies Are Down

A third-party API is returning 503 errors. Your BullMQ worker retries with exponential backoff: 1s, 2s, 4s, 8s, 16s — all 5 attempts fail. That's 5 wasted rate-limit credits and 31 seconds of blocked worker capacity for a job that was doomed from attempt 1. If 100 jobs hit the same endpoint simultaneously, your backlog grows 100× faster than necessary.

A circuit breaker tracks failure rate for a downstream dependency and fails fast when the dependency is unhealthy. The three states:

Closed: Normal operation — calls pass through
Open: Failures exceed threshold — all calls fail immediately
Half-open: After a cooldown, one probe request tests recovery

Opossum Circuit Breaker with a BullMQ Worker

The opossum library provides a production-grade circuit breaker. Wire it into your worker processor:

import { Worker, Job, UnrecoverableError } from 'bullmq';
import CircuitBreaker from 'opossum';

interface PaymentData {
  userId: string;
  amount: number;
  idempotencyKey: string;
}

async function chargePayment(data: PaymentData): Promise<{ chargeId: string }> {
  const response = await fetch('https://api.payments.com/charges', {
    method: 'POST',
    headers: {
      'Content-Type': 'application/json',
      'Idempotency-Key': data.idempotencyKey,
    },
    body: JSON.stringify({ userId: data.userId, amount: data.amount }),
  });

  if (!response.ok) {
    throw new Error(`Payment API returned ${response.status}`);
  }

  return response.json() as Promise<{ chargeId: string }>;
}

// Open after 5 failures in 30 seconds, stay open for 30 seconds
const paymentBreaker = new CircuitBreaker(chargePayment, {
  timeout: 10_000,
  errorThresholdPercentage: 50,
  resetTimeout: 30_000,
  rollingCountTimeout: 30_000,
  rollingCountBuckets: 3,
});

paymentBreaker.on('open', () => console.warn('[CIRCUIT] Payment API — OPEN'));
paymentBreaker.on('halfOpen', () => console.info('[CIRCUIT] Payment API — HALF-OPEN'));
paymentBreaker.on('close', () => console.info('[CIRCUIT] Payment API — CLOSED'));

const worker = new Worker<PaymentData>('payments', async (job: Job<PaymentData>) => {
  try {
    const result = await paymentBreaker.fire(job.data);
    return result;
  } catch (err) {
    if (paymentBreaker.opened || paymentBreaker.halfOpen) {
      // Circuit is open — don't waste retry attempts
      throw new UnrecoverableError(
        `Payment service is unavailable (circuit breaker is ${paymentBreaker.opened ? 'open' : 'half-open'})`,
      );
    }
    // Re-throw — BullMQ will retry based on job options
    throw err;
  }
}, {
  connection: { host: 'localhost', port: 6379 },
});

Custom Circuit Breaker for Lightweight Cases

When Opossum is too heavy, a simple state-machine breaker works well:

export interface CircuitBreakerOptions {
  failureThreshold: number;
  successThreshold: number;
  timeoutMs: number;
  resetTimeoutMs: number;
  name: string;
}

type CircuitState = 'closed' | 'open' | 'half-open';

export class CircuitBreakerState {
  private state: CircuitState = 'closed';
  private failureCount = 0;
  private successCount = 0;
  private nextAttemptTime = 0;

  constructor(private readonly options: CircuitBreakerOptions) {}

  async call<T>(fn: () => Promise<T>): Promise<T> {
    if (this.state === 'open') {
      if (Date.now() < this.nextAttemptTime) {
        throw new Error(`Circuit breaker "${this.options.name}" is open`);
      }
      this.state = 'half-open';
    }

    try {
      const result = await fn();
      this.onSuccess();
      return result;
    } catch (err) {
      this.onFailure();
      throw err;
    }
  }

  get isOpen(): boolean {
    return this.state === 'open';
  }

  get stateName(): CircuitState {
    return this.state;
  }

  private onSuccess(): void {
    this.failureCount = 0;
    if (this.state === 'half-open') {
      this.successCount++;
      if (this.successCount >= this.options.successThreshold) {
        this.successCount = 0;
        this.state = 'closed';
      }
    }
  }

  private onFailure(): void {
    this.failureCount++;
    this.successCount = 0;

    if (this.state === 'half-open') {
      this.state = 'open';
      this.nextAttemptTime = Date.now() + this.options.resetTimeoutMs;
    } else if (this.state === 'closed' && this.failureCount >= this.options.failureThreshold) {
      this.state = 'open';
      this.nextAttemptTime = Date.now() + this.options.resetTimeoutMs;
    }
  }
}

Circuit Breaker Best Practices

One breaker per dependency. A failing payment API shouldn't block email delivery jobs. Create separate breaker instances for each external service.
Log breaker state in failedReason. Include [CIRCUIT OPEN] or [CIRCUIT HALF-OPEN] in the error message so operations teams see why a job failed immediately versus after N retries.
Backoff interaction: When the circuit is closed, standard exponential backoff works. When the circuit opens, UnrecoverableError shortcuts all remaining attempts — saving those retries for when the API recovers.

Graceful Degradation — What Happens When All Retries Exhaust

All 5 retry attempts exhausted. The job is permanently failed. What now? With basic configuration, the answer is "nothing" — the job sits in the failed set and someone has to manually intervene. In a production system, you need automatic fallback behavior.

The FallbackRegistry Pattern

Register a fallback function for each job type. When retries are exhausted, the worker executes the fallback instead of letting the job die silently:

import { Job } from 'bullmq';

export type FallbackFn<TData, TResult> = (data: TData, failedReason: string) => Promise<TResult> | TResult;

export class FallbackRegistry {
  private readonly fallbacks = new Map<string, FallbackFn<unknown, unknown>>();

  register<TData, TResult>(jobName: string, fn: FallbackFn<TData, TResult>): void {
    this.fallbacks.set(jobName, fn);
  }

  async execute<TData, TResult>(job: Job<TData>): Promise<TResult> {
    const fn = this.fallbacks.get(job.name) as FallbackFn<TData, TResult> | undefined;
    if (!fn) {
      throw new Error(`No fallback registered for job type: ${job.name}`);
    }
    return fn(job.data, job.failedReason ?? 'Unknown error');
  }
}

Three Fallback Strategies

1. Stale/Cached data fallback — Return a previously computed result instead of failing:

const fallbacks = new FallbackRegistry();

fallbacks.register('render-report', async (data: { reportId: string }, reason: string) => {
  console.warn(`[FALLBACK] Report ${data.reportId} failed: ${reason}. Returning cached version.`);

  const cached = await getCachedReport(data.reportId);
  if (cached) {
    return { source: 'cache', reportId: data.reportId, content: cached };
  }

  // Last resort: return a placeholder
  return {
    source: 'placeholder',
    reportId: data.reportId,
    content: { status: 'unavailable', message: 'Report generation temporarily unavailable' },
  };
});

2. Compensation events — Send a message to another queue to handle the failure asynchronously:

import { Queue } from 'bullmq';

const compensationQueue = new Queue('compensation', {
  connection: { host: 'localhost', port: 6379 },
});

fallbacks.register('charge-order', async (data: { orderId: string; userId: string; amount: number }, reason: string) => {
  await compensationQueue.add('payment-failed', {
    orderId: data.orderId,
    userId: data.userId,
    amount: data.amount,
    reason,
    exhaustedAt: new Date().toISOString(),
  });

  return {
    status: 'compensation_enqueued',
    orderId: data.orderId,
    message: 'Payment failed after all retries. Compensation workflow initiated.',
  };
});

3. Human-approval flow — Route to a manual review queue with full context:

const humanReviewQueue = new Queue('manual-review', {
  connection: { host: 'localhost', port: 6379 },
});

fallbacks.register('fraud-review', async (data: { transactionId: string }, reason: string) => {
  await humanReviewQueue.add('review-required', {
    transactionId: data.transactionId,
    reason,
    payload: data,
    urgency: 'high',
    escalatedAt: new Date().toISOString(),
  });

  return {
    status: 'pending_review',
    transactionId: data.transactionId,
    message: 'Flagged for manual review after all retry attempts failed.',
  };
});

Wiring Fallbacks into Your Worker

import { Worker, Job } from 'bullmq';

const fallbacks = new FallbackRegistry();
// ... register fallbacks ...

const worker = new Worker('orders', async (job: Job) => {
  try {
    return await processOrder(job.data);
  } catch (err) {
    // All attempts exhausted — apply fallback
    if (job.attemptsMade >= (job.opts.attempts ?? 1)) {
      return fallbacks.execute(job);
    }
    throw err;
  }
}, {
  connection: { host: 'localhost', port: 6379 },
});

Graceful Degradation Best Practices

Fallback chaining: Try cached data first, then a compensation event, then human approval — each step is a more serious escalation.
Compensation events must be idempotent. The compensation worker should check if the action was already taken before executing.
Human-in-the-loop: Route permanently failed jobs to a manual-review queue with rich context (full error chain, attempt timestamps, payload snapshot) for operator decision-making.

Timeout-Aware Retry Strategies

Your job consistently takes 45 seconds. The lock duration is 30 seconds. BullMQ declares the job stalled. But the job isn't stalled — it's just slow. It's also not failed, but on retry it will be slow again because the timeout is wrong, not the business logic.

The fix: distinguish timeout failures (the job took too long) from logical failures (the job completed logic but produced an error), and treat each differently.

Timeout-Aware Worker with Distinct Error Types

import { Worker, Job } from 'bullmq';

class TimeoutError extends Error {
  constructor(message: string) {
    super(message);
    this.name = 'TimeoutError';
  }
}

class LogicalError extends Error {
  constructor(message: string) {
    super(message);
    this.name = 'LogicalError';
  }
}

function withTimeout<T>(promise: Promise<T>, ms: number, label: string): Promise<T> {
  return Promise.race([
    promise,
    new Promise<T>((_, reject) =>
      setTimeout(() => reject(new TimeoutError(`Operation "${label}" timed out after ${ms}ms`)), ms),
    ),
  ]);
}

async function processReportGeneration(data: { reportId: string }): Promise<{ url: string }> {
  const queryResult = await withTimeout(
    runDatabaseQuery(data.reportId),
    15_000,
    'database query',
  );

  const renderedReport = await withTimeout(
    renderReportTemplate(queryResult),
    20_000,
    'template rendering',
  );

  const uploadUrl = await withTimeout(
    uploadToStorage(renderedReport),
    10_000,
    'storage upload',
  );

  return { url: uploadUrl };
}

const worker = new Worker('reports', async (job: Job) => {
  try {
    const result = await processReportGeneration(job.data);
    return result;
  } catch (err) {
    if (err instanceof TimeoutError) {
      // Transient — retry with backoff
      await job.log(`[TIMEOUT] ${err.message}`);
      throw err;
    }
    if (err instanceof LogicalError) {
      // Permanent — don't waste retries
      await job.log(`[LOGICAL_ERROR] ${err.message}`);
      // Use UnrecoverableError for BullMQ >= v5
      throw err;
    }
    throw err;
  }
}, {
  connection: { host: 'localhost', port: 6379 },
});

Per-Attempt Timeout vs Total Job Timeout

When adding a job, configure attempts and backoff with timeouts in mind:

import { Queue } from 'bullmq';

const queue = new Queue('reports', {
  connection: { host: 'localhost', port: 6379 },
});

await queue.add('generate-report', { reportId: 'R-123' }, {
  attempts: 5,
  backoff: {
    type: 'exponential',
    delay: 2000,
  },
  removeOnFail: {
    age: 24 * 3600, // Keep failed jobs for 24 hours
    count: 100,
  },
});

Timeout Best Practices

Lock duration vs execution timeout: Set lockDuration high enough that the worker can complete its longest known sub-operation, or use job.extendLock() for genuinely long-running jobs. The Promise.race timeout is a business timeout — separate from BullMQ's lock mechanism.
Separate error types enable targeted monitoring. Prefix failed reasons with [TIMEOUT] or [LOGICAL] so you can filter by pattern in QueueHub.

Retry Deduplication with Idempotency Tokens

A job charges a credit card, then crashes after the charge succeeds but before BullMQ records the completion. The job is still in active state. BullMQ's stall detection re-queues the job. The worker retries it — and charges the credit card a second time.

This is different from BullMQ's built-in deduplication API (which prevents duplicate enqueueing). This pattern prevents duplicate side-effects when retrying.

The Idempotency Token Pattern

At the start of the processor function, check a token in Redis using SET NX with a TTL. If the token already exists, the side-effect was already completed — skip it:

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

const redis = new Redis({ host: 'localhost', port: 6379 });

function idempotencyKey(jobId: string, operation: string): string {
  return `idempotency:${operation}:${jobId}`;
}

async function executeOnce<T>(
  job: Job,
  operation: string,
  fn: () => Promise<T>,
  ttlMs: number = 300_000,
): Promise<T> {
  const key = idempotencyKey(job.id!, operation);

  // Try to acquire the idempotency lock
  const acquired = await redis.set(key, '1', 'PX', ttlMs, 'NX');
  if (acquired !== 'OK') {
    // The operation already completed — skip
    console.warn(`[IDEMPOTENCY] Skipping duplicate execution of ${operation} for job ${job.id}`);
    return { skipped: true, reason: 'already_executed' } as unknown as T;
  }

  try {
    const result = await fn();
    // Optionally cache the result for subsequent retries
    await redis.set(`${key}:result`, JSON.stringify(result), 'PX', ttlMs);
    return result;
  } catch (err) {
    // Operation failed — remove the token so the next retry can try again
    await redis.del(key);
    throw err;
  }
}

Wiring into your Worker

interface PaymentChargeData {
  userId: string;
  amount: number;
}

async function chargeStripeCustomer(data: PaymentChargeData): Promise<{ chargeId: string; amount: number }> {
  const response = await fetch('https://api.stripe.com/v1/charges', {
    method: 'POST',
    headers: {
      'Authorization': `Bearer sk_test_...`,
      'Content-Type': 'application/x-www-form-urlencoded',
      'Idempotency-Key': `charge-${data.userId}-${data.amount}`,
    },
    body: new URLSearchParams({
      amount: String(Math.round(data.amount * 100)),
      currency: 'usd',
      customer: data.userId,
    }),
  });

  if (!response.ok) {
    throw new Error(`Stripe charge failed: ${response.status}`);
  }

  return response.json() as Promise<{ chargeId: string; amount: number }>;
}

const worker = new Worker<PaymentChargeData>('payments', async (job: Job<PaymentChargeData>) => {
  return executeOnce(job, 'stripe-charge', () => chargeStripeCustomer(job.data));
}, {
  connection: { host: 'localhost', port: 6379 },
});

Idempotency Best Practices

Token TTL must exceed the maximum expected job processing time. This covers the gap between side-effect completion and BullMQ acknowledgment. 5 minutes is a good default.
Clean up tokens after successful completion. Use QueueEvents.on('completed') to explicitly delete tokens for high-throughput systems, though natural TTL expiry also works.
Combine with API-level idempotency keys. Stripe's Idempotency-Key header and your Redis-level token provide defense in depth. The Redis token catches cases where the API's idempotency window expires.
Use jobId + operationName as the key. Two different jobs should never share a token. SET NX prevents race conditions between concurrent retries.

Testing Retry Scenarios End-to-End

You've implemented checkpointing, circuit breakers, fallbacks, timeouts, and idempotency tokens. How do you know they work correctly? Manual testing with a running Redis instance is slow, non-deterministic, and doesn't cover edge cases like "circuit opens mid-batch."

Failure Simulation Strategy

Define a type that describes every failure scenario, then create a factory function that returns a simulated processor:

import { Job } from 'bullmq';

export type SimulationBehavior =
  | { type: 'fail'; onAttempt: number; error: Error }
  | { type: 'crashBeforeComplete'; onAttempt: number }
  | { type: 'partialThenFail'; onAttempt: number; checkpointAfter: number; error: Error }
  | { type: 'timeout'; onAttempt: number; durationMs: number }
  | { type: 'succeed' };

export function createSimulatedProcessor(
  behavior: SimulationBehavior,
  onProgress?: (job: Job, current: number) => Promise<void>,
) {
  return async (job: Job): Promise<{ simulated: true; attempt: number }> => {
    const attempt = job.attemptsMade + 1;

    if (behavior.type === 'succeed') {
      return { simulated: true, attempt };
    }

    if (behavior.type === 'fail' && behavior.onAttempt === attempt) {
      throw behavior.error;
    }

    if (behavior.type === 'crashBeforeComplete' && behavior.onAttempt === attempt) {
      throw new Error('Simulated crash');
    }

    if (behavior.type === 'partialThenFail' && behavior.onAttempt === attempt) {
      for (let i = 0; i < behavior.checkpointAfter; i++) {
        await onProgress?.(job, i);
      }
      throw behavior.error;
    }

    if (behavior.type === 'timeout' && behavior.onAttempt === attempt) {
      await new Promise((resolve) => setTimeout(resolve, behavior.durationMs));
      throw new Error(`Simulated timeout after ${behavior.durationMs}ms`);
    }

    // Default: succeed
    return { simulated: true, attempt };
  };
}

Integration Test for Checkpointing

Using Testcontainers for a real Redis instance:

import { describe, it, beforeAll, afterAll, expect } from 'vitest';
import { RedisContainer } from '@testcontainers/redis';
import { Worker, Queue, QueueEvents, Redis } from 'bullmq';
import { JobCheckpoint } from './JobCheckpoint';

describe('Job Checkpointing Retry', () => {
  let connection: Redis;
  let container: StartedRedisContainer;

  beforeAll(async () => {
    container = await new RedisContainer('redis:7-alpine')
      .withExposedPorts(6379)
      .start();

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

  afterAll(async () => {
    await connection.quit();
    await container.stop();
  });

  it('resumes from last checkpoint on retry', async () => {
    const queue = new Queue('checkpoint-test', { connection });
    const events = new QueueEvents('checkpoint-test', { connection });

    let checkpointHistory: number[] = [];

    const worker = new Worker('checkpoint-test', async (job) => {
      const checkpoint = new JobCheckpoint(job);
      let offset = (job.data as any).checkpoint?.offset ?? 0;
      const batchSize = 10;

      for (let i = offset; i < offset + batchSize; i++) {
        if (job.attemptsMade === 0 && i === 5) {
          await checkpoint.save({ offset: i + 1, processedCount: i + 1, cursor: null, lastUpdated: Date.now() });
          checkpointHistory.push(i + 1);
          throw new Error('Simulated failure at item 5');
        }
        await checkpoint.save({ offset: i + 1, processedCount: i + 1, cursor: null, lastUpdated: Date.now() });
        checkpointHistory.push(i + 1);
      }

      return { processed: true };
    }, {
      connection,
      concurrency: 1,
    });

    await queue.add('resumable-job', { checkpoint: null }, { attempts: 2 });

    const result = await new Promise<any>((resolve, reject) => {
      events.on('completed', ({ returnvalue }) => resolve(returnvalue));
      events.on('failed', ({ failedReason }) => reject(new Error(failedReason)));
    });

    // Attempt 1: items 0-4, fail at item 5 (checkpoint saves 5)
    // Attempt 2: resume from offset 5, process items 5-9
    expect(checkpointHistory).toEqual([1, 2, 3, 4, 5, 6, 7, 8, 9, 10]);

    await worker.close();
    await queue.close();
    await events.close();
  }, 15_000);
});

Integration Test for Circuit Breaker

describe('Circuit Breaker with Retry', () => {
  let connection: Redis;
  let container: StartedRedisContainer;

  beforeAll(async () => {
    container = await new RedisContainer('redis:7-alpine')
      .withExposedPorts(6379)
      .start();

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

  afterAll(async () => {
    await connection.quit();
    await container.stop();
  });

  it('fails fast (UnrecoverableError) when circuit is open', async () => {
    const queue = new Queue('circuit-test', { connection });
    const events = new QueueEvents('circuit-test', { connection });

    // Custom breaker configured to open after 2 failures
    const breaker = new CircuitBreakerState({
      failureThreshold: 2,
      successThreshold: 1,
      timeoutMs: 5000,
      resetTimeoutMs: 60_000,
      name: 'test-api',
    });

    // Manually open the breaker
    for (let i = 0; i < 3; i++) {
      try {
        await breaker.call(async () => { throw new Error('fail'); });
      } catch { /* expected */ }
    }

    const worker = new Worker('circuit-test', async (job) => {
      try {
        return await breaker.call(async () => {
          throw new Error('API error');
        });
      } catch (err) {
        if (breaker.isOpen) {
          throw new UnrecoverableError(
            `Service unavailable — circuit breaker is open. ${(err as Error).message}`,
          );
        }
        throw err;
      }
    }, { connection, concurrency: 1 });

    await queue.add('breaker-job', {}, { attempts: 5 });

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

    // Should fail on first attempt — not retry 5 times
    expect(failedReason).toContain('circuit breaker is open');

    // Verify only 1 attempt was made
    const jobs = await queue.getJobs(['failed']);
    expect(jobs[0]?.attemptsMade).toBe(1);

    await worker.close();
    await queue.close();
    await events.close();
  }, 15_000);
});

Testing Best Practices

Minimum test matrix per pattern:
- Checkpointing: resume after crash, resume after partial progress, no checkpoint data on first attempt
- Circuit breaker: open → fail fast, half-open → single probe, closed → normal operation
- Fallback: executes when retries exhausted, doesn't execute on intermediate attempts, compensation enqueued
- Timeout: timeout error retries, logical error does not, timeout message in failed reason
- Idempotency: token prevents duplicate execution, token removed on failure, token TTL expires correctly
Use separate Redis DB per test file (connectionOptions.db = 1) to avoid cross-test interference.
Prefer attempts: 2 with no backoff for fast test scenarios rather than mocking BullMQ internals.

QueueHub — See These Patterns in Production

QueueHub surfaces each pattern in its dashboard views:

Job detail view shows the progress field across retry attempts — verify checkpointing is resuming correctly by watching progress climb across attempts.
Failed jobs view shows failedReason — filter by [TIMEOUT], [LOGICAL], or circuit breaker is open to create targeted monitoring views.
Live tail view streams job.log() output — see checkpoint save events, circuit breaker transition messages, and fallback execution confirmations in real time.
Worker view shows active jobs — correlate circuit breaker "open" events with worker idle patterns.

Conclusion

Basic retry configuration answers when and how fast. These six patterns answer the harder questions:

Checkpointing turns wasted retries into resume points — use job.updateProgress() with a checkpoint token for any job that processes work incrementally.
Circuit breakers prevent retries from beating on a dead dependency — wire Opossum (or a custom state machine) into the processor and throw UnrecoverableError when the circuit is open.
Graceful degradation gives your system an answer for "what now?" — register fallback functions that return cached data, enqueue compensation events, or route to human review.
Timeout awareness separates transient slowness from logical failure — use Promise.race with distinct error types so each gets the right retry treatment.
Idempotency tokens prevent the most dangerous retry bug — duplicate side-effects — by using Redis SET NX before executing the side-effect.
Testing each pattern with a failure-simulation strategy gives you confidence that your retry pipeline handles every edge case.

Try QueueHub at queuehub.tech to monitor these patterns in production — watch checkpoint progress across retries, inspect circuit breaker states in job logs, and set up alerts for fallback execution counts.

Subscribe to the QueueHub blog for upcoming posts on OpenTelemetry tracing for job flows and Redis cluster benchmarking.