Timeouts, Retries and Circuit Breakers

// src/resilience/circuit-breaker.ts
import { ServiceUnavailableException } from '@nestjs/common';

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

// No @Injectable() here: the constructor takes plain numbers, which the
// injector cannot resolve from types, so this class is registered with a
// factory provider instead (below).
export class CircuitBreaker {
  private state: BreakerState = 'closed';
  private failures = 0;
  private openedAt = 0;

  constructor(
    private readonly threshold: number, // consecutive failures before opening
    private readonly coolOffMs: number, // how long to stay open
  ) {}

  async run<T>(call: () => Promise<T>): Promise<T> {
    if (this.state === 'open') {
      if (Date.now() - this.openedAt < this.coolOffMs) {
        // Fail without touching the network. This is the whole point: the
        // queue of pending calls never forms in the first place.
        throw new ServiceUnavailableException('billing circuit is open');
      }
      this.state = 'half-open'; // let exactly one call through as a probe
    }

    try {
      const result = await call();
      this.failures = 0;
      this.state = 'closed';
      return result;
    } catch (err) {
      this.failures += 1;
      if (this.state === 'half-open' || this.failures >= this.threshold) {
        this.state = 'open';
        this.openedAt = Date.now();
      }
      throw err;
    }
  }
}

// src/billing/billing.module.ts
import { Module } from '@nestjs/common';
import { ClientsModule, Transport } from '@nestjs/microservices';
import { BillingClient } from '../gateway/billing.client.js';
import { CircuitBreaker } from '../resilience/circuit-breaker.js';

@Module({
  imports: [
    ClientsModule.register([
      { name: 'BILLING', transport: Transport.TCP, options: { port: 3001 } },
    ]),
  ],
  providers: [
    BillingClient,
    { provide: CircuitBreaker, useFactory: () => new CircuitBreaker(5, 10_000) },
  ],
  exports: [BillingClient],
})
export class BillingModule {}

// src/main.ts
import { NestFactory } from '@nestjs/core';
import { MicroserviceOptions, Transport } from '@nestjs/microservices';
import { AppModule } from './app.module.js';

async function bootstrap() {
  const app = await NestFactory.createMicroservice<MicroserviceOptions>(AppModule, {
    transport: Transport.RMQ,
    options: {
      urls: ['amqp://localhost:5672'],
      queue: 'orders',
      // At most 10 unacknowledged messages per consumer. This is the backpressure
      // valve: without it one consumer will accept the whole backlog at once.
      prefetchCount: 10,
      // noAck: false means Nest does not acknowledge for you, so the handler
      // below must ack or nack every message.
      noAck: false,
      queueOptions: { durable: true },
    },
  });
  await app.listen();
}

bootstrap();

// src/orders/orders.consumer.ts
import { Controller, Logger } from '@nestjs/common';
import { Ctx, EventPattern, Payload, RmqContext } from '@nestjs/microservices';
import { OrdersService } from './orders.service.js';

@Controller()
export class OrdersConsumer {
  private readonly logger = new Logger(OrdersConsumer.name);

  constructor(private readonly orders: OrdersService) {}

  @EventPattern('order.created')
  async handle(@Payload() data: { orderId: string }, @Ctx() ctx: RmqContext) {
    const channel = ctx.getChannelRef();
    const message = ctx.getMessage();

    try {
      await this.orders.provision(data.orderId);
      channel.ack(message);
    } catch (err) {
      this.logger.error(`order.created failed for ${data.orderId}`, err as Error);
      // requeue: false. Requeueing sends it back to the head of the queue and
      // you spin on the same poison message; route it to a dead-letter
      // exchange configured on the queue instead.
      channel.nack(message, false, false);
    }
  }
}

// src/billing/billing.controller.ts
import { Controller } from '@nestjs/common';
import { KafkaRetriableException, MessagePattern, Payload, RpcException } from '@nestjs/microservices';
import { BillingService } from './billing.service.js';

@Controller()
export class BillingController {
  constructor(private readonly billing: BillingService) {}

  @MessagePattern('billing.quote')
  async quote(@Payload() body: { orderId: string }) {
    try {
      return await this.billing.quote(body.orderId);
    } catch (err) {
      if (err instanceof RangeError) {
        // A permanent failure: reply with an error the caller can read.
        // Retrying this would fail identically every time.
        throw new RpcException({ status: 'error', message: 'unknown order' });
      }
      // A transient failure. KafkaRetriableException bubbles out to kafkajs's
      // eachMessage callback instead of being serialised into a reply, so the
      // broker redelivers the message rather than the caller seeing an error.
      throw new KafkaRetriableException('price service unavailable');
    }
  }
}

// src/gateway/billing.client.ts (excerpt)
// this.client is a ClientKafkaProxy obtained from ClientsModule.
// status emits: 'connected' | 'disconnected' | 'rebalancing' | 'crashed' | 'stopped'
// A 'rebalancing' burst explains a wave of timeouts that is nobody's bug.
this.client.status.subscribe(status => this.logger.log(`kafka ${status}`));