Idempotency of in-flight operations in NestJS

← Back to the section

When NestJS receives SIGTERM, it starts an orderly shutdown: it drains the HTTP server, runs the beforeApplicationShutdown and onApplicationShutdown hooks, and after 60 seconds the process is killed forcibly anyway. If an operation was in progress at that moment, it is interrupted halfway.

The problem is not the interruption itself, but that on a pod restart the operation will run again: Kubernetes will restart the pod, the client will retry the request, Kafka will offer the same message again. If the operation is not ready to be re-run, a double effect occurs: two orders, a double charge, a duplicated email.

Idempotency is the property of an operation to produce the same result no matter how many times it is re-run. It is implemented differently depending on the context: HTTP, Kafka, or outbox.

HTTP POST and Idempotency-Key

HTTP GET and DELETE are idempotent on their own. POST is not: each call creates a new entity. To protect POST from duplicates, the client sends an Idempotency-Key header — a unique identifier of a specific business intent. The server remembers the response by this key and, on a repeated request, returns the saved result without re-running the logic.

How it works:

The guard checks the key before the business logic:

@Injectable()
export class IdempotencyGuard implements CanActivate {
  constructor(private readonly db: Pool) {}

  async canActivate(ctx: ExecutionContext): Promise<boolean> {
    const req = ctx.switchToHttp().getRequest<Request>();
    const key = req.headers['idempotency-key'] as string | undefined;
    if (!key) throw new BadRequestException('Idempotency-Key required');

    const existing = await this.db.query<{ response_body: unknown }>(
      'SELECT response_body FROM idempotency_record WHERE key = $1',
      [key],
    );
    if (existing.rows.length > 0) {
      ctx.switchToHttp().getResponse().json(existing.rows[0].response_body);
      return false;
    }
    req['idempotencyKey'] = key;
    return true;
  }
}

After successful processing, the controller saves the response:

@Post('/orders')
@UseGuards(IdempotencyGuard)
async create(
  @Headers('idempotency-key') key: string,
  @Body() dto: CreateOrderDto,
): Promise<OrderResponse> {
  const order = await this.orderService.create(dto);
  await this.db.query(
    'INSERT INTO idempotency_record(key, response_body) VALUES($1, $2) ON CONFLICT DO NOTHING',
    [key, order],
  );
  return order;
}

Without an Idempotency-Key, every repeated request on retry creates a new record — for one and the same client action.

Outbound HTTP with retry

The same problem arises when NestJS itself calls an external service with an automatic retry. If SIGTERM arrives during such a call, two requests may reach the provider — and both will be processed:

// Dangerous: on retry the provider gets two identical requests
async chargeCustomer(orderId: string, amount: Money): Promise<Receipt> {
  return this.httpService.axiosRef.post(
    `${this.paymentUrl}/charge`,
    { orderId, amount },
  ).then(r => r.data);
}

The right way is to pass an idempotency key, generated once per business operation, through all attempts:

async chargeCustomer(idempotencyKey: string, orderId: string, amount: Money): Promise<Receipt> {
  return this.httpService.axiosRef.post(
    `${this.paymentUrl}/charge`,
    { orderId, amount },
    { headers: { 'Idempotency-Key': idempotencyKey } },
  ).then(r => r.data);
}

In this case idempotencyKey is, for example, event.eventId from Kafka: it is immutable across retries and uniquely identifies the business operation.

kafkajs: processed_event in the same transaction

Kafka guarantees at-least-once delivery. On restart, the consumer may receive the same message again — especially if the offset didn't get committed before SIGTERM. The protection is a table of already-processed events, processed_event.

The key rule: the insert into processed_event and the side effect (the data change) must run in one transaction. If the transaction did not commit, everything will repeat correctly on the next run. If it did commit, a repeated insert will conflict on the unique index, and the handler will return early without duplication.

Implementation:

@Injectable()
export class OrderBillingConsumer implements OnApplicationBootstrap, BeforeApplicationShutdown {
  private consumer!: Consumer;

  constructor(
    private readonly kafka: Kafka,
    private readonly db: Pool,
  ) {}

  async onApplicationBootstrap(): Promise<void> {
    this.consumer = this.kafka.consumer({ groupId: 'billing-svc' });
    await this.consumer.connect();
    await this.consumer.subscribe({ topic: 'orders.confirmed', fromBeginning: false });

    await this.consumer.run({
      eachBatch: async ({ batch, resolveOffset, commitOffsetsIfNecessary, isRunning }) => {
        for (const message of batch.messages) {
          if (!isRunning()) break;

          const event: OrderConfirmedEvent = JSON.parse(message.value!.toString());
          const client = await this.db.connect();
          try {
            await client.query('BEGIN');
            const dedup = await client.query(
              'INSERT INTO processed_event(event_id, handler) VALUES($1, $2) ON CONFLICT DO NOTHING RETURNING event_id',
              [event.eventId, 'billing'],
            );
            if (dedup.rowCount === 0) {
              await client.query('ROLLBACK');
            } else {
              await client.query(
                'UPDATE "order" SET status = $1 WHERE id = $2',
                ['BILLED', event.orderId],
              );
              await client.query('COMMIT');
            }
          } catch (err) {
            await client.query('ROLLBACK');
            throw err;
          } finally {
            client.release();
          }

          resolveOffset(message.offset);
          await commitOffsetsIfNecessary();
        }
      },
    });
  }

  async beforeApplicationShutdown(): Promise<void> {
    await Promise.race([
      this.consumer.disconnect(),
      new Promise((_, reject) => setTimeout(() => reject(new Error('kafka timeout')), 20_000).unref()),
    ]);
  }
}

A common mistake is to call resolveOffset and commitOffsetsIfNecessary before the transaction finishes:

// Wrong: the offset is committed before the side effect
for (const message of batch.messages) {
  resolveOffset(message.offset);
  await commitOffsetsIfNecessary();   // the offset moved forward
  await this.processEvent(event);     // may not finish on SIGTERM
}

If SIGTERM comes after committing the offset but before processEvent completes, the side effect is lost with no chance of replay.

Outbox relay: a two-phase status

The outbox pattern means the service first writes events into an outbox_event table, and a separate relay process sends them to Kafka. A problem arises if SIGTERM comes between sending to Kafka and marking the event as PUBLISHED:

The solution is an intermediate PUBLISHING status. The event moves into it before sending; on SIGTERM between phases it stays in PUBLISHING and does not fall into the next PENDING selection. A separate cleanup job once an hour returns stuck PUBLISHING rows back to PENDING:

@Injectable()
export class OutboxRelayService implements BeforeApplicationShutdown {
  private running = false;
  private currentBatch: Promise<void> = Promise.resolve();

  constructor(
    private readonly db: Pool,
    private readonly producer: Producer,
  ) {}

  @Interval(1_000)
  async relay(): Promise<void> {
    if (this.running) return;
    this.running = true;
    this.currentBatch = this.processBatch().finally(() => { this.running = false; });
    await this.currentBatch;
  }

  private async processBatch(): Promise<void> {
    const client = await this.db.connect();
    try {
      // Phase 1: pick up PENDING and mark PUBLISHING
      const { rows } = await client.query<{ id: string; topic: string; payload: unknown }>(
        `UPDATE outbox_event SET status='PUBLISHING', locked_at=now()
         WHERE id IN (
           SELECT id FROM outbox_event WHERE status='PENDING' LIMIT 20 FOR UPDATE SKIP LOCKED
         )
         RETURNING id, topic, payload`,
      );

      for (const row of rows) {
        // Phase 2: send to Kafka
        await this.producer.send({
          topic: row.topic,
          messages: [{ value: JSON.stringify(row.payload), headers: { eventId: row.id } }],
        });
        // Phase 3: mark PUBLISHED
        await client.query(
          `UPDATE outbox_event SET status='PUBLISHED', published_at=now() WHERE id=$1`,
          [row.id],
        );
      }
    } finally {
      client.release();
    }
  }

  async beforeApplicationShutdown(): Promise<void> {
    await this.currentBatch;
  }
}

If SIGTERM comes between Phase 2 and Phase 3 — Kafka has already received the event, but in the database it is in PUBLISHING. After cleanup it will be sent again. That is why the consumer side must still deduplicate via processed_event — then a duplicate in Kafka is safe.

An alternative is to not use the PUBLISHING status at all and instead make consumer-side deduplication mandatory. Then the relay can send duplicates without worry, and operationally it is simpler.

In short

• Graceful shutdown gives time to finish an operation but does not guarantee it: on a force-kill after 60 seconds the operation is interrupted, and on restart it runs again.
• HTTP POST: the client sends an Idempotency-Key, the NestJS guard checks the key and returns the cached response on a repeated request.
• Outbound HTTP with retry: the idempotency key is generated once per business operation and passed into all attempts — otherwise the provider processes every retry independently.
• kafkajs handler: processed_event is inserted in the same transaction as the side effect; a uniqueness conflict on repeat = an early return without duplication.
• resolveOffset and commitOffsetsIfNecessary — only after the transaction commits: otherwise the offset moves forward and the side effect is lost with no replay.
• Outbox relay: an intermediate PUBLISHING status protects against a double send; a cleanup job returns stuck rows to PENDING.
• Idempotency is needed across the whole chain: client → NestJS handler → downstream service.

What to read next

• NestJS graceful shutdown configuration — enableShutdownHooks, force-deadline, the readiness flag.
• HTTP drain — server.close(), closeIdleConnections(), preStop sleep.
• Kafka shutdown — consumer.disconnect() with a timeout, eachBatch semantics.
• Database and persistence — the order of pool.end() / dataSource.destroy() in onApplicationShutdown.
• Budgets and observability — how to measure the actual shutdown duration.