natsclient

package
v1.0.0-beta.151 Latest Latest
Warning

This package is not in the latest version of its module.

 Go to latest Published: Jul 18, 2026 License: MIT Imports: 22 Imported by: 0

README

natsclient

NATS client with circuit breaker protection, automatic reconnection, and comprehensive JetStream/KV support for distributed edge systems.

Overview

The natsclient package provides a robust NATS client implementation designed for resilient operation in edge computing environments. It wraps the standard NATS Go client with additional reliability features including circuit breaker pattern for failure protection, exponential backoff for reconnection, and proper context propagation throughout all operations.

This package is the foundation for all NATS communication in the SemStreams framework, providing both core pub/sub functionality and advanced features like JetStream streams, consumers, and Key-Value stores. The client handles connection lifecycle management automatically, allowing applications to focus on business logic rather than connection state management.

Installation

import "github.com/c360/semstreams/natsclient"

Core Concepts

Circuit Breaker Pattern

The client implements a circuit breaker that prevents cascading failures by failing fast after a threshold of consecutive failures (default: 5). The circuit opens to prevent further attempts, then gradually tests the connection with exponential backoff.

Connection Lifecycle

Connection states transition through: Disconnected → Connecting → Connected → Reconnecting (on failure) → Connected. The client handles all transitions automatically with configurable callbacks for state changes.

KVStore Abstraction

A high-level abstraction over NATS KV providing automatic CAS (Compare-And-Swap) retry logic, JSON helpers, and consistent error handling for configuration management scenarios.

Distributed Tracing

All publish and request methods automatically propagate W3C-compliant trace context. If no trace exists in the context, one is auto-generated, ensuring every message can be correlated across services.

Headers injected:

  • traceparent - W3C Trace Context format (00-{trace_id}-{span_id}-{flags})
  • X-Trace-ID, X-Span-ID, X-Parent-Span-ID - Simplified headers for compatibility

Usage

Basic Example
// Create and connect to NATS
client, err := natsclient.NewClient("nats://localhost:4222")
if err != nil {
    return err
}

ctx := context.Background()
err = client.Connect(ctx)
if err != nil {
    return err
}
defer client.Close(ctx)

// Publish a message
err = client.Publish(ctx, "subject.name", []byte("message data"))

// Subscribe to messages
sub, err := client.Subscribe("subject.*", func(msg *nats.Msg) {
    // Handle message
    fmt.Printf("Received: %s\n", string(msg.Data))
})
Advanced Usage
// Create client with options
client, err := natsclient.NewClient("nats://localhost:4222",
    natsclient.WithMaxReconnects(-1),  // Infinite reconnects
    natsclient.WithReconnectWait(2*time.Second),
    natsclient.WithCircuitBreakerThreshold(10),
    natsclient.WithDisconnectCallback(func(err error) {
        log.Printf("Disconnected: %v", err)
    }),
    natsclient.WithReconnectCallback(func() {
        log.Println("Reconnected successfully")
    }),
)

// JetStream operations
js, err := client.JetStream()
stream, err := js.CreateStream(ctx, jetstream.StreamConfig{
    Name:     "EVENTS",
    Subjects: []string{"events.>"},
})

// Key-Value store with CAS
bucket, err := client.CreateKeyValueBucket(ctx, jetstream.KeyValueConfig{
    Bucket: "config",
})

kvStore := client.NewKVStore(bucket)
err = kvStore.UpdateJSON(ctx, "service.config", func(config map[string]any) error {
    config["enabled"] = true
    return nil
})
Tracing
// Traces are auto-generated - no action needed for basic tracing
err := client.Publish(ctx, "events.user", data)

// Explicit trace context
tc := natsclient.NewTraceContext()
ctx = natsclient.ContextWithTrace(ctx, tc)
err = client.Request(ctx, "service.action", data, 5*time.Second)

// Extract trace from incoming message
tc = natsclient.ExtractTrace(msg)
if tc != nil {
    // Create child span for downstream calls
    childCtx := natsclient.ContextWithTrace(ctx, tc.NewSpan())
    err = client.Publish(childCtx, "downstream.subject", response)
}

API Reference

Types
Client

Main client type providing NATS connectivity with circuit breaker protection.

type Client struct {
    // Internal fields not exposed
}
KVStore

High-level KV operations with built-in CAS support.

type KVStore struct {
    // Internal fields not exposed
}
KVOptions

Configuration for KV operations behavior.

type KVOptions struct {
    MaxRetries int           // Maximum CAS retry attempts (default: 3)
    RetryDelay time.Duration // Delay between retries (default: 10ms)
    Timeout    time.Duration // Operation timeout (default: 5s)
}
Functions
NewClient(url string, opts ...Option) (*Client, error)

Creates a new NATS client with the specified server URL and options.

(c *Client) Connect(ctx context.Context) error

Establishes connection to NATS server with circuit breaker protection.

(c *Client) Publish(ctx context.Context, subject string, data []byte) error

Publishes a message to the specified subject.

(c *Client) NewKVStore(bucket jetstream.KeyValue, opts ...func(*KVOptions)) *KVStore

Creates a KVStore instance for high-level KV operations.

(kv *KVStore) UpdateJSON(ctx context.Context, key string, updateFn func(map[string]any) error) error

Performs CAS update on JSON data with automatic retry on conflicts.

Interfaces
Logger
type Logger interface {
    Printf(format string, v ...any)
}

Optional logger interface for debug output.

Architecture

Design Decisions

Circuit Breaker over Simple Retry: Chose circuit breaker pattern to prevent cascade failures in distributed systems. After threshold failures, the circuit opens to fail fast rather than continuously retry, giving the system time to recover.

Context-First API: Every I/O operation requires context.Context as first parameter for proper cancellation and timeout support, essential for production systems.

KVStore Abstraction: Created high-level KV abstraction with built-in CAS retry logic to eliminate code duplication across services. Centralizes revision conflict handling and retry logic.

Integration Points
  • Dependencies: NATS server (2.x compatible)
  • Used By: All SemStreams services and components requiring messaging
  • Data Flow: Application → Client → Circuit Breaker → NATS Connection → Server

Configuration

Connection Options
WithMaxReconnects(-1)              // Infinite reconnects (default: 5)
WithReconnectWait(2*time.Second)   // Wait between reconnects (default: 2s)
WithTimeout(10*time.Second)        // Connection timeout (default: 5s)
WithCircuitBreakerThreshold(5)     // Failures before circuit opens (default: 5)
WithLogger(logger)                  // Enable debug logging
Callbacks
WithDisconnectCallback(func(err error) {
    // Handle disconnection event
})

WithReconnectCallback(func() {
    // Handle successful reconnection
})

WithClosedCallback(func() {
    // Handle connection closed
})

Error Handling

Error Types
var (
    ErrCircuitOpen        = errors.New("circuit breaker is open")
    ErrNotConnected       = errors.New("not connected to NATS")
    ErrKVKeyNotFound      = errors.New("kv: key not found")
    ErrKVRevisionMismatch = errors.New("kv: revision mismatch (concurrent update)")
    ErrKVMaxRetriesExceeded = errors.New("kv: max retries exceeded")
)
Error Detection
// Check specific errors
if errors.Is(err, natsclient.ErrCircuitOpen) {
    // Wait for circuit to close
}

// KV conflict detection
if errors.Is(err, natsclient.ErrKVRevisionMismatch) {
    // Handle concurrent update
}

// Helper functions for NATS errors
if natsclient.IsKVNotFoundError(err) {
    // Key doesn't exist
}

Testing

Test Utilities

The package provides comprehensive test utilities:

// Create test client with real NATS via testcontainers
testClient := natsclient.NewTestClient(t, 
    natsclient.WithJetStream(),
    natsclient.WithKV(),
)
defer testClient.Close()

client := testClient.Client
Testing Patterns
  • Uses real NATS server via testcontainers (no mocks)
  • Tests actual behavior including connection lifecycle
  • Thread-safe testing with proper synchronization
  • Comprehensive circuit breaker scenario testing

Performance Considerations

  • Concurrency: Thread-safe for concurrent use from multiple goroutines
  • Memory: Scales with number of active subscriptions and consumers
  • Throughput: Limited by network latency and NATS server performance
  • Circuit Breaker: Adds minimal overhead, fails fast when open

Examples

Resilient Publisher
package main

import (
    "context"
    "log"
    "time"

    "github.com/c360/semstreams/natsclient"
)

func main() {
    client, err := natsclient.NewClient("nats://localhost:4222",
        natsclient.WithMaxReconnects(-1),
        natsclient.WithLogger(log.Default()),
    )
    if err != nil {
        log.Fatal(err)
    }
    
    ctx := context.Background()
    if err := client.Connect(ctx); err != nil {
        log.Fatal(err)
    }
    defer client.Close(ctx)
    
    // Publish with automatic reconnection handling
    for {
        err := client.Publish(ctx, "telemetry.data", []byte("sensor reading"))
        if err != nil {
            if errors.Is(err, natsclient.ErrCircuitOpen) {
                log.Println("Circuit open, waiting...")
                time.Sleep(5 * time.Second)
                continue
            }
            log.Printf("Publish error: %v", err)
        }
        time.Sleep(time.Second)
    }
}
Configuration Management with KV
// Manage service configuration with optimistic locking
bucket, _ := client.CreateKeyValueBucket(ctx, jetstream.KeyValueConfig{
    Bucket:   "config",
    History:  5,
    Replicas: 3,
})

kvStore := client.NewKVStore(bucket)

// Atomic configuration update with retry
err = kvStore.UpdateJSON(ctx, "services.processor", func(config map[string]any) error {
    // This function may be called multiple times on conflict
    config["workers"] = 10
    config["timeout"] = "30s"
    return nil
})
  • service: Uses natsclient for service communication
  • component: Components receive natsclient for messaging
  • config: Manager uses KV store for runtime configuration

License

MIT

Documentation

Overview

Package natsclient provides a client for managing NATS connections with circuit breaker pattern.

Package natsclient provides a robust NATS client with circuit breaker protection, automatic reconnection, and comprehensive JetStream/KV support for distributed edge systems.

The natsclient package wraps the standard NATS Go client with additional reliability features including circuit breaker pattern for failure protection, exponential backoff for reconnection, and proper context propagation throughout all operations. It serves as the foundation for all NATS communication in the StreamKit framework.

Core Features

Circuit Breaker Pattern: Prevents cascading failures by failing fast after a threshold of consecutive failures (default: 5). The circuit opens to prevent further attempts, then gradually tests the connection with exponential backoff.

Connection Lifecycle Management: Handles connection states automatically through the lifecycle: Disconnected → Connecting → Connected → Reconnecting → Connected. The client manages all transitions with configurable callbacks for state changes.

JetStream Support: Full support for JetStream streams, consumers, and Key-Value stores with proper error handling and circuit breaker integration.

KVStore Abstraction: High-level abstraction over NATS KV providing automatic CAS (Compare-And-Swap) retry logic, JSON helpers, and consistent error handling for configuration management scenarios.

Basic Usage

Creating and connecting to NATS: 

client, err := natsclient.NewClient("nats://localhost:4222")
if err != nil {
    return err
}

ctx := context.Background()
err = client.Connect(ctx)
if err != nil {
    return err
}
defer client.Close(ctx)

// Publish a message
err = client.Publish(ctx, "subject.name", []byte("message data"))

// Subscribe to messages (receives full *nats.Msg for access to Subject, Data, Headers)
err = client.Subscribe(ctx, "subject.*", func(msgCtx context.Context, msg *nats.Msg) {
    // Handle message with context (30s timeout per message)
    // For wildcard subscriptions, msg.Subject contains the actual subject
    fmt.Printf("Received on %s: %s\n", msg.Subject, string(msg.Data))
})

Advanced Configuration

Creating client with options: 

client, err := natsclient.NewClient("nats://localhost:4222",
    natsclient.WithMaxReconnects(-1),  // Infinite reconnects
    natsclient.WithReconnectWait(2*time.Second),
    natsclient.WithCircuitBreakerThreshold(10),
    natsclient.WithDisconnectCallback(func(err error) {
        log.Printf("Disconnected: %v", err)
    }),
    natsclient.WithReconnectCallback(func() {
        log.Println("Reconnected successfully")
    }),
)

JetStream Operations

Working with JetStream streams and consumers: 

// Create a stream
stream, err := client.CreateStream(ctx, jetstream.StreamConfig{
    Name:     "EVENTS",
    Subjects: []string{"events.>"},
})

// Publish to stream
err = client.PublishToStream(ctx, "events.user.created", []byte(`{"user_id": "123"}`))

// Consume from stream (receives full jetstream.Msg for access to Subject, Data, Headers)
// Handler is responsible for calling msg.Ack() after processing
err = client.ConsumeStream(ctx, "EVENTS", "events.>", func(msg jetstream.Msg) {
    // Process event - msg.Subject() contains actual subject for wildcard filters
    processEvent(msg.Subject(), msg.Data())
    msg.Ack()
})

Key-Value Store

Using KVStore for configuration management with atomic updates: 

// Create or get KV bucket
bucket, err := client.CreateKeyValueBucket(ctx, jetstream.KeyValueConfig{
    Bucket:   "config",
    History:  5,
    Replicas: 3,
})

// Create KVStore wrapper
kvStore := client.NewKVStore(bucket)

// Atomic JSON update with automatic CAS retry
err = kvStore.UpdateJSON(ctx, "service.config", func(config map[string]any) error {
    // This function may be called multiple times on conflict
    config["enabled"] = true
    config["workers"] = 10
    return nil
})

// Get JSON value
var config map[string]any
err = kvStore.GetJSON(ctx, "service.config", &config)

Circuit Breaker Pattern

The circuit breaker protects against cascading failures: 

// Circuit states:
// - Closed: Normal operation, requests pass through
// - Open: Failures exceeded threshold, failing fast
// - Half-Open: Testing if system recovered

err := client.Connect(ctx)
if errors.Is(err, natsclient.ErrCircuitOpen) {
    // Circuit is open, wait for it to test recovery
    log.Println("Circuit breaker is open, backing off...")
    time.Sleep(client.Backoff())
    // Retry later
}

Circuit breaker configuration: 

client, err := natsclient.NewClient(url,
    natsclient.WithCircuitBreakerThreshold(5),  // Open after 5 failures
    natsclient.WithMaxBackoff(time.Minute),     // Max backoff duration
)

Connection Status and Health

Monitoring connection health: 

// Check current status
status := client.Status()
switch status {
case natsclient.StatusConnected:
    // Healthy and ready
case natsclient.StatusReconnecting:
    // Temporarily disconnected, reconnecting
case natsclient.StatusCircuitOpen:
    // Circuit breaker is open
case natsclient.StatusDisconnected:
    // Not connected
}

// Get detailed status
statusInfo := client.GetStatus()
log.Printf("Status: %v, Failures: %d, RTT: %v",
    statusInfo.Status,
    statusInfo.FailureCount,
    statusInfo.RTT)

// Wait for connection
ctx, cancel := context.WithTimeout(context.Background(), 30*time.Second)
defer cancel()
err := client.WaitForConnection(ctx)

Health monitoring with callbacks: 

client, err := natsclient.NewClient(url,
    natsclient.WithHealthCheck(10*time.Second),
    natsclient.WithHealthChangeCallback(func(healthy bool) {
        if healthy {
            log.Println("Connection restored")
        } else {
            log.Println("Connection lost")
        }
    }),
)

Error Handling

The package defines specific error types for different failure scenarios: 

var (
    ErrCircuitOpen        = errors.New("circuit breaker is open")
    ErrNotConnected       = errors.New("not connected to NATS")
    ErrConnectionTimeout  = errors.New("connection timeout")
)

Error detection patterns: 

err := client.Publish(ctx, "subject", data)
if err != nil {
    // Check for circuit breaker
    if errors.Is(err, natsclient.ErrCircuitOpen) {
        // Back off and retry later
        return
    }

    // Check for connection issues
    if errors.Is(err, natsclient.ErrNotConnected) {
        // Trigger reconnection
        return
    }

    // Other error
    log.Printf("Publish failed: %v", err)
}

KV-specific error handling: 

err := kvStore.UpdateJSON(ctx, key, updateFn)
if err != nil {
    // Check for key not found
    if natsclient.IsKVNotFoundError(err) {
        // Key doesn't exist, create it
    }

    // Check for conflict (CAS failed after retries)
    if natsclient.IsKVConflictError(err) {
        // Too many concurrent updates
    }
}

Connection Options

Available configuration options: 

WithMaxReconnects(n int)              // Maximum reconnection attempts (-1 = infinite)
WithReconnectWait(d time.Duration)    // Wait between reconnection attempts
WithTimeout(d time.Duration)          // Connection timeout
WithDrainTimeout(d time.Duration)     // Timeout for graceful shutdown
WithPingInterval(d time.Duration)     // Health check interval
WithCircuitBreakerThreshold(n int)    // Failures before circuit opens
WithMaxBackoff(d time.Duration)       // Maximum backoff duration
WithLogger(logger Logger)             // Custom logger for debug output
WithHealthCheck(d time.Duration)      // Enable health monitoring
WithClientName(name string)           // Client identification

Authentication and Security

Username/password authentication: 

client, err := natsclient.NewClient(url,
    natsclient.WithCredentials("username", "password"),
)

Token authentication: 

client, err := natsclient.NewClient(url,
    natsclient.WithToken("auth-token"),
)

TLS configuration: 

client, err := natsclient.NewClient(url,
    natsclient.WithTLS(true),
    natsclient.WithTLSCerts("client.crt", "client.key"),
    natsclient.WithTLSCA("ca.crt"),
)

Note: Credentials are cleared from memory when the client is closed.

Testing

The package provides test utilities for integration testing: 

func TestMyService(t *testing.T) {
    // Create test client with real NATS via testcontainers
    testClient := natsclient.NewTestClient(t,
        natsclient.WithJetStream(),
        natsclient.WithKV(),
    )
    defer testClient.Close()

    client := testClient.Client

    // Test with real NATS server
    err := client.Publish(ctx, "test.subject", []byte("test data"))
    assert.NoError(t, err)
}

Testing patterns:

  • Uses real NATS server via testcontainers (no mocks)
  • Tests actual behavior including connection lifecycle
  • Thread-safe testing with proper synchronization
  • Comprehensive circuit breaker scenario testing

Thread Safety

The Client type is thread-safe and can be used concurrently from multiple goroutines:

  • All public methods are safe for concurrent use
  • Connection state is managed with atomic operations and mutexes
  • Subscriptions and consumers can be created from any goroutine
  • Close() can only be called once (subsequent calls are no-ops)

Performance Considerations

Concurrency: Thread-safe for concurrent use from multiple goroutines. No artificial concurrency limits - scales with available system resources.

Memory: Memory usage scales with number of active subscriptions and consumers. Each subscription maintains its own message buffer. Health monitoring adds minimal overhead (one goroutine with configurable interval).

Throughput: Limited primarily by network latency and NATS server performance. Circuit breaker adds negligible overhead in normal operation and fails fast when open.

Connection Lifecycle: Reconnection uses exponential backoff to avoid overwhelming the server during failures. Maximum backoff is configurable (default: 1 minute).

Distributed Tracing

The package provides W3C-compliant trace context propagation for distributed tracing. All publish and request methods automatically generate trace context if none exists, ensuring complete observability across the message flow.

Trace headers are injected into all outbound NATS messages:

  • traceparent: W3C Trace Context header (00-{trace_id}-{span_id}-{flags})
  • X-Trace-ID: Simplified trace ID header
  • X-Span-ID: Current span identifier
  • X-Parent-Span-ID: Parent span for nested operations

Using trace context: 

// Traces are auto-generated if not present
err := client.Publish(ctx, "subject", data)

// Or provide explicit trace context
tc := natsclient.NewTraceContext()
ctx = natsclient.ContextWithTrace(ctx, tc)
err := client.Publish(ctx, "subject", data)

// Extract trace from received message
tc := natsclient.ExtractTrace(msg)
if tc != nil {
    log.Printf("Trace ID: %s, Span ID: %s", tc.TraceID, tc.SpanID)
}

Creating child spans for nested operations: 

parentTC, _ := natsclient.TraceContextFromContext(ctx)
childTC := parentTC.NewSpan()
childCtx := natsclient.ContextWithTrace(ctx, childTC)
err := client.Request(childCtx, "service.action", data, timeout)

The unified RPC error contract (ADR-060)

A request/reply is EITHER a success body (nil Go error) OR a single typed error value. A SubscribeForRequests handler that returns a Go error has it sent as a header-classified reply:

  • Headers: X-Status: error + X-Error-Class: invalid|transient|fatal
  • X-Error-Code: <stable machine code> (when coded)
  • Body: the standard {"message": "...", "detail": {...}} envelope

There is no in-band error channel and no legacy "error: <msg>" body — a reply with no X-Status header is success.

Callers use RequestClassified / RequestWithRetryClassified, which surface the classified error via the err return (transport AND handler failures, uniformly): 

data, err := c.RequestClassified(ctx, "subject", body, 5*time.Second)
if err != nil {
    if errors.Is(err, errs.ErrRevisionMismatch) { /* CAS: re-read, retry */ }
    if errs.IsInvalid(err)   { /* 4xx — bad input */ }
    if errs.IsTransient(err) { /* retry */ }
    var ce *errs.ClassifiedError
    if errors.As(err, &ce)   { /* ce.Code, ce.Detail */ }
}
json.Unmarshal(data, &resp) // success body; err already handled

FOOTGUN: plain Request() / RequestWithHeaders() do NOT inspect the X-Status header — they return the error envelope body with err == nil, which a caller would silently json.Unmarshal as a zero-valued success. New callers MUST use RequestClassified (or run ClassifyReply on the reply when they need to send custom headers). See feedback_silent_handler_error_payload_audit memory.

Handler-side, return a classified error so the headers carry truth: 

return nil, errs.ClassifiedCode(errs.ErrorInvalid,
    graph.ErrorCodeEntityNotFound, fmt.Errorf("not found: %s", req.ID))
// Consumer via RequestClassified: errs.IsInvalid(err) == true,
// errors.As → ce.Code == "entity_not_found"; gateways map that to 404.

Architecture Integration

The natsclient package integrates with StreamKit components:

  • service: Services use natsclient for pub/sub communication
  • config: Manager uses KV store for runtime configuration
  • component: Components receive natsclient for messaging
  • engine: Flow engine coordinates component communication via NATS

Data flow: 

Application → Client → Circuit Breaker → NATS Connection → NATS Server

Design Decisions

Circuit Breaker over Simple Retry: Chose circuit breaker pattern to prevent cascade failures in distributed systems. After threshold failures, the circuit opens to fail fast rather than continuously retry, giving the system time to recover.

Context-First API: Every I/O operation requires context.Context as first parameter for proper cancellation and timeout support, essential for production systems.

KVStore Abstraction: Created high-level KV abstraction with built-in CAS retry logic to eliminate code duplication across services. Centralizes revision conflict handling and retry logic.

Testcontainers over Mocks: Integration tests use real NATS server via testcontainers to catch actual integration issues. Mock-based testing can miss edge cases in the NATS protocol implementation.

Examples

Resilient publisher with automatic reconnection: 

package main

import (
    "context"
    "log"
    "time"

    "github.com/c360studio/semstreams/natsclient"
)

func main() {
    client, err := natsclient.NewClient("nats://localhost:4222",
        natsclient.WithMaxReconnects(-1),
        natsclient.WithLogger(log.Default()),
    )
    if err != nil {
        log.Fatal(err)
    }

    ctx := context.Background()
    if err := client.Connect(ctx); err != nil {
        log.Fatal(err)
    }
    defer client.Close(ctx)

    // Publish with automatic reconnection handling
    for {
        err := client.Publish(ctx, "telemetry.data", []byte("sensor reading"))
        if err != nil {
            if errors.Is(err, natsclient.ErrCircuitOpen) {
                log.Println("Circuit open, waiting...")
                time.Sleep(5 * time.Second)
                continue
            }
            log.Printf("Publish error: %v", err)
        }
        time.Sleep(time.Second)
    }
}

Configuration management with atomic updates: 

// Manage service configuration with optimistic locking
bucket, _ := client.CreateKeyValueBucket(ctx, jetstream.KeyValueConfig{
    Bucket:   "config",
    History:  5,
    Replicas: 3,
})

kvStore := client.NewKVStore(bucket)

// Atomic configuration update with automatic retry
err = kvStore.UpdateJSON(ctx, "services.processor", func(config map[string]any) error {
    // This function may be called multiple times on conflict
    config["workers"] = 10
    config["timeout"] = "30s"
    return nil
})

For more examples and detailed usage, see the README.md in this directory.

Package natsclient — the unified RPC error contract (ADR-060, gh#93).

A request/reply is EITHER a success body (with no error header) OR a single typed error value. There is no in-band error channel. A handler error from a SubscribeForRequests handler — and from direct msg.Respond callers that opt in via RespondError — travels as wire headers plus a standard JSON error body: 

X-Status:      error
X-Error-Class: transient | invalid | fatal
X-Error-Code:  entity_not_found | revision_mismatch | ...   (when coded)
body:          {"message": "<text>", "detail": {...}}

ClassifyReply reconstructs a *errs.ClassifiedError from these: the class drives errs.IsInvalid / IsTransient / IsFatal, the code is reached via errors.As(err, &ce) → ce.Code, the one control-flow sentinel round-trips via errors.Is(err, errs.ErrRevisionMismatch), and the structured detail via ce.Detail. (The earlier additive window — a legacy "error: <msg>" body kept alongside the headers — was retired with the PR-D breaking change.)

Caller pattern: 

data, err := c.RequestClassified(ctx, subj, body, timeout)
if err != nil {
    // transport OR classified handler error, uniformly:
    if errors.Is(err, errs.ErrRevisionMismatch) { /* re-read, retry */ }
    if errs.IsInvalid(err) { /* 4xx */ }
    if errs.IsTransient(err) { /* retry */ }
    var ce *errs.ClassifiedError
    if errors.As(err, &ce) { /* ce.Code, ce.Detail */ }
    return err
}
json.Unmarshal(data, &resp) // data is a success body; err already handled

Handler pattern: return (nil, err) — SubscribeForRequests calls RespondError for you. HTTP semantics (404 vs 400) belong at the gateway, which reads ce.Code (e.g. entity_not_found → 404) rather than substring-sniffing.

Footgun — plain Request() / RequestWithHeaders() still don't classify

The plain Request() / RequestWithHeaders() methods return the raw reply body and do NOT inspect the X-Status header. A handler error reply carries the {message, detail} body with err == nil, so 

data, err := c.Request(...)
if err != nil { return err }
json.Unmarshal(data, &resp)   // mis-decodes an error body as success

silently mis-decodes failures. New callers MUST use RequestClassified / RequestWithRetryClassified (or run ClassifyReply on the reply themselves when they need to send custom headers). See feedback_silent_handler_error_payload_audit.md.

Sentinel chains do not survive the wire boundary

classifiedFromHeader reconstructs a fresh *errs.ClassifiedError from the headers + body. An arbitrary inner sentinel chain (jetstream.ErrKeyNotFound, etc.) is LOST in transit; what round-trips is the {class, code, message, detail} contract — including errs.ErrRevisionMismatch by code. Surface any other distinction a consumer must branch on as a code (ce.Code).

Package natsclient provides request/reply pattern support for NATS.

Package natsclient provides JetStream stream management utilities.

Package natsclient provides testcontainers-based NATS infrastructure for testing.

Package natsclient provides typed subject patterns for compile-time type safety.

Index

Constants

View Source 
const (
	// HeaderStatus is set to HeaderStatusError on reply messages that
	// represent a handler-side failure. Absent on success replies.
	HeaderStatus = "X-Status"

	// HeaderErrorClass carries the pkg/errs.ErrorClass value as a
	// lowercase string: "transient", "invalid", or "fatal". Set
	// only when HeaderStatus == HeaderStatusError.
	HeaderErrorClass = "X-Error-Class"

	// HeaderErrorCode carries the ADR-060 stable machine Code for the
	// failure (the graph.ErrorCode* values: "entity_not_found",
	// "revision_mismatch", ...). Additive over the gh#93 header set:
	// legacy callers ignore it; ClassifyReply reads it into
	// (*errs.ClassifiedError).Code so errors.Is(err, ErrRevisionMismatch)
	// and ce.Code discrimination work across the wire. Set ONLY when the
	// handler error carries a non-empty Code, so existing uncoded handler
	// errors are byte-for-byte unchanged on the wire (the reply body is
	// also unchanged — Code rides the header; the standard error body for
	// Detail lands with the breaking PR).
	HeaderErrorCode = "X-Error-Code"
)

Header keys for the header-classified error convention. A failure reply carries X-Status: error plus the class/code headers; ClassifyReply reconstructs the typed error from them.

View Source 
const (
	HeaderStatusError = "error"

	ErrorClassTransient = "transient"
	ErrorClassInvalid   = "invalid"
	ErrorClassFatal     = "fatal"
)

Values used in HeaderStatus / HeaderErrorClass.

View Source 
const (
	// MaxKVLiteralTokenBytes is the SemStreams byte budget for one literal KV token.
	MaxKVLiteralTokenBytes = 512
	// MaxKVLiteralKeyBytes is the SemStreams byte budget for one complete literal KV key.
	MaxKVLiteralKeyBytes = 1024
	// MaxKVLiteralKeyTokens is the SemStreams arity budget for one complete literal KV key.
	MaxKVLiteralKeyTokens = 64
	// MaxKVWildcardFilterBytes is the SemStreams byte budget for one complete KV filter.
	MaxKVWildcardFilterBytes = 1024
	// MaxKVWildcardFilterTokens is the SemStreams arity budget for one complete KV filter.
	MaxKVWildcardFilterTokens = 64
	// MaxKVOpaqueTokenInputBytes is the largest byte string accepted by the v1 opaque codec.
	MaxKVOpaqueTokenInputBytes = 254
	// MaxKVOpaqueTokenBytes is the largest token produced or accepted by the v1 opaque codec.
	MaxKVOpaqueTokenBytes = 511
)
View Source 
const (
	// ErrorCodeKVTokenInvalid classifies invalid literal KV tokens.
	ErrorCodeKVTokenInvalid = "kv_token_invalid"
	// ErrorCodeKVKeyInvalid classifies invalid literal KV keys.
	ErrorCodeKVKeyInvalid = "kv_key_invalid"
	// ErrorCodeKVFilterInvalid classifies invalid KV wildcard filters.
	ErrorCodeKVFilterInvalid = "kv_filter_invalid"
	// ErrorCodeKVTokenEncodeInvalid classifies unsupported opaque-token input.
	ErrorCodeKVTokenEncodeInvalid = "kv_token_encode_invalid"
	// ErrorCodeKVTokenDecodeInvalid classifies invalid opaque-token encodings.
	ErrorCodeKVTokenDecodeInvalid = "kv_token_decode_invalid"
)
View Source 
const (
	// KVReasonEmpty identifies an empty whole input.
	KVReasonEmpty = "empty"
	// KVReasonBytes identifies a whole-input byte-budget violation.
	KVReasonBytes = "bytes"
	// KVReasonTokens identifies a whole-input token-count violation.
	KVReasonTokens = "tokens"
	// KVReasonEmptyToken identifies an empty position in a key or filter.
	KVReasonEmptyToken = "empty_token"
	// KVReasonTokenBytes identifies a per-token byte-budget violation.
	KVReasonTokenBytes = "token_bytes"
	// KVReasonSeparator identifies a separator in a literal token.
	KVReasonSeparator = "separator"
	// KVReasonWildcard identifies a wildcard in a literal position.
	KVReasonWildcard = "wildcard"
	// KVReasonPosition identifies a wildcard in a forbidden position.
	KVReasonPosition = "position"
	// KVReasonAlphabet identifies a byte outside the literal alphabet.
	KVReasonAlphabet = "alphabet"
	// KVReasonVersion identifies an unknown opaque codec version.
	KVReasonVersion = "version"
	// KVReasonHex identifies malformed opaque hexadecimal.
	KVReasonHex = "hex"
	// KVReasonNoncanonical identifies valid but non-canonical opaque hexadecimal.
	KVReasonNoncanonical = "noncanonical"
)
View Source 
const (
	// KVDetailReason is the mandatory stable reason detail key.
	KVDetailReason = "reason"
	// KVDetailMeasuredBytes reports the rejected byte count.
	KVDetailMeasuredBytes = "measured_bytes"
	// KVDetailAllowedBytes reports the applicable byte budget.
	KVDetailAllowedBytes = "allowed_bytes"
	// KVDetailMeasuredTokens reports the rejected token count.
	KVDetailMeasuredTokens = "measured_tokens"
	// KVDetailAllowedTokens reports the applicable token-count budget.
	KVDetailAllowedTokens = "allowed_tokens"
	// KVDetailTokenIndex reports the zero-based failing position.
	KVDetailTokenIndex = "token_index"
)
View Source 
const (
	// DefaultReadinessProbeTimeout bounds each attempt. Short so a
	// not-yet-subscribed responder fails fast and retries instead of consuming a
	// full query timeout per attempt.
	DefaultReadinessProbeTimeout = 2 * time.Second
	// DefaultReadinessBudget bounds the TOTAL wall-clock spent waiting for the
	// responder to come up before the error surfaces.
	DefaultReadinessBudget = 30 * time.Second
)

Readiness-gated read defaults (ADR-060 sibling doctrine, third bucket — see docs/operations/07-nats-request-retry.md). A readiness-gated read tolerates a not-yet-subscribed responder at cold start / after reconnect: it retries with a SHORT per-attempt timeout up to a bounded TOTAL budget, returning the first reply. Because the per-attempt timeout is short and the total is bounded, a genuinely hung responder fails within the budget rather than a full-timeout×N storm — so it does NOT mask a hung responder the way retrying a full-timeout query would. Distinct from Request (steady-state query: timeout = real signal, no retry) and RequestWithRetry (mutation: retry-any at full timeout).

View Source 
const (
	// TraceparentHeader is the W3C standard trace context header
	// Format: 00-{trace_id}-{span_id}-{flags}
	// trace_id: 32 hex chars (16 bytes), span_id: 16 hex chars (8 bytes)
	TraceparentHeader = "traceparent"

	// TraceIDHeader is a simplified header for internal use
	TraceIDHeader = "X-Trace-ID"
	// SpanIDHeader is a simplified header for internal use
	SpanIDHeader = "X-Span-ID"
	// ParentSpanHeader is a simplified header for internal use
	ParentSpanHeader = "X-Parent-Span-ID"
)

W3C Trace Context headers

View Source 
const DefaultRequestTimeout = 5 * time.Second

DefaultRequestTimeout is the default timeout for request/reply operations.

Variables

View Source 
var (
	ErrNotConnected      = stderrors.New("not connected to NATS")
	ErrCircuitOpen       = stderrors.New("circuit breaker is open")
	ErrConnectionTimeout = stderrors.New("connection timeout")
)

Error messages

View Source 
var (
	ErrKVKeyNotFound        = errors.New("kv: key not found")
	ErrKVKeyExists          = errors.New("kv: key already exists")
	ErrKVRevisionMismatch   = errors.New("kv: revision mismatch (concurrent update)")
	ErrKVMaxRetriesExceeded = errors.New("kv: max retries exceeded")
)

Well-known errors matching Graph processor patterns

View Source 
var ErrGraphBucketRetention = errors.New("live graph bucket has lifecycle retention (ADR-068 D1: forbidden)")

ErrGraphBucketRetention is returned by CheckNoLifecycleRetention when a live graph bucket carries a lifecycle-eviction config. Sentinel so a boot path can classify it (fail-closed) distinctly from a transient status error.

Functions

func BucketLastSeq 

func BucketLastSeq(ctx context.Context, bucket jetstream.KeyValue) (uint64, error)

BucketLastSeq returns the backing stream's LastSeq for a KV bucket — the highest sequence ever assigned to the bucket, read fresh from the server on every call.

It is the query-time "target" for revision-lag readiness (ADR-066): every committed write to the bucket has a Revision() <= LastSeq, and LastSeq is monotonic even under History=1 — a purge raises FirstSeq but never lowers LastSeq. LastSeq shares the same sequence space as KeyValueEntry.Revision() (both are stream sequence numbers), so a caller can compare an indexed-revision watermark directly against it. Prefer this over the last watch entry's Delta()==0: that cache is stale exactly in the committed-but-not-yet-delivered window this target must see through.

func BucketRetention 

func BucketRetention(ctx context.Context, bucket jetstream.KeyValue) (maxAge time.Duration, maxBytes int64, err error)

BucketRetention returns a bucket's lifecycle-eviction config from its backing stream: maxAge (the KV bucket's TTL — age eviction) and maxBytes (size eviction; 0/negative = unlimited). Callers on the live graph assert both are non-binding — NATS age/size eviction is reachability-blind and would drop entities with live inbound edges (ADR-068 D1). Mirrors BucketLastSeq: the KeyValueStatus interface does not surface these, so we read the concrete JetStream-backed status's stream config.

func CheckNoLifecycleRetention 

func CheckNoLifecycleRetention(name string, maxAge time.Duration, maxBytes int64) error

CheckNoLifecycleRetention is the pure D1 guardrail: it errors if a bucket's retention config is binding. maxAge > 0 (a TTL) is always a violation. maxBytes > 0 is a violation too — a size cap on the live graph evicts by size, which is reachability-blind; a deployment that genuinely wants a non-binding crash backstop should size it far above steady state and treat hitting it as an alert, not configure it here. Pure + no I/O so it is unit-testable; the boot path pairs it with BucketRetention.

func ClassifyReply 

func ClassifyReply(msg *nats.Msg) ([]byte, error)

ClassifyReply inspects a reply message and returns either the success body (when no error signal is present) or a classified error suitable for branching with errs.IsInvalid / IsTransient / IsFatal.

ADR-060: a failure is signalled ONLY by the X-Status: error header. The body is the standard {message, detail} envelope; the message + detail + the X-Error-Class / X-Error-Code headers reconstruct a *errs.ClassifiedError so errors.As(err, &ce) reaches ce.Code/ce.Detail and errors.Is(err, errs.ErrRevisionMismatch) round-trips. The legacy "error: " body fallback is gone (every producer header-stamps via RespondError/ReplyError).

func ConsumeWithHeartbeat 

func ConsumeWithHeartbeat(
	ctx context.Context,
	msg jetstream.Msg,
	heartbeatInterval time.Duration,
	work func(context.Context) error,
) error

ConsumeWithHeartbeat runs work in a goroutine while periodically calling msg.InProgress() to reset the AckWait clock. This allows short AckWait values for failure detection while supporting arbitrarily long processing.

Ack/Nak ownership: this function calls Ack, NakWithDelay, or Nak on the message. The caller must NOT call these methods when using this helper.

On work success: msg.Ack() On permanent work error: msg.Term() so structurally invalid data is not retried On other work error: msg.NakWithDelay(30s) to allow breathing room before retry On context cancellation: msg.NakWithDelay(5s) for graceful shutdown On InProgress failure: returns error (message will be redelivered by server)

func ContextWithTrace 

func ContextWithTrace(ctx context.Context, tc *TraceContext) context.Context

ContextWithTrace returns a context with trace information

func DecodeKVOpaqueToken 

func DecodeKVOpaqueToken(token string) ([]byte, error)

DecodeKVOpaqueToken decodes one canonical v1 opaque token. It rejects unknown versions, malformed hexadecimal, and uppercase non-canonical forms.

func DetachContextWithTrace 

func DetachContextWithTrace(parent context.Context, timeout time.Duration) (context.Context, context.CancelFunc)

DetachContextWithTrace creates a new context that preserves trace context but resets deadline/cancellation. This is useful for publishing error responses when the original context has expired but trace continuity is still needed.

func EncodeKVOpaqueToken 

func EncodeKVOpaqueToken(raw []byte) (string, error)

EncodeKVOpaqueToken encodes arbitrary bytes as one canonical v1 literal KV token. Callers must explicitly choose opaque storage for their key axis.

func FilteredKeys 

func FilteredKeys(ctx context.Context, kv jetstream.KeyValue, pattern string) ([]string, error)

FilteredKeys returns keys matching a NATS wildcard pattern from a raw jetstream.KeyValue bucket. Use this for components that hold jetstream.KeyValue directly instead of *KVStore. The pattern should be a valid NATS subject filter (e.g., "0.>" for level-0 community keys). Returns nil, nil when no keys match.

func InjectTrace 

func InjectTrace(ctx context.Context, msg *nats.Msg)

InjectTrace adds trace headers to a NATS message from context

func IsKVConflictError 

func IsKVConflictError(err error) bool

IsKVConflictError checks if error indicates a conflict (key exists or wrong revision)

func IsKVNotFoundError 

func IsKVNotFoundError(err error) bool

IsKVNotFoundError checks if error indicates key absence — either a never-created key (jetstream.ErrKeyNotFound) or a tombstoned key (jetstream.ErrKeyDeleted). Both surface to callers as "the key is not there"; UpdateWithRetry relies on this to set revision=0 and route the updateFn down the create path.

The NATS Go SDK's public Get already maps ErrKeyDeleted → ErrKeyNotFound, so in the common KVStore.Get path only ErrKeyNotFound reaches us. The ErrKeyDeleted branch defends paths that bypass that mapping (Watch handler entry-error chains, GetRevision wrappers, future SDK changes). See issue #122 and feedback_jetstream_sentinel_set_coverage.

func IsNoResponders 

func IsNoResponders(err error) bool

IsNoResponders reports whether err is (or wraps) the NATS "no responders" transport error — the responder is not subscribed. This is TRANSIENT at startup / after a reconnect (retry via a readiness-gated read), and distinct from a timeout of an EXISTING responder (which signals it is hung — surface, don't retry). Note (per request_integration_test.go): whether an absent responder surfaces as ErrNoResponders vs a plain timeout is server-config dependent, so a false here does NOT prove a responder exists — it only confirms the fast-fail no-responders signal when the server sends it.

func RespondError 

func RespondError(msg *nats.Msg, err error) error

RespondError writes a header-classified error reply to msg: the X-Status / X-Error-Class / X-Error-Code headers carry the class + code, and the body is the {message, detail} envelope (ADR-060).

Used by SubscribeForRequests internally + by direct-msg.Respond handlers that opt in to the convention.

When to reach for RespondError vs (*Client).ReplyError:

  • Handler has *nats.Msg in scope (most common — direct Subscribe callback): use RespondError(msg, err). Free function; reads the reply subject off msg.
  • Handler has only a reply subject + *Client (e.g. deferred-reply forwarder): use c.ReplyError(ctx, replyTo, err). Method; publishes via the client connection.

Returns nil + no-op when err is nil (treat as success — caller should have used msg.Respond with success data). Returns errMissingReplySubject when the inbound message had no reply subject; caller can ignore (the request was fire-and-forget).

func TerminateDelivery 

func TerminateDelivery(err error) error

TerminateDelivery marks err for JetStream Term handling. Transient and cancellation errors must be returned unchanged so their existing NAK paths remain intact.

func ValidateKVLiteralKey 

func ValidateKVLiteralKey(key string) error

ValidateKVLiteralKey validates a complete dot-separated literal KV key. It rejects wildcards and empty positions and never changes key identity.

func ValidateKVLiteralToken 

func ValidateKVLiteralToken(token string) error

ValidateKVLiteralToken validates one literal NATS KV key position. It never rewrites, normalizes, or encodes the supplied token.

func ValidateKVWildcardFilter 

func ValidateKVWildcardFilter(filter string) error

ValidateKVWildcardFilter validates a complete KV search filter. Wildcards are accepted only as a complete '*' token or a complete final '>' token.

Types

type Client 

type Client struct {
	// contains filtered or unexported fields
}

Client manages NATS connections with circuit breaker pattern

func NewClient 

func NewClient(urls string, opts ...ClientOption) (*Client, error)

NewClient creates a new NATS client with optional configuration. The urls parameter accepts comma-separated NATS server URLs for clustering support (e.g., "nats://server1:4222,nats://server2:4222").

func (*Client) Backoff 

func (m *Client) Backoff() time.Duration

Backoff returns the current backoff duration

func (*Client) Close 

func (m *Client) Close(ctx context.Context) error

Close closes the NATS connection

func (*Client) Connect 

func (m *Client) Connect(ctx context.Context) error

Connect establishes connection to NATS server

func (*Client) ConnectionOptions 

func (m *Client) ConnectionOptions() []nats.Option

ConnectionOptions returns the NATS connection options

func (*Client) ConsumeDurable 

func (c *Client) ConsumeDurable(
	ctx context.Context,
	cfg StreamConsumerConfig,
	heartbeat time.Duration,
	handler func(context.Context, []byte) error,
) error

ConsumeDurable runs a durable at-least-once consumer over a stream with a plain handler: func(ctx, []byte) error — acked on nil, nak-with-delay on error, and held past AckWait by an InProgress heartbeat while the handler runs (see ConsumeWithHeartbeat). The consumer NEVER handles a raw jetstream.Msg for ack semantics; the []byte is the raw message data and envelope decoding (payload-registry BaseMessage, etc.) stays ABOVE natsclient — natsclient does not depend on the message package.

The framework owns the at-least-once ack/heartbeat/redelivery pattern here once, so a durable consumer (e.g. gated-DAG dispatch — ADR-070) does not re-hand-roll ConsumeStreamWithConfig + ConsumeWithHeartbeat + ack per call site.

heartbeat MUST be safely below the effective AckWait: a heartbeat that first fires only after AckWait has already expired lets the server redeliver a still-running unit (duplicate work). ConsumeDurable ENFORCES this (heartbeat*2 <= effective AckWait) and returns an error on a violating config, rather than merely documenting it (ADR-070 B3).

func (*Client) ConsumeStream 

func (m *Client) ConsumeStream(ctx context.Context, streamName, subject string, handler func(jetstream.Msg)) error

ConsumeStream creates a consumer for a stream. Handler receives the full jetstream.Msg to access Subject, Data, Headers, etc. This is essential for wildcard subscriptions where the actual subject differs from the pattern. The handler is responsible for calling msg.Ack() after processing.

func (*Client) ConsumeStreamWithConfig 

func (c *Client) ConsumeStreamWithConfig(
	ctx context.Context,
	cfg StreamConsumerConfig,
	handler func(ctx context.Context, msg jetstream.Msg),
) error

ConsumeStreamWithConfig creates a JetStream consumer with full configuration. The handler receives the raw jetstream.Msg which includes Ack(), Nak(), and Term() methods. Handler MUST call one of these methods to acknowledge the message.

func (*Client) ConsumeStreamWithConfigContexts 

func (c *Client) ConsumeStreamWithConfigContexts(
	setupCtx context.Context,
	handlerCtx context.Context,
	cfg StreamConsumerConfig,
	handler func(ctx context.Context, msg jetstream.Msg),
) error

ConsumeStreamWithConfigContexts separates bounded setup I/O from callback lifetime. setupCtx governs stream lookup and consumer creation; handlerCtx is only the parent for delivered-message contexts after setup succeeds.

func (*Client) CreateKeyValueBucket 

func (m *Client) CreateKeyValueBucket(ctx context.Context, cfg jetstream.KeyValueConfig) (jetstream.KeyValue, error)

CreateKeyValueBucket creates or gets a KV bucket with configuration

func (*Client) CreateStream 

func (m *Client) CreateStream(ctx context.Context, cfg jetstream.StreamConfig) (jetstream.Stream, error)

CreateStream creates a JetStream stream

func (*Client) DeleteKeyValueBucket 

func (m *Client) DeleteKeyValueBucket(ctx context.Context, name string) error

DeleteKeyValueBucket deletes a KV bucket

func (*Client) EnsureStream 

func (c *Client) EnsureStream(ctx context.Context, cfg jetstream.StreamConfig) (jetstream.Stream, error)

EnsureStream creates a stream if it doesn't exist, or returns the existing one.

func (*Client) Failures 

func (m *Client) Failures() int32

Failures returns the current failure count

func (*Client) GetConnection 

func (m *Client) GetConnection() *nats.Conn

GetConnection returns the current NATS connection

func (*Client) GetKeyValueBucket 

func (m *Client) GetKeyValueBucket(ctx context.Context, name string) (jetstream.KeyValue, error)

GetKeyValueBucket gets an existing KV bucket

func (*Client) GetStatus 

func (m *Client) GetStatus() *Status

GetStatus returns current status information

func (*Client) GetStream 

func (m *Client) GetStream(ctx context.Context, name string) (jetstream.Stream, error)

GetStream gets an existing JetStream stream

func (*Client) IsHealthy 

func (m *Client) IsHealthy() bool

IsHealthy returns true if the connection is healthy

func (*Client) JetStream 

func (m *Client) JetStream() (jetstream.JetStream, error)

JetStream returns the JetStream context

func (*Client) ListKeyValueBuckets 

func (m *Client) ListKeyValueBuckets(ctx context.Context) ([]string, error)

ListKeyValueBuckets lists all KV buckets

func (*Client) MaxReconnects 

func (m *Client) MaxReconnects() int

MaxReconnects returns the maximum number of reconnection attempts

func (*Client) NewKVStore 

func (c *Client) NewKVStore(bucket jetstream.KeyValue, opts ...func(*KVOptions)) *KVStore

NewKVStore creates a new KV store with the given bucket

func (*Client) OnHealthChange 

func (m *Client) OnHealthChange(fn func(bool))

OnHealthChange sets a callback for health status changes

func (*Client) PingInterval 

func (m *Client) PingInterval() time.Duration

PingInterval returns the interval for health checks

func (*Client) Publish 

func (m *Client) Publish(ctx context.Context, subject string, data []byte) error

Publish publishes a message to a NATS subject

func (*Client) PublishAsyncComplete 

func (m *Client) PublishAsyncComplete() <-chan struct{}

PublishAsyncComplete returns a channel that closes when every outstanding async publish has been acknowledged by the server. A producer waits on it to drain before shutdown. When JetStream is unavailable it returns an already-closed channel so a drain loop does not block forever.

func (*Client) PublishAsyncPending 

func (m *Client) PublishAsyncPending() int

PublishAsyncPending returns the number of async publishes enqueued but not yet acknowledged. Returns 0 when JetStream is unavailable.

func (*Client) PublishBatchToStream 

func (m *Client) PublishBatchToStream(ctx context.Context, subject string, msgs [][]byte) error

PublishBatchToStream publishes every message in msgs to one subject via the async path, waits for all acks (bounded by ctx), and returns a single aggregate error. Per-subject ordering from this single calling goroutine is preserved (jetstream-go async is in-order per connection, absent a NoResponders retry — see design.md §1). It is the convenience path for bursty producers that do not need per-message futures (gh#470).

The drain waits on THIS batch's own futures, not the connection-global PublishAsyncComplete, so a concurrent async producer on the same Client cannot make this batch over-wait. If ctx is cancelled before all acks arrive, it returns the context error rather than hanging; the already-enqueued publishes still resolve in the background (and feed the circuit breaker via the async error handler on a connection fault). An enqueue error stops further enqueuing but already-enqueued messages are still drained. Ack failures are recorded against the breaker once, by asyncPublishErrHandler — this loop only collects them for the returned error (recording here too would double-count).

func (*Client) PublishToStream 

func (m *Client) PublishToStream(ctx context.Context, subject string, data []byte) error

PublishToStream publishes to a JetStream stream with automatic trace context propagation. If no trace context exists in ctx, one is auto-generated for distributed tracing.

func (*Client) PublishToStreamAsync 

func (m *Client) PublishToStreamAsync(ctx context.Context, subject string, data []byte) (jetstream.PubAckFuture, error)

PublishToStreamAsync publishes to a JetStream stream WITHOUT blocking on the PubAck, returning a jetstream.PubAckFuture the caller inspects (Ok()/Err()) for the eventual server acknowledgement. A single producer goroutine can pipeline many of these past the synchronous ack-RTT ceiling (gh#470).

The enqueue itself is synchronous: an open circuit returns ErrCircuitOpen, a disconnected client returns ErrNotConnected, and a full in-flight window past the stall wait returns jetstream's ErrTooManyStalledMsgs — in all of which the returned future is nil. Trace context injection is preserved. Failed acks are delivered on the future's Err() channel AND recorded against the circuit breaker via the connection-level async error handler.

Ordering: jetstream-go preserves per-subject order per connection, so a single caller publishing to one subject gets in-order storage. Cross-goroutine ordering is the caller's responsibility (as with the synchronous path).

func (*Client) PublishToStreamAsyncWithMsgID 

func (m *Client) PublishToStreamAsyncWithMsgID(ctx context.Context, subject string, data []byte, msgID string) (jetstream.PubAckFuture, error)

PublishToStreamAsyncWithMsgID is PublishToStreamAsync stamping the Nats-Msg-Id header for server-side duplicate detection. It carries the same ADR-055 T1 idempotency contract as the synchronous PublishToStreamWithMsgID: pass a DETERMINISTIC msgID per logical event so a retry/redelivery carries the same ID; dedup holds only within the stream's configured Duplicates window. An empty msgID is equivalent to PublishToStreamAsync (no dedup).

func (*Client) PublishToStreamWithAck 

func (c *Client) PublishToStreamWithAck(
	ctx context.Context,
	subject string,
	data []byte,
) (*jetstream.PubAck, error)

PublishToStreamWithAck publishes a message to a JetStream subject with acknowledgment. If AutoCreate is true and the stream doesn't exist, it will be created. Trace context is auto-generated if not present, and propagated via NATS message headers.

func (*Client) PublishToStreamWithMsgID 

func (m *Client) PublishToStreamWithMsgID(ctx context.Context, subject string, data []byte, msgID string) error

PublishToStreamWithMsgID publishes to a JetStream stream stamping the Nats-Msg-Id header so the server's duplicate-detection window collapses re-publishes/redeliveries of the same logical event to a single store.

This is the producer half of the at-least-once idempotency contract (ADR-055 §5, "T1"): graph-ingest's stream consumer is at-least-once and MergeEntity APPENDS triples on merge, so a redelivered born-once entity payload would double-apply its triples without dedup. Callers pass a DETERMINISTIC msgID for the logical event (e.g. "<loopID>:spawn", "<entityID>:v<n>") so a retry/redelivery carries the same ID.

Scope of the guarantee: dedup only holds WITHIN the stream's configured duplicate window (config.StreamConfig.Duplicates; the NATS server default is 2m when unset). Redelivery outside that window — e.g. DeliverPolicy:all replay on consumer recreation — can still re-append; see ADR-055 Open Question #1. An empty msgID is equivalent to PublishToStream (no dedup), so this is a safe drop-in.

func (*Client) RTT 

func (m *Client) RTT() (time.Duration, error)

RTT returns the round-trip time to the NATS server

func (*Client) ReconnectWait 

func (m *Client) ReconnectWait() time.Duration

ReconnectWait returns the wait duration between reconnection attempts

func (*Client) Reply 

func (c *Client) Reply(ctx context.Context, replyTo string, data []byte) error

Reply sends a reply to a request message. This is typically used by service handlers to respond to requests.

func (*Client) ReplyError 

func (c *Client) ReplyError(ctx context.Context, replyTo string, err error) error

ReplyError sends a header-classified error reply via the client's Publish path. Companion to Reply / ReplyWithHeaders for handlers that don't have the inbound *nats.Msg in scope.

Returns nil + no-op when err is nil OR replyTo is empty.

func (*Client) ReplyWithHeaders 

func (c *Client) ReplyWithHeaders(ctx context.Context, replyTo string, data []byte, headers map[string]string) error

ReplyWithHeaders sends a reply with custom headers.

func (*Client) Request 

func (c *Client) Request(ctx context.Context, subject string, data []byte, timeout time.Duration) ([]byte, error)

Request performs a synchronous request/reply operation. It publishes a message to the subject and waits for a response. The timeout parameter controls how long to wait for a response. If timeout is 0, DefaultRequestTimeout is used.

Request is the right call for QUERIES (read paths) — failures surface to the caller, who decides whether to retry, fall back, or raise an error. For MUTATIONS use RequestWithRetry instead; without retry, transient "no responders" errors during startup races / responder restarts cause silent data loss. See docs/operations/07-nats-request-retry.md for the full rule.

func (*Client) RequestClassified 

func (c *Client) RequestClassified(ctx context.Context, subject string, data []byte, timeout time.Duration) ([]byte, error)

RequestClassified is the recommended caller-side replacement for Request + body-prefix sniffing. It performs the request, then runs the reply through ClassifyReply so the returned error covers both transport and handler failure modes uniformly.

Transport failures (no responders, timeout) are returned as the underlying error and classify as ErrorTransient via pkg/errs.IsTransient — caller's existing retry logic on IsTransient continues to fire.

Handler failures arrive as a *errs.ClassifiedError reconstructed from the X-Error-Class / X-Error-Code headers + {message, detail} body. Caller branches on errs.IsInvalid / IsTransient / IsFatal.

Use this for QUERIES. For MUTATIONS where the responder is idempotent AND emits classified errors, use RequestWithRetryClassified — retrying on a hung query masks responder problems as latency. See docs/operations/07-nats-request-retry.md for the full rule.

func (*Client) RequestReady 

func (c *Client) RequestReady(
	ctx context.Context, subject string, data []byte, probeTimeout, budget time.Duration,
) ([]byte, error)

RequestReady performs a readiness-gated read: a QUERY that tolerates a not-yet-subscribed responder at cold start / after reconnect. It retries with probeTimeout per attempt up to a total budget, returning the first reply's data. Zero probeTimeout/budget use the Default* values above.

Use for the FIRST read on boot / initial reconcile / after a reconnect, where "no responder" means "not ready yet." For steady-state reads use Request (timeout = real signal). See docs/operations/07-nats-request-retry.md.

func (*Client) RequestReadyClassified 

func (c *Client) RequestReadyClassified(
	ctx context.Context, subject string, data []byte, probeTimeout, budget time.Duration,
) ([]byte, error)

RequestReadyClassified is the classified sibling of RequestReady (the readiness-gated read — the third bucket of the request doctrine). It runs the short-timeout, budget-bounded readiness loop, then runs the final reply through ClassifyReply so transport failures (no-responders / probe timeout, retried until the budget) and handler failures (X-Error-Class/Code headers) arrive uniformly.

Use for the FIRST read on boot / initial reconcile / after a reconnect, where a not-yet-subscribed responder must not degrade to a full-query-timeout hang. For steady-state reads use RequestClassified (timeout = real signal). See docs/operations/07-nats-request-retry.md.

func (*Client) RequestWithHeaders 

func (c *Client) RequestWithHeaders(
	ctx context.Context,
	subject string,
	data []byte,
	headers map[string]string,
	timeout time.Duration,
) (*nats.Msg, error)

RequestWithHeaders performs a request/reply operation with custom headers. Headers are passed as a map and converted to NATS message headers. Returns the full NATS message to allow access to response headers.

func (*Client) RequestWithRetry 

func (c *Client) RequestWithRetry(
	ctx context.Context,
	subject string,
	data []byte,
	timeout time.Duration,
	retry RetryConfig,
) ([]byte, error)

RequestWithRetry performs a request with configurable retry on failure. This is useful for handling transient "no responders" errors in NATS where the subscriber may not be ready when the request arrives.

Use this for MUTATIONS (write paths) where the responder is idempotent — the retry path may re-deliver the same request if the first attempt's response gets lost, so the responder must converge to the same state on duplicate receives. For QUERIES use Request instead; retrying on timeout masks hung responders as latency. See docs/operations/07-nats-request-retry.md for the full rule.

Footgun: when the responder returns a Go error, SubscribeForRequests wire-encodes the failure as a legacy "error: <msg>" text body with nil err. This method returns reply.Data on transport success without running it through ClassifyReply, so callers that json.Unmarshal the body silently corrupt on handler errors. For mutation paths that want both retry AND classified error handling, use RequestWithRetryClassified (closes the matrix gap filed as gh#192).

func (*Client) RequestWithRetryClassified 

func (c *Client) RequestWithRetryClassified(
	ctx context.Context,
	subject string,
	data []byte,
	timeout time.Duration,
	retry RetryConfig,
) ([]byte, error)

RequestWithRetryClassified is the retry-aware sibling of RequestClassified. It runs RequestWithRetry's retry-on-transport- failure loop and then runs the final reply through ClassifyReply so the returned error covers both transport (retried away) AND handler failure modes uniformly. Closes gh#192: pre-beta.93 the consumer-side method matrix exposed Request / RequestClassified / RequestWithRetry but had no retry+classify combination, forcing mutation-path callers to either skip the classified contract or wrap RequestWithRetry in a `json.Valid` pre-decode guard (the shape semteams shipped in cmd/semteams/tools/addsource/executor.go before this method existed).

Use this for MUTATIONS where the responder is idempotent AND emits classified errors. The classified contract round-trips per the same rules as RequestClassified: transport failures arrive as the underlying error after the retry budget exhausts (classifies as ErrorTransient via pkg/errs.IsTransient); handler failures arrive as *errs.ClassifiedError reconstructed from the X-Error-Class / X-Error-Code headers + {message, detail} body. Caller branches on errs.IsInvalid / IsTransient / IsFatal.

For QUERIES use RequestClassified; retrying on a hung query masks responder problems as latency. See docs/operations/07-nats-request-retry.md for the full mutation-vs-query rule.

func (*Client) SetConnection 

func (m *Client) SetConnection(conn *nats.Conn)

SetConnection sets the NATS connection (for testing)

func (*Client) Status 

func (m *Client) Status() ConnectionStatus

Status returns the current connection status

func (*Client) StopAllConsumers 

func (c *Client) StopAllConsumers()

StopAllConsumers stops all active consumers.

func (*Client) StopAndDeleteConsumer 

func (c *Client) StopAndDeleteConsumer(ctx context.Context, streamName, consumerName string) error

StopAndDeleteConsumer stops a specific consumer and deletes the durable consumer from the server. WARNING: This permanently removes the consumer's position. Use only for test cleanup or when you intentionally want to reset consumer state.

func (*Client) StopConsumer 

func (c *Client) StopConsumer(streamName, consumerName string)

StopConsumer stops a specific consumer by stream and consumer name. This stops the local consume context but does not delete the durable consumer from the server.

func (*Client) Subscribe 

func (m *Client) Subscribe(ctx context.Context, subject string, handler func(context.Context, *nats.Msg)) (*Subscription, error)

Subscribe subscribes to a NATS subject with context propagation. Each message handler receives the full *nats.Msg to access Subject, Data, Headers, etc. This is essential for wildcard subscriptions where the actual subject differs from the pattern. The context is derived from the parent context with a 30-second timeout for message processing. Returns a Subscription handle that can be used to unsubscribe.

func (*Client) SubscribeForRequests 

func (c *Client) SubscribeForRequests(
	ctx context.Context,
	subject string,
	handler func(ctx context.Context, data []byte) ([]byte, error),
) (*Subscription, error)

SubscribeForRequests subscribes to a subject and handles request/reply patterns. The handler receives the message data and reply subject, and should return the response data or an error. Returns the Subscription so the caller can unsubscribe when done. This is a convenience method for implementing request/reply services.

func (*Client) URLs 

func (m *Client) URLs() string

URLs returns the NATS server URLs (comma-separated for clustering)

func (*Client) WaitForBucket 

func (m *Client) WaitForBucket(ctx context.Context, name string, timeout time.Duration) (jetstream.KeyValue, error)

WaitForBucket waits for a KV bucket to become available, retrying until the timeout expires or the context is cancelled. Use this when a component depends on a bucket created by another component with unpredictable startup timing.

For more advanced patterns (background recovery, loss detection), use pkg/resource.Watcher directly.

func (*Client) WaitForConnection 

func (m *Client) WaitForConnection(ctx context.Context) error

WaitForConnection waits for the connection to be established

func (*Client) WithHealthCheck 

func (m *Client) WithHealthCheck(interval time.Duration)

WithHealthCheck enables health monitoring with a specified interval

type ClientOption 

type ClientOption func(*Client) error

ClientOption is a functional option for configuring the Client

func WithCircuitBreakerThreshold 

func WithCircuitBreakerThreshold(threshold int32) ClientOption

WithCircuitBreakerThreshold sets the number of failures before opening circuit

func WithCompression 

func WithCompression(enabled bool) ClientOption

WithCompression enables message compression

func WithConnectionLossTimeout 

func WithConnectionLossTimeout(grace time.Duration) ClientOption

WithConnectionLossTimeout configures how long the client tolerates a continuous broker outage before the connection-lost callback fires.

When set together with WithConnectionLostCallback, a timer arms on the first disconnect; if the connection has not recovered within grace, the callback runs once with the original disconnect error. A reconnect before the deadline cancels the timer. A non-positive grace disables the watchdog (legacy behavior — onConnectionLost never fires automatically).

The callback only signals — it does not call os.Exit or otherwise act on the client. Callers decide the policy (graceful shutdown, alerting, degraded-mode flag, etc.).

func WithConnectionLostCallback 

func WithConnectionLostCallback(fn func(error)) ClientOption

WithConnectionLostCallback sets a callback for when connection is completely lost

func WithCredentials 

func WithCredentials(username, password string) ClientOption

WithCredentials sets username and password for authentication

func WithDisconnectCallback 

func WithDisconnectCallback(fn func(error)) ClientOption

WithDisconnectCallback sets a callback for disconnection events This is in addition to NATS's built-in disconnect handler

func WithDrainTimeout 

func WithDrainTimeout(d time.Duration) ClientOption

WithDrainTimeout sets the timeout for draining on disconnect

func WithHealthChangeCallback 

func WithHealthChangeCallback(fn func(healthy bool)) ClientOption

WithHealthChangeCallback sets a callback for health status changes

func WithHealthInterval 

func WithHealthInterval(d time.Duration) ClientOption

WithHealthInterval sets the interval for health monitoring

func WithLogger 

func WithLogger(logger *slog.Logger) ClientOption

WithLogger sets a structured logger for the client. If nil, defaults to slog.Default().

func WithMaxBackoff 

func WithMaxBackoff(d time.Duration) ClientOption

WithMaxBackoff sets the maximum backoff duration for circuit breaker

func WithMaxReconnects 

func WithMaxReconnects(maxN int) ClientOption

WithMaxReconnects sets the maximum number of reconnection attempts (-1 for infinite)

func WithMetrics 

func WithMetrics(registry *metric.MetricsRegistry) ClientOption

WithMetrics enables JetStream metrics collection using the provided registry. Metrics will track streams and consumers created through this client.

func WithName 

func WithName(name string) ClientOption

WithName sets the client name for identification

func WithPingInterval 

func WithPingInterval(d time.Duration) ClientOption

WithPingInterval sets the ping interval for connection health checks

func WithReconnectCallback 

func WithReconnectCallback(fn func()) ClientOption

WithReconnectCallback sets a callback for reconnection events This is in addition to NATS's built-in reconnect handler

func WithReconnectWait 

func WithReconnectWait(d time.Duration) ClientOption

WithReconnectWait sets the wait time between reconnection attempts

func WithTLS 

func WithTLS(certFile, keyFile, caFile string) ClientOption

WithTLS enables TLS with optional certificate paths

func WithTimeout 

func WithTimeout(d time.Duration) ClientOption

WithTimeout sets the connection timeout

func WithToken 

func WithToken(token string) ClientOption

WithToken sets a token for authentication

type Codec 

type Codec[T any] interface {
	Marshal(T) ([]byte, error)
	Unmarshal([]byte, *T) error
}

Codec defines the serialization interface for typed subjects. Implementations handle marshaling and unmarshaling of typed payloads.

type ConnectionStatus 

type ConnectionStatus int

ConnectionStatus represents the state of the NATS connection 

const (
	StatusDisconnected ConnectionStatus = iota
	StatusConnecting
	StatusConnected
	StatusReconnecting
	StatusCircuitOpen
)

Possible connection statuses

func (ConnectionStatus) String 

func (s ConnectionStatus) String() string

String returns the string representation of ConnectionStatus

type JSONCodec 

type JSONCodec[T any] struct{}

JSONCodec provides JSON serialization for typed subjects. This is the default codec for most use cases.

func (JSONCodec[T]) Marshal 

func (c JSONCodec[T]) Marshal(v T) ([]byte, error)

Marshal serializes a value to JSON bytes.

func (JSONCodec[T]) Unmarshal 

func (c JSONCodec[T]) Unmarshal(data []byte, v *T) error

Unmarshal deserializes JSON bytes into a value.

type KVEntry 

type KVEntry struct {
	Key      string
	Value    []byte
	Revision uint64
}

KVEntry wraps a KV entry with its revision for CAS operations

type KVOptions 

type KVOptions struct {
	MaxRetries            int           // Maximum CAS retry attempts
	RetryDelay            time.Duration // Initial delay between retries
	Timeout               time.Duration // Operation timeout
	MaxValueSize          int           // Maximum size for values (default: 1MB)
	UseExponentialBackoff bool          // Enable exponential backoff with jitter
	MaxRetryDelay         time.Duration // Maximum delay between retries
}

KVOptions configures KV operations behavior

func DefaultKVOptions 

func DefaultKVOptions() KVOptions

DefaultKVOptions returns sensible defaults matching Graph processor

type KVStore 

type KVStore struct {
	// contains filtered or unexported fields
}

KVStore provides high-level KV operations with built-in CAS support

func (*KVStore) AssertNoLifecycleRetention 

func (kv *KVStore) AssertNoLifecycleRetention(ctx context.Context, name string) error

AssertNoLifecycleRetention reads this bucket's retention config and returns ErrGraphBucketRetention if it is binding — the boot-time D1 guardrail for a live graph bucket (ADR-068). A retention config here means some process won the get-or-create race with a TTL/size cap; fail-closed rather than silently expire graph state.

func (*KVStore) Create 

func (kv *KVStore) Create(ctx context.Context, key string, value []byte) (uint64, error)

Create only creates if key doesn't exist (returns error if exists)

func (*KVStore) Delete 

func (kv *KVStore) Delete(ctx context.Context, key string) error

Delete removes a key from the bucket

func (*KVStore) Get 

func (kv *KVStore) Get(ctx context.Context, key string) (*KVEntry, error)

Get retrieves a value with its revision for CAS operations

func (*KVStore) Keys 

func (kv *KVStore) Keys(ctx context.Context) ([]string, error)

Keys returns all keys in the bucket

func (*KVStore) KeysByFilter 

func (kv *KVStore) KeysByFilter(ctx context.Context, pattern string) ([]string, error)

KeysByFilter returns all keys matching an exact NATS subject filter. Unlike KeysByPrefix, this supports fixed-position wildcards such as "*.org.platform.domain.system.type.instance".

A filtered key listing is a correctness boundary for derived-index reconciliation. The NATS KeyLister closes its channel when ctx is cancelled, but it does not expose a terminal error. Callers must therefore reject the collected slice when the context expires; returning it as success would turn a partial owner snapshot into authoritative truth.

func (*KVStore) KeysByPrefix 

func (kv *KVStore) KeysByPrefix(ctx context.Context, prefix string) ([]string, error)

KeysByPrefix returns all keys matching the given prefix. Uses JetStream KV's native key filtering for efficient server-side filtering. The prefix is automatically converted to a NATS wildcard pattern (prefix + ">").

func (*KVStore) Put 

func (kv *KVStore) Put(ctx context.Context, key string, value []byte) (uint64, error)

Put creates or updates a key without revision check (last writer wins)

func (*KVStore) Update 

func (kv *KVStore) Update(ctx context.Context, key string, value []byte, revision uint64) (uint64, error)

Update performs CAS update with explicit revision

func (*KVStore) UpdateJSON 

func (kv *KVStore) UpdateJSON(ctx context.Context, key string,
	updateFn func(current map[string]any) error) error

UpdateJSON performs CAS update on JSON data with automatic retry

func (*KVStore) UpdateWithRetry 

func (kv *KVStore) UpdateWithRetry(ctx context.Context, key string,
	updateFn func(current []byte) ([]byte, error)) error

UpdateWithRetry performs CAS update with automatic retry on conflicts If the key doesn't exist, it creates it

func (*KVStore) Watch 

func (kv *KVStore) Watch(ctx context.Context, pattern string) (jetstream.KeyWatcher, error)

Watch creates a watcher for key changes Note: Watch does not apply timeout as it creates a long-lived watcher

type PermanentDeliveryError 

type PermanentDeliveryError struct {
	// contains filtered or unexported fields
}

PermanentDeliveryError marks a handler failure as structurally permanent for this exact message. ConsumeWithHeartbeat terminates the JetStream delivery instead of retrying it. Unwrap preserves the handler's typed error contract.

func (*PermanentDeliveryError) Error 

func (e *PermanentDeliveryError) Error() string

func (*PermanentDeliveryError) Unwrap 

func (e *PermanentDeliveryError) Unwrap() error

type RetryConfig 

type RetryConfig struct {
	MaxRetries        int           // Number of retry attempts (default: 3)
	InitialBackoff    time.Duration // First retry delay (default: 100ms)
	MaxBackoff        time.Duration // Cap on backoff growth (default: 2s)
	BackoffMultiplier float64       // Exponential growth factor (default: 2.0)
}

RetryConfig configures retry behavior for requests.

func DefaultRetryConfig 

func DefaultRetryConfig() RetryConfig

DefaultRetryConfig returns sensible defaults for retry configuration.

type Status 

type Status struct {
	Status          ConnectionStatus
	FailureCount    int32
	LastFailureTime time.Time
	Reconnects      int32
	RTT             time.Duration
}

Status holds runtime status information for the NATS manager

type StreamAutoCreateConfig 

type StreamAutoCreateConfig struct {
	// Subjects for the stream. If empty, derived from FilterSubject.
	Subjects []string

	// Storage type: "file" (default) or "memory"
	Storage string

	// Retention policy: "limits" (default), "interest", "work_queue"
	Retention string

	// MaxAge is the maximum age of messages (default 7 days).
	MaxAge time.Duration

	// Duplicates is the server-side duplicate-detection window for the
	// Nats-Msg-Id header (ADR-055 §5 "T1"). Zero leaves the NATS server
	// default (2m). Must be <= MaxAge or the server rejects creation;
	// ensureStreamForConsumer clamps it down to MaxAge when it exceeds it.
	Duplicates time.Duration

	// MaxBytes is the maximum total size (0 = unlimited).
	MaxBytes int64

	// MaxMsgs is the maximum number of messages (0 = unlimited).
	MaxMsgs int64

	// Replicas is the number of replicas (default 1).
	Replicas int
}

StreamAutoCreateConfig configures automatic stream creation.

func DefaultStreamConfig 

func DefaultStreamConfig() *StreamAutoCreateConfig

DefaultStreamConfig returns default auto-create configuration.

type StreamConsumerConfig 

type StreamConsumerConfig struct {
	// StreamName is the name of the stream to consume from (required).
	StreamName string

	// ConsumerName is the durable consumer name. If empty, creates an ephemeral consumer.
	ConsumerName string

	// FilterSubject filters messages within the stream. If empty, receives all messages.
	FilterSubject string

	// DeliverPolicy determines where to start delivering messages.
	// Options: "all" (default), "last", "new", "by_start_time"
	DeliverPolicy string

	// AckPolicy determines how messages are acknowledged.
	// Options: "explicit" (default), "none", "all"
	AckPolicy string

	// MaxDeliver is the maximum number of delivery attempts (0 = unlimited).
	MaxDeliver int

	// AckWait is how long to wait for an ack before redelivery.
	// Default is 30 seconds.
	AckWait time.Duration

	// MaxAckPending limits the number of outstanding (unacknowledged) messages
	// that can be delivered to a consumer. This provides backpressure to prevent
	// overwhelming the consumer. 0 leaves it unset, so the NATS server applies its
	// default of 1000 for explicit-ack consumers; -1 means unlimited (gh#480).
	MaxAckPending int

	// AutoCreate enables automatic stream creation if it doesn't exist.
	AutoCreate bool

	// AutoCreateConfig is used when auto-creating a stream.
	// If nil, defaults are used based on FilterSubject.
	AutoCreateConfig *StreamAutoCreateConfig

	// BackOff overrides AckWait per retry attempt. Index 0 is the first retry
	// wait, index 1 is the second, and so on. The last value is used for all
	// subsequent retries. If empty, AckWait applies uniformly.
	BackOff []time.Duration

	// MessageTimeout is the context timeout for processing each message.
	// This timeout is passed to the handler and should accommodate the full
	// processing time including any downstream calls (e.g., LLM requests).
	// Default is 30 seconds if not specified.
	MessageTimeout time.Duration

	// DisableMessageTimeout keeps the handler context bound to the consumer
	// lifecycle. Use only when the handler applies its own ordinary work deadline
	// and needs to retain an in-flight delivery across that deadline.
	DisableMessageTimeout bool
}

StreamConsumerConfig configures a JetStream consumer.

type Subject 

type Subject[T any] struct {
	// Pattern is the NATS subject pattern (may include wildcards for subscribe)
	Pattern string

	// Codec handles serialization/deserialization
	Codec Codec[T]
}

Subject represents a typed NATS subject with compile-time type safety. It binds a subject pattern to a specific payload type and codec.

Example: 

var WorkflowStarted = Subject[WorkflowStartedEvent]{
    Pattern: "workflow.events.started",
    Codec:   JSONCodec[WorkflowStartedEvent]{},
}

// Type-safe publish
err := WorkflowStarted.Publish(ctx, client, event)

// Type-safe subscribe
sub, err := WorkflowStarted.Subscribe(ctx, client, func(ctx context.Context, event WorkflowStartedEvent) error {
    // event is already typed - no assertions needed
    return nil
})

func NewSubject 

func NewSubject[T any](pattern string) Subject[T]

NewSubject creates a typed subject with a JSON codec (most common case).

func NewSubjectWithCodec 

func NewSubjectWithCodec[T any](pattern string, codec Codec[T]) Subject[T]

NewSubjectWithCodec creates a typed subject with a custom codec.

func (Subject[T]) Publish 

func (s Subject[T]) Publish(ctx context.Context, client *Client, payload T) error

Publish sends a typed payload to the subject. The payload is serialized using the subject's codec before publishing.

func (Subject[T]) PublishToStream 

func (s Subject[T]) PublishToStream(ctx context.Context, client *Client, payload T) error

PublishToStream sends a typed payload to a JetStream subject. The payload is serialized using the subject's codec before publishing.

func (Subject[T]) Subscribe 

func (s Subject[T]) Subscribe(ctx context.Context, client *Client, handler func(context.Context, T) error) (*Subscription, error)

Subscribe creates a subscription to the subject with type-safe message handling. The handler receives deserialized payloads directly.

func (Subject[T]) SubscribeWithMsg 

func (s Subject[T]) SubscribeWithMsg(ctx context.Context, client *Client, handler func(context.Context, *nats.Msg, T) error) (*Subscription, error)

SubscribeWithMsg creates a subscription that provides both the typed payload and raw message. Use this when you need access to message metadata (subject, headers, etc.).

type Subscription 

type Subscription struct {
	// contains filtered or unexported fields
}

Subscription wraps a NATS subscription for lifecycle management

func (*Subscription) Unsubscribe 

func (s *Subscription) Unsubscribe() error

Unsubscribe unsubscribes from the subject

type TemporalResolver 

type TemporalResolver struct {
	// contains filtered or unexported fields
}

TemporalResolver provides efficient timestamp-based KV queries with caching

func NewTemporalResolver 

func NewTemporalResolver(ctx context.Context, bucket jetstream.KeyValue) (*TemporalResolver, error)

NewTemporalResolver creates a resolver for timestamp-based queries The context is used for the cache background cleanup goroutine lifecycle

func NewTemporalResolverWithCache 

func NewTemporalResolverWithCache(
	ctx context.Context,
	bucket jetstream.KeyValue,
	cacheTTL time.Duration,
) (*TemporalResolver, error)

NewTemporalResolverWithCache creates a resolver with custom cache TTL The context is used for the cache background cleanup goroutine lifecycle

func (*TemporalResolver) Close 

func (tr *TemporalResolver) Close() error

Close shuts down the temporal resolver and its cache

func (*TemporalResolver) GetAtTimestamp 

func (tr *TemporalResolver) GetAtTimestamp(
	ctx context.Context,
	key string,
	targetTime time.Time,
) (jetstream.KeyValueEntry, error)

GetAtTimestamp finds the entity state that was current at the given timestamp Uses binary search for O(log n) performance with caching

func (*TemporalResolver) GetInTimeRange 

func (tr *TemporalResolver) GetInTimeRange(
	ctx context.Context,
	key string,
	startTime, endTime time.Time,
) ([]jetstream.KeyValueEntry, error)

GetInTimeRange finds all entity states within a time range Returns states ordered by timestamp

func (*TemporalResolver) GetRangeAtTimestamp 

func (tr *TemporalResolver) GetRangeAtTimestamp(
	ctx context.Context,
	keys []string,
	targetTime time.Time,
) (map[string]jetstream.KeyValueEntry, error)

GetRangeAtTimestamp finds multiple entities at a specific timestamp Useful for reconstructing entire system state at time T

func (*TemporalResolver) GetRangeInTimeRange 

func (tr *TemporalResolver) GetRangeInTimeRange(
	ctx context.Context,
	keys []string,
	startTime, endTime time.Time,
) (map[string][]jetstream.KeyValueEntry, error)

GetRangeInTimeRange finds multiple entities within a time range Returns a map of key -> entries within the range

func (*TemporalResolver) GetStats 

func (tr *TemporalResolver) GetStats() *cache.Statistics

GetStats returns cache statistics for monitoring

type TestClient 

type TestClient struct {
	Client        *Client // Drop-in replacement for existing natsclient.Client
	URL           string
	MonitoringURL string // Read-only URL for the actual mapped NATS monitoring endpoint
	BucketPrefix  string // Prefix applied to all KV bucket names for test isolation
	// contains filtered or unexported fields
}

TestClient provides testcontainers-based NATS for testing

func NewSharedTestClient 

func NewSharedTestClient(opts ...TestOption) (*TestClient, error)

NewSharedTestClient creates a new NATS test container for use in TestMain Unlike NewTestClient, this doesn't require testing.T and returns errors

func NewTestClient 

func NewTestClient(t testing.TB, opts ...TestOption) *TestClient

NewTestClient creates a new NATS test container Accepts testing.TB so it works with both *testing.T and *testing.B

func (*TestClient) CreateKVBucket 

func (tc *TestClient) CreateKVBucket(ctx context.Context, name string) (jetstream.KeyValue, error)

CreateKVBucket is a helper for creating KV buckets during tests. The bucket prefix is automatically applied if configured.

func (*TestClient) CreateStream 

func (tc *TestClient) CreateStream(ctx context.Context, name string, subjects []string) (jetstream.Stream, error)

CreateStream is a helper for creating JetStream streams during tests

func (*TestClient) GetKVBucket 

func (tc *TestClient) GetKVBucket(ctx context.Context, name string) (jetstream.KeyValue, error)

GetKVBucket is a helper for getting existing KV buckets during tests. The bucket prefix is automatically applied if configured.

func (*TestClient) GetNativeConnection 

func (tc *TestClient) GetNativeConnection() *gonats.Conn

GetNativeConnection returns the underlying NATS connection for direct access

func (*TestClient) GetStream 

func (tc *TestClient) GetStream(ctx context.Context, name string) (jetstream.Stream, error)

GetStream is a helper for getting existing JetStream streams during tests

func (*TestClient) IsReady 

func (tc *TestClient) IsReady() bool

IsReady checks if the NATS connection is ready for use

func (*TestClient) PrefixedBucketName 

func (tc *TestClient) PrefixedBucketName(name string) string

PrefixedBucketName returns the full bucket name with prefix applied. Use this when you need to pass bucket names to components that create their own buckets.

func (*TestClient) Terminate 

func (tc *TestClient) Terminate() error

Terminate manually terminates the container and client (usually handled by t.Cleanup)

type TestOption 

type TestOption func(*testConfig)

TestOption for configuring test client

func WithBucketPrefix 

func WithBucketPrefix(prefix string) TestOption

WithBucketPrefix sets a prefix for all KV bucket names to enable test isolation. When tests run in parallel, each test can use a unique prefix (e.g., test name) to avoid bucket name collisions.

func WithE2EDefaults 

func WithE2EDefaults() TestOption

WithE2EDefaults configures NATS with settings good for end-to-end tests

func WithFastStartup 

func WithFastStartup() TestOption

WithFastStartup configures NATS for fastest possible RUNTIME (short connection timeout, no extra features). Container startup itself still needs a realistic budget — Docker API responsiveness under parallel test pressure routinely exceeds 10s even when NATS itself is ready in under a second. Use the package default (30s) for the container startup wait; the "fast" part is the 2s connection timeout once the container is up.

func WithFileStorage 

func WithFileStorage() TestOption

WithFileStorage enables file-backed JetStream storage instead of the default memory-only store. Use this when tests create many KV buckets or write large volumes of data that would exceed the 256MB default memory limit.

func WithIntegrationDefaults 

func WithIntegrationDefaults() TestOption

WithIntegrationDefaults configures NATS with settings good for integration tests

func WithJetStream 

func WithJetStream() TestOption

WithJetStream enables JetStream for tests that need it

func WithKV 

func WithKV() TestOption

WithKV enables KV store for tests that need it

func WithKVBuckets 

func WithKVBuckets(buckets ...string) TestOption

WithKVBuckets pre-creates specific KV buckets

func WithMinimalFeatures 

func WithMinimalFeatures() TestOption

WithMinimalFeatures configures NATS with only basic pub/sub (fastest startup)

func WithNATSVersion 

func WithNATSVersion(version string) TestOption

WithNATSVersion specifies a specific NATS server version to use

func WithProductionLike 

func WithProductionLike() TestOption

WithProductionLike configures NATS with settings that mimic production

func WithStartTimeout 

func WithStartTimeout(timeout time.Duration) TestOption

WithStartTimeout sets the container startup timeout

func WithStreams 

func WithStreams(streams ...TestStreamConfig) TestOption

WithStreams pre-creates JetStream streams for testing

func WithTestTimeout 

func WithTestTimeout(timeout time.Duration) TestOption

WithTestTimeout sets the connection timeout for test client

type TestStreamConfig 

type TestStreamConfig struct {
	Name     string
	Subjects []string
}

TestStreamConfig defines a stream to pre-create for testing

type TraceContext 

type TraceContext struct {
	TraceID      string // 32 hex chars (16 bytes)
	SpanID       string // 16 hex chars (8 bytes)
	ParentSpanID string // 16 hex chars, empty for root span
	Sampled      bool
}

TraceContext holds trace information propagated through the system

func ExtractTrace 

func ExtractTrace(msg *nats.Msg) *TraceContext

ExtractTrace reads trace headers from a NATS message

func ExtractTraceFromJetStream 

func ExtractTraceFromJetStream(headers nats.Header) *TraceContext

ExtractTraceFromJetStream reads trace headers from a JetStream message

func NewTraceContext 

func NewTraceContext() *TraceContext

NewTraceContext creates a new trace context with generated IDs

func ParseTraceparent 

func ParseTraceparent(header string) (*TraceContext, error)

ParseTraceparent parses W3C traceparent header Format: {version}-{trace_id}-{span_id}-{flags} Example: 00-0af7651916cd43dd8448eb211c80319c-b7ad6b7169203331-01

func TraceContextFromContext 

func TraceContextFromContext(ctx context.Context) (*TraceContext, bool)

TraceContextFromContext extracts trace context from context

func (*TraceContext) FormatTraceparent 

func (tc *TraceContext) FormatTraceparent() string

FormatTraceparent formats trace context as W3C traceparent

func (*TraceContext) NewSpan 

func (tc *TraceContext) NewSpan() *TraceContext

NewSpan creates a child span from existing trace context

Source Files

View all Source files

Jump to

Keyboard shortcuts

? : This menu
/ : Search site
f or F : Jump to
y or Y : Canonical URL
go.dev uses cookies from Google to deliver and enhance the quality of its services and to analyze traffic. Learn more.