Client Interceptors

Interceptors for outgoing gRPC client calls -- observability, resilience, and custom logic.

Mount application-wide client interceptors once in createServer({ outgoingInterceptors }) (or createCatalogClient({ outgoingInterceptors }) outside a server): the chain then runs on every catalog route -- ctx.call, ctx.stream, and server.client() -- see Outgoing interceptors. The per-transport examples below are for transport-specific middleware, or for clients built without the catalog.

OTel Client Interceptor

createOtelClientInterceptor() instruments outgoing RPC calls with OpenTelemetry tracing and metrics:

import { createGrpcTransport } from '@connectrpc/connect-node';
import { createOtelClientInterceptor } from '@connectum/otel';

const transport = createGrpcTransport({
  baseUrl: 'http://user-service:5001',
  interceptors: [
    createOtelClientInterceptor({
      serverAddress: 'user-service',   // Required
      serverPort: 5001,
    }),
  ],
});

The interceptor:

• Injects trace context into outgoing requests via propagation.inject() -- downstream services receive the parent span
• Creates SpanKind.CLIENT spans with OTel semantic conventions
• Records rpc.client.* metrics (duration, request/response size)

Options

Option	Type	Default	Description
serverAddress	string	(required)	Target server address (server.address attribute)
serverPort	number	--	Target server port (server.port attribute)
withoutTracing	boolean	false	Disable span creation (metrics only)
withoutMetrics	boolean	false	Disable metric recording (tracing only)
filter	OtelFilter	--	Skip specific RPCs from instrumentation
attributeFilter	OtelAttributeFilter	--	Exclude specific span attributes
recordMessages	boolean	false	Record message direction, sequence number, and estimated size in span events

Trace Context Propagation

When both server and client interceptors are configured, trace context flows automatically across service boundaries:

// Service A: server OTel interceptor + client OTel interceptor
const server = createServer({
  services: [routes],
  interceptors: [
    createOtelInterceptor({ serverPort: 5000 }),  // Server spans
  ],
});

const userTransport = createGrpcTransport({
  baseUrl: 'http://user-service:5001',
  interceptors: [
    createOtelClientInterceptor({                  // Client spans
      serverAddress: 'user-service',
      serverPort: 5001,
    }),
  ],
});

In a trace viewer (Jaeger, Grafana Tempo), you'll see a single trace spanning both services with linked spans.

Auth Client Interceptors

@connectum/auth ships two client-side factories for outbound service-to-service authentication. Use them on any createGrpcTransport / createConnectTransport instead of hand-rolling header wiring.

import { createGrpcTransport } from '@connectrpc/connect-node';
import {
  createClientBearerInterceptor,
  createClientGatewayInterceptor,
} from '@connectum/auth';

// Forward a user's Bearer token (or a refreshable service token) to the upstream
const upstream = createGrpcTransport({
  baseUrl: 'http://upstream-service:5000',
  interceptors: [
    createClientBearerInterceptor({
      token: async () => (await getAccessToken()).accessToken,
    }),
  ],
});

// Trusted service-to-service call behind a shared-secret gateway
const internal = createGrpcTransport({
  baseUrl: 'http://internal-service:5000',
  interceptors: [
    createClientGatewayInterceptor({
      secret: process.env.GATEWAY_SECRET!,
      subject: 'order-service',
      roles: ['service', 'order-writer'],
    }),
  ],
});

• createClientBearerInterceptor sets Authorization: Bearer <token>; the token may be a static string or an async factory invoked before each request (useful for refresh flows).
• createClientGatewayInterceptor sets x-gateway-secret, x-auth-subject, and optionally x-auth-roles so the receiving service can reconstruct the AuthContext via createGatewayAuthInterceptor without re-authenticating.

Full guidance: Client-Side Auth Interceptors.

Resilience for Clients

Use createDefaultInterceptors() on client transports for circuit breaker, timeout, and retry. Disable server-only interceptors:

import { createDefaultInterceptors } from '@connectum/interceptors';

const transport = createGrpcTransport({
  baseUrl: 'http://inventory-service:5000',
  interceptors: [
    createOtelClientInterceptor({
      serverAddress: 'inventory-service',
      serverPort: 5000,
    }),
    ...createDefaultInterceptors({
      circuitBreaker: { threshold: 5 },
      timeout: { duration: 5_000 },
      retry: { maxRetries: 2 },
      // Disable server-only interceptors
      bulkhead: false,
      errorHandler: false,
      serializer: false,
      validation: false,
    }),
  ],
});

For timeout and retry cancellation behavior, including streaming boundaries and signal-unaware handlers, see Cancellation and Streaming Scope.

Circuit Breaker Behavior

Each circuit breaker instance tracks consecutive failures classified by its failure predicate. Use a separate instance for each downstream transport:

State	Behavior
Closed	Requests pass through normally
Open	Requests fail immediately with Unavailable (no downstream call)
Half-Open	A single probe request is allowed; success closes, failure re-opens

The default threshold is 5 consecutive classified failures. Infrastructure codes (Unknown, DeadlineExceeded, Internal, Unavailable, DataLoss, and ResourceExhausted) count by default; business responses such as NotFound and InvalidArgument do not. Non-ConnectError failures count too. After the circuit opens, the next call after the cooldown can probe it. Streaming calls bypass the breaker by default (skipStreaming: true). Exact options are in CircuitBreakerOptions.

Per-Service Configuration

Create separate transports with different resilience settings for each downstream service:

// Critical service: aggressive retry, short timeout
const paymentTransport = createGrpcTransport({
  baseUrl: 'http://payment-service:5000',
  interceptors: [
    createOtelClientInterceptor({ serverAddress: 'payment-service', serverPort: 5000 }),
    ...createDefaultInterceptors({
      timeout: { duration: 3_000 },
      retry: { maxRetries: 3 },
      circuitBreaker: { threshold: 3 },
      bulkhead: false, errorHandler: false, serializer: false, validation: false,
    }),
  ],
});

// Non-critical service: lenient timeout, fewer retries
const recommendationTransport = createGrpcTransport({
  baseUrl: 'http://recommendation-service:5000',
  interceptors: [
    createOtelClientInterceptor({ serverAddress: 'recommendation-service', serverPort: 5000 }),
    ...createDefaultInterceptors({
      timeout: { duration: 10_000 },
      retry: { maxRetries: 1 },
      circuitBreaker: { threshold: 10 },
      bulkhead: false, errorHandler: false, serializer: false, validation: false,
    }),
  ],
});

Client Metrics

createRpcClientMetrics() provides standalone client metrics following OTel semantic conventions:

import { createRpcClientMetrics, getMeter } from '@connectum/otel';

const meter = getMeter();
const clientMetrics = createRpcClientMetrics(meter);

Metric	Name	Unit	Description
callDuration	rpc.client.call.duration	seconds	Histogram of call durations
requestSize	rpc.client.request.size	bytes	Histogram of request sizes
responseSize	rpc.client.response.size	bytes	Histogram of response sizes

These metrics are recorded automatically when using createOtelClientInterceptor() (unless withoutMetrics: true).

Streaming Instrumentation

Both server and client interceptors automatically instrument streaming RPCs (client streaming, server streaming, and bidirectional).

Span lifecycle for streaming calls:

1. Span starts when the RPC begins
2. Individual rpc.message events are recorded for each sent/received message (when recordMessages is enabled)
3. Span ends when the stream is fully consumed, errors, or is broken

This ensures accurate duration measurement for long-lived streams.

This span lifecycle differs from the client metric lifecycle: streaming rpc.client.call.duration is recorded when the transport returns the response stream, before it is consumed. Streaming request and response size metrics are recorded as 0. rpc.message.uncompressed_size is also an estimate: messages without a toBinary() method produce 0.

Streaming Attributes

Attribute	Value	Description
rpc.message.id	sequential number	Message sequence number within the stream
rpc.message.type	"SENT" / "RECEIVED"	Message direction
rpc.message.uncompressed_size	bytes (estimated)	Estimated message size
network.transport	"tcp"	Network transport protocol

Related

• Service Communication -- overview, transport configuration, service discovery
• Communication Patterns -- request-response, fan-out, streaming
• Distributed Tracing -- server/client interceptors, deep tracing
• Interceptors -- server-side interceptor chain
• Gateway Authentication -- server side of createClientGatewayInterceptor
• Client-Side Auth Interceptors -- bearer and gateway credentials
• @connectum/otel -- Package Guide
• @connectum/otel API -- Full API Reference