Idempotent Jobs

At-least-once delivery means a job will eventually run twice, so the handler has to be safe to repeat.

Syntaxqueue.add(name, data, { jobId?: string, deduplication?: { id: string; ttl?: number } })

BullMQ delivers a job at least once, not exactly once. A worker takes a job, charges a card, and is killed before it can mark the job completed. Its lock expires, the stalled check returns the job to wait, and another worker charges the card again. A retry after a half-finished attempt does the same thing. This is not a bug to configure away; it is the cost of durability.

Idempotent means repeatable

A job is idempotent when running it twice leaves the system exactly as running it once would. There are two places to work on that, and you usually want both.

Stop the duplicate being queued

Give the job a deterministic jobId. Adding a job with an id that already exists does nothing — no second job. deduplication: { id, ttl } does a similar thing with its own expiring key instead of relying on the job staying in Redis.

Make the body safe anyway

Dedupe at the edge is a filter, not a guarantee: once a completed job is removed its id is free again. So make the effect repeatable. Use a natural key with a unique constraint, or write a marker row keyed by job.id inside the same transaction as the effect, so either both land or neither does. Checking "has this run?" in a separate query before acting is not enough — two workers both read "no".

Example

Example · typescript
// src/payments/processed-job.entity.ts
import { Column, Entity, PrimaryColumn } from 'typeorm';

@Entity('processed_jobs')
export class ProcessedJob {
  // The primary key is the whole mechanism: a second insert of the same job
  // id cannot succeed, so the transaction around it cannot commit twice.
  @PrimaryColumn({ name: 'job_id', type: 'text' })
  jobId!: string;

  @Column({ type: 'text' })
  queue!: string;

  @Column({ name: 'processed_at', type: 'timestamptz', default: () => 'now()' })
  processedAt!: Date;
}

// src/payments/charge.processor.ts
import { Processor, WorkerHost } from '@nestjs/bullmq';
import { Logger } from '@nestjs/common';
import { Job } from 'bullmq';
import { DataSource, QueryFailedError } from 'typeorm';
import { Payment } from './payment.entity';
import { ProcessedJob } from './processed-job.entity';

interface ChargeJob {
  orderId: string;
  amountCents: number;
}

@Processor('payments', { concurrency: 4 })
export class ChargeProcessor extends WorkerHost {
  private readonly logger = new Logger(ChargeProcessor.name);

  constructor(private readonly dataSource: DataSource) {
    super();
  }

  async process(job: Job<ChargeJob>): Promise<void> {
    const jobId = job.id as string;

    if (job.attemptsStarted > 1) {
      this.logger.warn(`${jobId} is on try ${job.attemptsStarted}`);
    }

    try {
      // The marker and the effect commit together or not at all.
      await this.dataSource.transaction(async (manager) => {
        await manager.insert(ProcessedJob, { jobId, queue: 'payments' });

        await manager.insert(Payment, {
          orderId: job.data.orderId,
          amountCents: job.data.amountCents,
          status: 'captured',
        });
      });
    } catch (error) {
      // 23505 is PostgreSQL's unique_violation: the marker row already exists,
      // so an earlier attempt committed. The transaction rolled back, nothing
      // was charged twice, and there is nothing left to do.
      if (
        error instanceof QueryFailedError &&
        (error.driverError as { code?: string }).code === '23505'
      ) {
        this.logger.log(`${jobId} already processed; skipping`);
        return;
      }

      throw error; // anything else is a real failure: let BullMQ retry
    }
  }
}

When to use it

  • Stop a double-clicked checkout from charging a customer twice.
  • Collapse a burst of 200 product edits into a single reindex job per tenant.
  • Make a worker crash mid-job harmless, so the retry does not re-send an invoice.
  • Let a cron fire in all three replicas while only one job actually gets queued.

More examples

A deterministic jobId collapses duplicate adds

This is the cheapest dedupe there is, and it is where to start whenever the caller can retry: a double-clicked button, a client with retries, a cron that fires in every replica. The retention comment is the part that bites in production - dedupe by jobId lasts exactly as long as you keep the job.

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

@Injectable()
export class OrdersService {
  constructor(@InjectQueue('payments') private readonly payments: Queue) {}

  // Safe to call from a retried HTTP request, or from two replicas at once.
  async requestCharge(orderId: string, amountCents: number): Promise<void> {
    await this.payments.add(
      'charge',
      { orderId, amountCents },
      {
        // One job per order, ever - while the job still exists in Redis.
        // Must not be all digits and must not contain ':'.
        jobId: `charge-${orderId}`,
        attempts: 5,
        backoff: { type: 'exponential', delay: 2000 },
        // Keeping completed jobs keeps their ids taken. removeOnComplete: 1000
        // (shorthand for { count: 1000 }) would free the id after 1000 more
        // jobs, and the same charge could then be queued again.
        removeOnComplete: false,
        removeOnFail: false,
      },
    );
  }
}

Deduplication with a TTL for a chatty webhook

jobId dedupes forever (as long as the job lives); a deduplication TTL dedupes for a window, which is what throttling wants. Reach for it for "something changed, rebuild X" work, where coalescing a burst into one run is correct and losing the extra triggers costs nothing.

Example · typescript
// src/search/reindex.service.ts
import { Injectable } from '@nestjs/common';
import { InjectQueue } from '@nestjs/bullmq';
import { Queue } from 'bullmq';

@Injectable()
export class ReindexService {
  constructor(@InjectQueue('search') private readonly search: Queue) {}

  // A tenant that edits 200 products in a minute should not trigger 200 reindexes.
  async requestReindex(tenantId: string): Promise<void> {
    const job = await this.search.add(
      'reindex',
      { tenantId },
      {
        // While a key for this id is alive, further adds with the same id are
        // dropped. The key expires on its own after ttl milliseconds, unlike
        // a jobId, whose uniqueness depends on the job staying in Redis.
        deduplication: { id: `reindex-${tenantId}`, ttl: 60_000 },
        attempts: 3,
      },
    );

    // Set on jobs that carry a deduplication key, so you can log the window.
    if (job.deduplicationId) {
      // job.removeDeduplicationKey() would release it early.
    }
  }
}

Passing an idempotency key to the third party

The key must be derived from stable job data; a random UUID generated inside process() changes on every retry and defeats the whole mechanism. Use this for any outbound call with a side effect - payments, email sends, SMS - where the vendor offers an idempotency key.

Example · typescript
// src/payments/psp.service.ts
import { Injectable } from '@nestjs/common';
import { Job } from 'bullmq';

@Injectable()
export class PspService {
  // Your database transaction cannot roll back someone else's charge, so the
  // remote side needs its own dedupe. Most payment APIs accept a key and
  // return the ORIGINAL response for a repeated one.
  async charge(job: Job<{ orderId: string; amountCents: number }>): Promise<string> {
    const response = await fetch('https://api.example-psp.test/v1/charges', {
      method: 'POST',
      headers: {
        'content-type': 'application/json',
        // Derived from the job, not from Date.now() or a random value, so
        // every retry of this job sends the same key.
        'idempotency-key': `order-${job.data.orderId}`,
        authorization: `Bearer ${process.env.PSP_SECRET_KEY}`,
      },
      body: JSON.stringify({ amount: job.data.amountCents, currency: 'eur' }),
    });

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

    const charge = (await response.json()) as { id: string };
    return charge.id;
  }
}

Related

nest queues bullmqnest queue retriesnest queue processorsnest transactionsnest cloud queues

Discussion

  • Be the first to comment on this lesson.