# Configuration & Resilience (/integrations/nestjs/how-to/configuration-and-resilience)



This guide covers the production-hardening options of `@kubemq/nestjs-transport`: securing the connection, surviving broker restarts, controlling serialization, and protecting handlers with dead-letter routing, validation, idempotency, backpressure, and a client-side circuit breaker. Every option below is read directly from the transport's option interfaces, so the types and defaults match the shipped package.

All examples assume a running broker. For local development, start one with Docker:

<RunKubeMQ ports="[50000]" />

Port `50000` is the gRPC port the transport connects to. `@kubemq/nestjs-transport` is built on the native `kubemq-js` SDK, so no HTTP connector flag is required.

## Prerequisites [#prerequisites]

* `@kubemq/nestjs-transport` and `kubemq-js` installed, with `KubeMQServer` and/or `KubeMQModule` already registered in a NestJS app (see [Getting Started with NestJS](/integrations/nestjs/tutorials/getting-started))
* The broker above running and reachable
* `class-validator` and `class-transformer` installed if you use the `validate` handler option (see [Payload Validation](#payload-validation))

<Callout type="info">
  **Time units.** Fields ending in `*Seconds` (such as `callbackTimeoutSeconds`, `waitTimeoutSeconds`, `ttlSeconds`) are expressed in **seconds**. Fields ending in `*Ms` (such as `initialDelayMs`, `maxDelayMs`, `resetTimeout`) are expressed in **milliseconds**. Mixing them up is the most common configuration mistake.
</Callout>

## TLS & Authentication [#tls--authentication]

The `tls` and `credentials` options apply to both `KubeMQServer` (the microservice strategy) and the client registered via `KubeMQModule.register()`. Configure them on every connection that talks to a secured broker.

### TLS (server certificate verification) [#tls-server-certificate-verification]

Set `tls.enabled` and point `caCert` at the CA bundle that signed the broker certificate.

```typescript title="main.ts"
import { KubeMQServer } from '@kubemq/nestjs-transport';

const kubemqServer = new KubeMQServer({
  address: 'kubemq-server:50000',
  clientId: 'orders-server',
  tls: {
    enabled: true,
    caCert: process.env.KUBEMQ_CA_CERT ?? '/path/to/ca.pem',
  },
});
```

### Mutual TLS (mTLS) [#mutual-tls-mtls]

For mutual authentication, add `clientCert` and `clientKey` so the broker can verify the client identity as well.

```typescript title="main.ts"
const kubemqServer = new KubeMQServer({
  address: 'kubemq-server:50000',
  clientId: 'orders-server',
  tls: {
    enabled: true,
    caCert: process.env.KUBEMQ_CA_CERT ?? '/path/to/ca.pem',
    clientCert: process.env.KUBEMQ_CLIENT_CERT ?? '/path/to/client.pem',
    clientKey: process.env.KUBEMQ_CLIENT_KEY ?? '/path/to/client-key.pem',
  },
});
```

### Token authentication [#token-authentication]

When the broker enforces token (JWT) authentication, pass the token through `credentials`. Keep the token out of source — read it from an environment variable or secret store.

```typescript title="main.ts"
const kubemqServer = new KubeMQServer({
  address: 'kubemq-server:50000',
  clientId: 'orders-server',
  credentials: process.env.KUBEMQ_TOKEN,
});
```

<Callout type="info">
  `tls` and `credentials` are independent and can be combined: TLS (or mTLS) secures the transport channel, while `credentials` authorizes the client. The client side uses the same option names — apply identical settings in `KubeMQModule.register({ ... })` so both ends agree.
</Callout>

## Reconnection [#reconnection]

Both server and client accept a `reconnect` policy (a kubemq-js `ReconnectionPolicy`) that controls automatic recovery after a broker drop, plus `waitForConnection` to decide whether startup blocks until the first connection succeeds.

```typescript title="main.ts"
const kubemqServer = new KubeMQServer({
  address: 'localhost:50000',
  clientId: 'orders-server',
  waitForConnection: false,
  reconnect: {
    maxAttempts: -1,
    initialDelayMs: 500,
    maxDelayMs: 30_000,
    multiplier: 2.0,
    jitter: 'full',
  },
});
```

<TypeTable
  type="{
  maxAttempts: { type: 'number', default: '-1', description: 'Maximum reconnection attempts; -1 means unlimited.' },
  initialDelayMs: { type: 'number', default: '500', description: 'Initial backoff interval in milliseconds.' },
  maxDelayMs: { type: 'number', default: '30000', description: 'Maximum backoff interval in milliseconds.' },
  multiplier: { type: 'number', default: '2.0', description: 'Exponential backoff multiplier applied each attempt.' },
  jitter: { type: &#x22;'none' | 'full' | 'decorrelated'&#x22;, default: &#x22;'full'&#x22;, description: 'Jitter strategy applied to the backoff delay.' },
}"
/>

The `multiplier` grows the delay exponentially from `initialDelayMs` up to the `maxDelayMs` ceiling, and `jitter` randomizes each delay to avoid a thundering-herd reconnect storm when many clients reconnect at once.

**`waitForConnection`** (default: `true`) gates startup:

* `true` — `startAllMicroservices()` waits for a successful connection before completing. Startup fails fast if the broker is unreachable.
* `false` — startup completes immediately and the transport retries the connection in the background using the `reconnect` policy. Use this when the app must boot even if the broker is temporarily down.

```typescript title="main.ts (non-blocking startup)"
const kubemqServer = new KubeMQServer({
  address,
  clientId: 'orders-server',
  waitForConnection: false, // app starts even if the broker is down
  reconnect: { maxAttempts: -1, initialDelayMs: 1000, maxDelayMs: 30_000 },
});

app.connectMicroservice({ strategy: kubemqServer });
await app.startAllMicroservices(); // returns immediately; retries in the background
```

## Custom Serialization [#custom-serialization]

The transport serializes payloads with JSON by default (`JsonSerializer` / `JsonDeserializer`). To change the wire format, supply a `serializer` and `deserializer` that implement the `KubeMQSerializer` / `KubeMQDeserializer` interfaces.

```typescript title="msgpack.serializer.ts"
import type { KubeMQSerializer, KubeMQDeserializer } from '@kubemq/nestjs-transport';
import * as msgpack from 'msgpackr';

export class MsgPackSerializer implements KubeMQSerializer {
  serialize(value: any): Uint8Array {
    return msgpack.encode(value);
  }
}

export class MsgPackDeserializer implements KubeMQDeserializer {
  deserialize(data: Uint8Array, _tags?: Record<string, string>): any {
    return msgpack.decode(data);
  }
}
```

The package also ships built-in `MessagePackSerializer` / `MessagePackDeserializer` and `ProtobufSerializer` / `ProtobufDeserializer` so you do not have to hand-roll the common cases.

The critical rule: configure the **same** serializer pair on every end of the channel. A mismatch — for example a MessagePack producer talking to a JSON consumer — surfaces as a `SerializationError` when the consumer fails to decode the body.

```typescript title="main.ts (server)"
import { MsgPackSerializer, MsgPackDeserializer } from './msgpack.serializer';

const kubemqServer = new KubeMQServer({
  address: 'localhost:50000',
  clientId: 'orders-server',
  serializer: new MsgPackSerializer(),
  deserializer: new MsgPackDeserializer(),
});
```

```typescript title="app.module.ts (client)"
KubeMQModule.register({
  name: 'KUBEMQ_SERVICE',
  address: 'localhost:50000',
  serializer: new MsgPackSerializer(),
  deserializer: new MsgPackDeserializer(),
})
```

<Callout type="warn">
  If only one side is changed, decode failures appear as a `SerializationError` on the receiver. When you switch serialization, roll it out to producers and consumers together.
</Callout>

## Dead-Letter Routing on Handlers [#dead-letter-routing-on-handlers]

Any handler decorator accepts `deadLetterChannel` and `maxRetries` (from `KubeMQHandlerBaseOptions`). When a handler throws, the transport retries up to `maxRetries` times; once retries are exhausted, the message is routed to the dead-letter channel and a `DeadLetterError` is surfaced.

```typescript title="order.handler.ts"
import { QueueHandler, KubeMQQueueContext } from '@kubemq/nestjs-transport';
import { Payload, Ctx } from '@nestjs/microservices';

export class OrderHandler {
  @QueueHandler('orders.process', {
    deadLetterChannel: 'orders.process.dlq',
    maxRetries: 3,
  })
  async handleProcess(@Payload() data: any, @Ctx() ctx: KubeMQQueueContext) {
    await this.process(data); // a thrown error here counts as one failed attempt
  }
}
```

The retry default depends on whether a DLQ is configured:

<TypeTable
  type="{
  deadLetterChannel: { type: 'string', default: 'undefined', description: 'Channel that receives messages after retries are exhausted.' },
  maxRetries: { type: 'number', default: '0 (no DLQ) / 3 (DLQ set)', description: 'Retry attempts before routing to the DLQ. Defaults to 0 when deadLetterChannel is unset, or 3 when it is set and maxRetries is omitted.' },
}"
/>

When a message is routed, the transport sends it to the DLQ as a queue message and stamps diagnostic tags so downstream tooling can triage failures:

```text title="DLQ message tags"
x-dlq-source-channel        # the original channel
x-dlq-source-pattern-type   # command | query | event | event_store | queue
x-dlq-source-id             # original message id
x-dlq-failure-reason        # the handler error message
x-dlq-retry-count           # how many attempts were made
x-dlq-timestamp             # ISO timestamp of routing
```

The resulting `DeadLetterError` carries `sourceChannel`, `dlqChannel`, `retryCount`, and the `originalError`, with a message of the form `Message routed to DLQ "orders.process.dlq" after 3 retries on "orders.process"`.

## Payload Validation [#payload-validation]

Pass a DTO class through a handler's `validate` option to run [class-validator](https://github.com/typestack/class-validator) against every incoming payload before your handler body executes. Invalid payloads raise a `MessageValidationError` and the handler is not invoked.

```typescript title="create-order.dto.ts"
import { IsString, IsNumber, Min } from 'class-validator';

export class CreateOrderDto {
  @IsString()
  name!: string;

  @IsNumber()
  @Min(0)
  total!: number;
}
```

```typescript title="order.handler.ts"
import { CommandHandler, KubeMQCommandContext } from '@kubemq/nestjs-transport';
import { Payload, Ctx } from '@nestjs/microservices';
import { CreateOrderDto } from './create-order.dto';

export class OrderHandler {
  @CommandHandler('orders.create', { validate: CreateOrderDto })
  handleCreate(@Payload() data: CreateOrderDto, @Ctx() ctx: KubeMQCommandContext) {
    return { orderId: 'order-123', status: 'created' };
  }
}
```

The `validate` option is available on `@CommandHandler`, `@QueryHandler`, `@EventHandler`, `@EventStoreHandler`, and `@QueueHandler`. Validation transforms the plain payload into a DTO instance and collects each failing property and its constraints; the raised `MessageValidationError` exposes the `channel` and a `violations` array, with a message such as `Message validation failed on "orders.create": 2 violation(s)`.

Validation requires the `class-validator` and `class-transformer` packages. Install them when you use `validate`:

```bash
npm install class-validator class-transformer
```

You can toggle the feature globally with the server `validation` flag. Set it to `false` to disable class-validator integration across all handlers regardless of per-handler `validate` options.

```typescript title="main.ts"
const kubemqServer = new KubeMQServer({
  address: 'localhost:50000',
  clientId: 'orders-server',
  validation: false, // globally disable DTO validation
});
```

<Callout type="warn">
  If a handler declares `validate` but `class-validator` / `class-transformer` are not installed, the transport throws a `ConfigurationError` at validation time rather than silently skipping validation. Install both packages, or remove the `validate` option.
</Callout>

## Idempotency (QueueHandler) [#idempotency-queuehandler]

Queues guarantee at-least-once delivery, which means a message can be redelivered after a visibility-timeout expiry or a `nack`. Add an `idempotency` config to a `@QueueHandler` to deduplicate redeliveries within a time window — duplicates raise a `DuplicateMessageError` and are not processed twice.

```typescript title="order.handler.ts"
import { QueueHandler, KubeMQQueueContext } from '@kubemq/nestjs-transport';
import { Payload, Ctx } from '@nestjs/microservices';

export class OrderHandler {
  @QueueHandler('orders.process', {
    idempotency: {
      ttlSeconds: 300,
      maxCacheSize: 10_000,
    },
  })
  async handleProcess(@Payload() data: any, @Ctx() ctx: KubeMQQueueContext) {
    await this.chargeOnce(data); // safe against redelivery within the TTL window
  }
}
```

<TypeTable
  type="{
  ttlSeconds: { type: 'number', default: '300', description: 'Deduplication window in seconds. A message seen again within this window is treated as a duplicate.' },
  maxCacheSize: { type: 'number', default: '10000', description: 'Maximum number of message keys retained. The oldest entries are evicted (FIFO) when the cap is reached.' },
}"
/>

The cache is an in-memory, TTL-bounded, FIFO-evicting store of message keys. A `DuplicateMessageError` carries the `channel` and the `idempotencyKey`, with a message such as `Duplicate message on "orders.process" with idempotency key "..."`.

<Callout type="info">
  The idempotency cache is per-process and in-memory: it does not survive a restart and is not shared across replicas. It protects against broker-level redelivery to a single consumer, not against the same logical message being delivered to two different instances.
</Callout>

## Concurrency & Backpressure [#concurrency--backpressure]

`concurrency` (or its alias `maxConcurrent`) caps how many handler executions run at once for a channel. When both are set, `concurrency` takes precedence. Work that arrives while the limit is saturated waits in an internal FIFO buffer bounded by `maxQueueDepth`; once that buffer is full, the transport raises a `BackpressureOverflowError`.

```typescript title="order.handler.ts"
import { EventHandler, KubeMQContext } from '@kubemq/nestjs-transport';
import { Payload, Ctx } from '@nestjs/microservices';

export class OrderHandler {
  @EventHandler('orders.updated', {
    concurrency: 10,    // at most 10 handler executions in flight
    maxQueueDepth: 1000, // up to 1000 waiting before overflow
  })
  async handleUpdated(@Payload() data: any, @Ctx() ctx: KubeMQContext) {
    await this.project(data);
  }
}
```

<TypeTable
  type="{
  concurrency: { type: 'number', default: 'unbounded', description: 'Max concurrent handler executions. Alias of maxConcurrent; takes precedence when both are set.' },
  maxConcurrent: { type: 'number', default: 'unbounded', description: 'Max concurrent handler executions.' },
  maxQueueDepth: { type: 'number', default: '1000', description: 'Max depth of the internal FIFO buffer used when concurrency is limited.' },
}"
/>

On overflow, the behavior depends on the pattern: for queue handlers the message is nacked/requeued so the broker can redeliver it later, while for non-queue patterns (events) the excess work is dropped. The `BackpressureOverflowError` carries the `channel` and `maxQueueDepth`, with a message such as `Backpressure queue exceeded max depth (1000) on "orders.updated"`.

## Client Circuit Breaker [#client-circuit-breaker]

The client supports a `circuitBreaker` that wraps outbound operations. After a run of consecutive failures it trips to **open** and short-circuits further calls with a `CircuitBreakerOpenError` instead of hammering an unhealthy broker. After `resetTimeout`, it moves to **half-open** and admits a small number of probe requests; a success closes the circuit, a failure reopens it.

```typescript title="app.module.ts"
import { Module } from '@nestjs/common';
import { KubeMQModule } from '@kubemq/nestjs-transport';

@Module({
  imports: [
    KubeMQModule.register({
      name: 'KUBEMQ_SERVICE',
      address: 'localhost:50000',
      clientId: 'orders-client',
      circuitBreaker: {
        failureThreshold: 5,
        resetTimeout: 30_000,
        halfOpenRequests: 1,
      },
    }),
  ],
})
export class AppModule {}
```

<TypeTable
  type="{
  failureThreshold: { type: 'number', default: '5', description: 'Consecutive failures that trip the circuit from closed to open.' },
  resetTimeout: { type: 'number', default: '30000', description: 'Milliseconds to wait in the open state before allowing half-open probes.' },
  halfOpenRequests: { type: 'number', default: '1', description: 'Number of probe requests admitted while half-open.' },
}"
/>

The state machine is straightforward:

<Mermaid
  chart="stateDiagram-v2
    [*] --> Closed
    Closed --> Open: failures >= failureThreshold
    Open --> HalfOpen: after resetTimeout
    HalfOpen --> Closed: probe succeeds
    HalfOpen --> Open: probe fails"
/>

While open (or once the half-open probe budget is spent), calls fail immediately with a `CircuitBreakerOpenError` whose message is `Circuit breaker is open — request rejected`. A single success in the half-open state resets the failure counter and closes the circuit.

## Error Handling & Exception Filters [#error-handling--exception-filters]

The transport normalizes failures into a `KubeMQRpcException` — a NestJS `RpcException` subclass carrying a structured `KubeMQRpcError`:

```typescript title="KubeMQRpcError"
interface KubeMQRpcError {
  statusCode: number;
  message: string;
  kubemqCode: string;
  kubemqCategory: string;
  channel?: string;
}
```

Two helpers produce these exceptions: `mapErrorToRpcException(error, channel?, verbose?)` maps a kubemq-js error (and a `CircuitBreakerOpenError`) onto an appropriate `statusCode` and `kubemqCode`, and `mapToRpcException(type, channel, errorMessage?, verbose?)` wraps any handler failure as a generic `HANDLER_ERROR`. For example, a `CircuitBreakerOpenError` maps to `statusCode: 503` with `kubemqCode: 'CIRCUIT_BREAKER_OPEN'`.

Catch `KubeMQRpcException` with a NestJS exception filter to translate broker errors into HTTP responses or structured logs:

```typescript title="kubemq-exception.filter.ts"
import { Catch, ExceptionFilter, ArgumentsHost, Logger } from '@nestjs/common';
import { Response } from 'express';
import { KubeMQRpcException } from '@kubemq/nestjs-transport';
import type { KubeMQRpcError } from '@kubemq/nestjs-transport';

function isKubeMQRpcError(val: unknown): val is KubeMQRpcError {
  return typeof val === 'object' && val !== null && 'statusCode' in val;
}

@Catch(KubeMQRpcException)
export class KubeMQExceptionFilter implements ExceptionFilter {
  private readonly logger = new Logger(KubeMQExceptionFilter.name);

  catch(exception: KubeMQRpcException, host: ArgumentsHost): void {
    const response = host.switchToHttp().getResponse<Response>();
    const raw = exception.getError();
    const error = isKubeMQRpcError(raw) ? raw : null;
    const statusCode = error?.statusCode ?? 500;

    if (error) {
      this.logger.warn(
        `KubeMQ error: ${error.message} (code: ${error.kubemqCode}, ` +
          `category: ${error.kubemqCategory}, channel: ${error.channel})`,
      );
    }

    response.status(statusCode).json({
      statusCode,
      error: 'KubeMQ Error',
      timestamp: new Date().toISOString(),
    });
  }
}
```

By default, exception messages are sanitized to `Transport operation failed` so raw broker internals do not leak to clients. During debugging, enable `verboseErrors` on the server or client options to include the underlying broker message in the exception:

```typescript title="main.ts"
const kubemqServer = new KubeMQServer({
  address: 'localhost:50000',
  clientId: 'orders-server',
  verboseErrors: true, // include raw broker detail in KubeMQRpcException
});
```

<Callout type="warn">
  Leave `verboseErrors: false` (the default) in production — it prevents broker-internal error text from reaching API consumers. Enable it only while diagnosing an issue.
</Callout>

## Related [#related]

* [Getting Started with NestJS](/integrations/nestjs/tutorials/getting-started) for the base server, module, and client setup these options extend.
* [CloudEvents Authentication](/connectors/cloudevents/how-to/authentication) for the connector-style HTTP integration's security model.