Configuration & 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:

docker run -d \  --name kubemq \  -p 50000:50000 \  -e KUBEMQ_TOKEN=YOUR_LICENSE_KEY \  europe-docker.pkg.dev/kubemq/images/kubemq:next

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

• @kubemq/nestjs-transport and kubemq-js installed, with KubeMQServer and/or KubeMQModule already registered in a NestJS app (see Getting Started with NestJS)
• The broker above running and reachable
• class-validator and class-transformer installed if you use the validate handler option (see Payload Validation)

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)

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

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)

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

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

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.

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

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.

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',
  },
});

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.

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

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.

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.

import { MsgPackSerializer, MsgPackDeserializer } from './msgpack.serializer';

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

KubeMQModule.register({
  name: 'KUBEMQ_SERVICE',
  address: 'localhost:50000',
  serializer: new MsgPackSerializer(),
  deserializer: new MsgPackDeserializer(),
})

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.

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:

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:

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

Pass a DTO class through a handler's validate option to run class-validator against every incoming payload before your handler body executes. Invalid payloads raise a MessageValidationError and the handler is not invoked.

import { IsString, IsNumber, Min } from 'class-validator';

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

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

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:

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.

const kubemqServer = new KubeMQServer({
  address: 'localhost:50000',
  clientId: 'orders-server',
  validation: false, // globally disable DTO validation
});

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.

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
  }
}

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 "...".

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.

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);
  }
}

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

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.

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 {}

The state machine is straightforward:

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

The transport normalizes failures into a KubeMQRpcException — a NestJS RpcException subclass carrying a structured 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:

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:

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

Related

• Getting Started with NestJS for the base server, module, and client setup these options extend.
• CloudEvents Authentication for the connector-style HTTP integration's security model.