NestJS microservices: reliable reconnection, retries and failure handling

import 'reflect-metadata';
import { Controller, Logger, Module } from '@nestjs/common';
import { NestFactory } from '@nestjs/core';
import { MessagePattern, Payload, RpcException, Transport, type MicroserviceOptions } from '@nestjs/microservices';
import { setTimeout as sleep } from 'node:timers/promises';

interface ChargeCommand { idempotencyKey: string; orderId: string; amountCents: number; delayMs?: number }
interface ChargeResult { chargeId: string; orderId: string; amountCents: number }

/** In production this is a table with a unique key or Redis SET NX with a TTL, not process memory. */
const processed = new Map<string, Promise<ChargeResult>>();
let executions = 0;

@Controller()
class PaymentsController {
  private readonly logger = new Logger('Payments');

  @MessagePattern('payments.charge')
  charge(@Payload() cmd: ChargeCommand): Promise<ChargeResult> {
    if (!cmd?.idempotencyKey || !Number.isInteger(cmd.amountCents) || cmd.amountCents <= 0) {
      throw new RpcException({ code: 'INVALID_CHARGE', message: 'idempotencyKey and a positive amountCents are required' });
    }
    const existing = processed.get(cmd.idempotencyKey);
    if (existing) {
      this.logger.log(`duplicate ${cmd.idempotencyKey}: returning the stored result`);
      return existing;
    }
    const work = (async () => {
      executions++;
      if (cmd.delayMs) await sleep(cmd.delayMs);
      return { chargeId: `ch_${executions}`, orderId: cmd.orderId, amountCents: cmd.amountCents };
    })();
    processed.set(cmd.idempotencyKey, work);
    work.catch(() => processed.delete(cmd.idempotencyKey)); // a failed attempt may be retried
    return work;
  }

  @MessagePattern('payments.executions')
  count() { return executions; }
}

@Module({ controllers: [PaymentsController] })
class PaymentsModule {}

const app = await NestFactory.createMicroservice<MicroserviceOptions>(PaymentsModule, {
  transport: Transport.TCP,
  options: { host: '127.0.0.1', port: Number(process.env.PORT ?? 4721) },
  logger: ['log', 'error', 'warn'],
});
app.enableShutdownHooks();
await app.listen();
console.log('READY');

import type { ClientProxy } from '@nestjs/microservices';
import { firstValueFrom, retry, timeout, timer, TimeoutError } from 'rxjs';

/** Errors that mean "the message may not have arrived": safe to retry an idempotent command. */
export function isTransientError(err: unknown): boolean {
  if (err instanceof TimeoutError) return true;
  const e = err as { code?: string; message?: string };
  return ['ECONNREFUSED', 'ECONNRESET', 'EPIPE'].includes(e?.code ?? '') || e?.message === 'Connection closed';
}

export function send<T>(client: Pick<ClientProxy, 'send'>, pattern: string, data: unknown, opts = { timeoutMs: 2000, retries: 6, baseDelayMs: 200, maxDelayMs: 2000 }) {
  return firstValueFrom(
    client.send<T>(pattern, data).pipe(
      timeout({ first: opts.timeoutMs }),
      retry({
        count: opts.retries,
        delay: (err, attempt) => {
          if (!isTransientError(err)) throw err; // business errors (RpcException) are not retried
          const backoff = Math.min(opts.baseDelayMs * 2 ** (attempt - 1), opts.maxDelayMs);
          const jitter = Math.random() * backoff * 0.2;
          console.log(`  retry ${attempt} in ${Math.round(backoff + jitter)} ms after: ${(err as Error).message ?? (err as { code?: string }).code}`);
          return timer(backoff + jitter);
        },
      }),
    ),
  );
}

import { Inject, Injectable, Module, type OnApplicationBootstrap } from '@nestjs/common';
import { ClientsModule, Transport, type ClientProxy } from '@nestjs/microservices';
import { send } from './call.js';

@Module({
  imports: [
    ClientsModule.register([
      { name: 'PAYMENTS', transport: Transport.TCP, options: { host: process.env.PAYMENTS_HOST ?? '127.0.0.1', port: 4721 } },
      // Without retryAttempts the Redis transport does not reconnect after the connection drops.
      { name: 'EVENTS', transport: Transport.REDIS, options: { host: process.env.REDIS_HOST ?? '127.0.0.1', port: 6379, retryAttempts: 20, retryDelay: 1000 } },
    ]),
  ],
  exports: [ClientsModule],
})
export class ClientsConfigModule {}

@Injectable()
export class CheckoutService implements OnApplicationBootstrap {
  constructor(@Inject('PAYMENTS') private readonly payments: ClientProxy) {}

  async onApplicationBootstrap() {
    this.payments.status.subscribe((s) => console.log(`payments client: ${s}`));
  }

  charge(orderId: string, amountCents: number) {
    // One key per business operation, reused on every retry of that operation.
    const idempotencyKey = `charge:${orderId}`;
    return send<{ chargeId: string }>(this.payments, 'payments.charge', { idempotencyKey, orderId, amountCents });
  }
}

import amqp from 'amqplib';

/** Run once per environment (or keep in your broker definitions file). */
export async function declareInvoiceTopology(url: string) {
  const conn = await amqp.connect(url);
  const ch = await conn.createChannel();
  await ch.assertExchange('invoices.dlx', 'direct', { durable: true });
  // Work queue: rejected messages go to the dead-letter exchange.
  await ch.assertQueue('invoices', {
    durable: true,
    deadLetterExchange: 'invoices.dlx',
    deadLetterRoutingKey: 'invoices.retry',
  });
  // Retry queue: no consumers; after 30 s a message dead-letters back to the work queue.
  await ch.assertQueue('invoices.retry', {
    durable: true,
    messageTtl: 30_000,
    deadLetterExchange: '',          // the default exchange routes by queue name
    deadLetterRoutingKey: 'invoices',
  });
  await ch.bindQueue('invoices.retry', 'invoices.dlx', 'invoices.retry');
  // Parking queue: messages a human has to look at.
  await ch.assertQueue('invoices.parked', { durable: true });
  await ch.close();
  await conn.close();
}

import 'reflect-metadata';
import { Controller, Logger, Module } from '@nestjs/common';
import { NestFactory } from '@nestjs/core';
import { Ctx, EventPattern, Payload, RmqContext, Transport, type MicroserviceOptions } from '@nestjs/microservices';

interface InvoiceRequested { idempotencyKey: string; invoiceId: string }
class PermanentError extends Error {}
const MAX_DELIVERIES = 5;

@Controller()
class InvoicesController {
  private readonly logger = new Logger('Invoices');

  @EventPattern<string>('invoice.requested')
  async handle(@Payload() event: InvoiceRequested, @Ctx() ctx: RmqContext) {
    const channel = ctx.getChannelRef();
    const msg = ctx.getMessage();
    try {
      await this.process(event);              // idempotent: redelivery must be harmless
      channel.ack(msg);
    } catch (err) {
      // RabbitMQ counts dead-lettering per queue and reason in the x-death header.
      const xDeath = (msg.properties.headers?.['x-death'] ?? []) as Array<{ queue: string; reason: string; count: number }>;
      const deaths = xDeath.find((d) => d.queue === 'invoices' && d.reason === 'rejected')?.count ?? 0;
      if (err instanceof PermanentError || deaths + 1 >= MAX_DELIVERIES) {
        this.logger.error(`parking ${event?.invoiceId}: ${(err as Error).message}`);
        channel.sendToQueue('invoices.parked', msg.content, {
          persistent: true,
          headers: { ...msg.properties.headers, 'x-error': (err as Error).message },
        });
        channel.ack(msg);                     // remove it from the work queue
      } else {
        channel.nack(msg, false, false);      // requeue=false: dead-letter to invoices.retry
      }
    }
  }

  private async process(event: InvoiceRequested) {
    if (!event?.invoiceId) throw new PermanentError('missing invoiceId');
    // ... look up event.idempotencyKey in a durable store, do the work, record the key ...
  }
}

@Module({ controllers: [InvoicesController] })
class WorkerModule {}

const app = await NestFactory.createMicroservice<MicroserviceOptions>(WorkerModule, {
  transport: Transport.RMQ,
  options: {
    urls: [process.env.AMQP_URL ?? 'amqp://localhost:5672'],
    queue: 'invoices',
    noAck: false,                  // manual acknowledgement
    prefetchCount: 10,             // at most 10 unacknowledged messages per consumer
    queueOptions: {                // must match the declared queue exactly
      durable: true,
      deadLetterExchange: 'invoices.dlx',
      deadLetterRoutingKey: 'invoices.retry',
    },
    socketOptions: { heartbeatIntervalInSeconds: 30, reconnectTimeInSeconds: 5 },
  },
});
app.enableShutdownHooks();
await app.listen();

import { Controller, Get, Module } from '@nestjs/common';
import { Transport, type RmqOptions, type TcpClientOptions } from '@nestjs/microservices';
import { HealthCheck, HealthCheckService, MicroserviceHealthIndicator, TerminusModule } from '@nestjs/terminus';

@Controller('health')
class HealthController {
  constructor(private readonly health: HealthCheckService, private readonly ms: MicroserviceHealthIndicator) {}

  /** Readiness: can this instance reach the services it depends on? */
  @Get('ready')
  @HealthCheck()
  ready() {
    return this.health.check([
      () => this.ms.pingCheck<TcpClientOptions>('payments', {
        transport: Transport.TCP,
        options: { host: process.env.PAYMENTS_HOST ?? '127.0.0.1', port: 4721 },
      }).withTimeout(1500),
      () => this.ms.pingCheck<RmqOptions>('rabbitmq', {
        transport: Transport.RMQ,
        options: { urls: [process.env.AMQP_URL ?? 'amqp://localhost:5672'] },
      }).withTimeout(1500),
    ]);
  }
}

@Module({ imports: [TerminusModule], controllers: [HealthController] })
export class HealthModule {}