Retries and Dead Letters

attempts and backoff decide how often a failed job is retried; BullMQ has no dead-letter queue, so you build one.

Syntaxbackoff?: number | { type: 'fixed' | 'exponential'; delay: number; jitter?: number }

A job fails when process() throws. What happens next is decided entirely by the options the job was added with.

attempts and backoff

attempts is the total number of tries, and it defaults to 1 — so by default nothing is retried. backoff is the wait between tries. backoff: 5000 is shorthand for { type: 'fixed', delay: 5000 }; the two are the same thing. { type: 'exponential', delay: 1000 } waits 2 ^ (attemptsMade - 1) * delay milliseconds, so 1s, 2s, 4s. Add jitter between 0 and 1 to spread a thundering herd. Between attempts the job sits in the delayed set, not in your process, so a deploy does not lose it.

Failing on purpose

Some errors will never succeed: a 404 from the API, a malformed payload. Throw UnrecoverableError from bullmq and the job skips its remaining attempts and fails now.

Where a dead job goes

Nowhere. BullMQ has no dead-letter queue: a job that exhausts its attempts moves into the failed set and stays there, because removeOnFail defaults to keeping it. That set is your dead-letter store, readable with queue.getFailed() and replayable with job.retry(). If you want alerting or a separate queue, do it in an @OnWorkerEvent('failed') handler, after checking that this really was the last attempt.

Example

Example · typescript
// src/webhooks/webhook.producer.ts
import { Injectable } from '@nestjs/common';
import { InjectQueue } from '@nestjs/bullmq';
import { Queue } from 'bullmq';

@Injectable()
export class WebhookProducer {
  constructor(@InjectQueue('webhook') private readonly queue: Queue) {}

  async deliver(url: string, payload: unknown): Promise<void> {
    await this.queue.add(
      'deliver',
      { url, payload },
      {
        attempts: 5, // one try plus four retries
        // 2s, 4s, 8s, 16s, each shortened by up to 30% so a vendor outage does
        // not send every pending job back at the same instant.
        backoff: { type: 'exponential', delay: 2000, jitter: 0.3 },
        removeOnComplete: { count: 1000 },
        removeOnFail: { age: 14 * 24 * 60 * 60 }, // keep two weeks of failures (seconds)
      },
    );
  }
}

// src/webhooks/webhook.processor.ts
import { Processor, WorkerHost } from '@nestjs/bullmq';
import { Logger } from '@nestjs/common';
import { Job, UnrecoverableError } from 'bullmq';

@Processor('webhook', { concurrency: 10 })
export class WebhookProcessor extends WorkerHost {
  private readonly logger = new Logger(WebhookProcessor.name);

  async process(job: Job<{ url: string; payload: unknown }>): Promise<void> {
    const { url, payload } = job.data;

    const response = await fetch(url, {
      method: 'POST',
      headers: { 'content-type': 'application/json' },
      body: JSON.stringify(payload),
      signal: AbortSignal.timeout(10_000), // never hang on to the job lock
    });

    if (response.ok) {
      return;
    }

    // A 4xx that is not 429 means the request itself is wrong, and the next
    // four attempts would be wrong in the same way. Fail now and keep the
    // reason: UnrecoverableError skips the remaining attempts.
    if (response.status >= 400 && response.status < 500 && response.status !== 429) {
      throw new UnrecoverableError(`${url} rejected the delivery: ${response.status}`);
    }

    // attemptsStarted is the number of this try, starting at 1.
    this.logger.warn(`${url} returned ${response.status} on try ${job.attemptsStarted}`);

    // Any other throw is retryable: BullMQ moves the job to "delayed".
    throw new Error(`${url} returned ${response.status}`);
  }
}

When to use it

  • Retry a flaky payment webhook four times with exponential backoff instead of losing it.
  • Fail a job immediately on a 422 so it does not occupy the queue for twenty minutes.
  • Alert on a job only once it has exhausted every attempt, not on each failure.
  • Replay two hours of failed deliveries after a vendor outage is resolved.

More examples

A dead-letter queue built from the failed event

The attemptsMade check is the whole trick: without it you dead-letter a job four times on its way to succeeding on the fifth. Reach for a separate queue when a human or a different process must triage failures - the dead queue can have its own processor that files a ticket or posts to Slack.

Example · typescript
// src/webhooks/webhook.processor.ts (same class as above)
import { InjectQueue, OnWorkerEvent, Processor, WorkerHost } from '@nestjs/bullmq';
import { Job, Queue } from 'bullmq';

@Processor('webhook', { concurrency: 10 })
export class WebhookProcessor extends WorkerHost {
  constructor(
    @InjectQueue('webhook-dead') private readonly deadLetters: Queue,
  ) {
    super();
  }

  async process(job: Job): Promise<void> {
    // ...as above
  }

  // Fires after EVERY failed attempt, not only the last one.
  @OnWorkerEvent('failed')
  async onFailed(job: Job | undefined, error: Error): Promise<void> {
    if (!job) {
      return;
    }

    const limit = job.opts.attempts ?? 1;

    // attemptsMade counts the tries that have already failed. Below the limit,
    // BullMQ is going to retry on its own and this is not a dead job yet.
    if (job.attemptsMade < limit) {
      return;
    }

    await this.deadLetters.add(
      'dead',
      {
        originalQueue: 'webhook',
        originalJobId: job.id,
        name: job.name,
        data: job.data,
        failedReason: error.message,
        stacktrace: job.stacktrace,
      },
      { attempts: 1, removeOnComplete: false },
    );
  }
}

A custom backoff that honours Retry-After

The built-in strategies only know the attempt number. A custom one receives the thrown error too, which is how you obey a Retry-After header instead of guessing. Set it on the worker and select it per job with backoff: { type: 'custom' }; capping the result keeps a long outage from scheduling a retry days out.

Example · typescript
// src/webhooks/webhook.processor.ts
import { Processor, WorkerHost } from '@nestjs/bullmq';
import { Job, MinimalJob } from 'bullmq';

// A rate-limited vendor tells you when to come back. Encode that error, then
// let the backoff strategy read it.
export class RetryAfterError extends Error {
  constructor(public readonly retryAfterMs: number) {
    super(`rate limited, retry in ${retryAfterMs}ms`);
  }
}

@Processor('webhook', {
  concurrency: 10,
  settings: {
    // Named strategies are chosen with backoff: { type: 'custom' } on the job.
    // Return milliseconds; 0 means "back of the waiting list" and -1 means
    // "stop retrying and fail now".
    backoffStrategy: (
      attemptsMade: number,
      type?: string,
      err?: Error,
      job?: MinimalJob,
    ): number => {
      if (err instanceof RetryAfterError) {
        return err.retryAfterMs;
      }
      return Math.min(2 ** attemptsMade * 1000, 60_000); // capped exponential
    },
  },
})
export class WebhookProcessor extends WorkerHost {
  async process(job: Job<{ url: string }>): Promise<void> {
    const response = await fetch(job.data.url, { method: 'POST' });

    if (response.status === 429) {
      const retryAfter = Number(response.headers.get('retry-after') ?? 60);
      throw new RetryAfterError(retryAfter * 1000); // header is in seconds
    }
  }
}

An endpoint to inspect and replay failed jobs

The failed set is the dead-letter store, so you need a way to read and replay it. This is the shape to build after the first incident where a vendor was down for an hour: fix the cause, then retry-all. resetAttemptsMade is the detail people miss, which is why a replay sometimes fails again within milliseconds.

Example · typescript
// src/admin/failed-jobs.controller.ts
import { Controller, Get, NotFoundException, Param, Post, UseGuards } from '@nestjs/common';
import { InjectQueue } from '@nestjs/bullmq';
import { Queue } from 'bullmq';
import { AdminGuard } from '../auth/admin.guard';

@UseGuards(AdminGuard) // this is production data: do not ship it open
@Controller('admin/webhook/failed')
export class FailedJobsController {
  constructor(@InjectQueue('webhook') private readonly queue: Queue) {}

  @Get()
  async list() {
    // start and end are an inclusive index range over the failed set.
    const jobs = await this.queue.getFailed(0, 49);

    return jobs.map((job) => ({
      id: job.id,
      name: job.name,
      attemptsMade: job.attemptsMade,
      failedReason: job.failedReason,
      failedAt: job.finishedOn,
    }));
  }

  @Post(':id/retry')
  async retry(@Param('id') id: string) {
    const job = await this.queue.getJob(id);

    if (!job) {
      throw new NotFoundException(`No job ${id}`);
    }

    // Moves it from failed back to wait. Without resetAttemptsMade it keeps
    // its counter and may fail again immediately, having no attempts left.
    await job.retry('failed', { resetAttemptsMade: true });

    return { retried: id };
  }

  @Post('retry-all')
  async retryAll() {
    // Moves failed jobs back to wait in batches of count per iteration.
    await this.queue.retryJobs({ count: 500, state: 'failed' });
    return { ok: true };
  }
}

Related

nest queues bullmqnest queue processorsnest queue idempotencynest exception filtersnest logging

Discussion

  • Be the first to comment on this lesson.