natsclient

package
v1.0.0-beta.159 Latest Latest
Warning

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

 Go to latest Published: Jul 31, 2026 License: MIT Imports: 27 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")
    }),
)

Who owns a stream's limits

A stream's limits belong to the component that DECLARES it. If you only READ a stream, bind it by name with GetStream and declare nothing: 

if _, err := client.GetStream(ctx, streamName); err != nil {
    if errors.Is(err, jetstream.ErrStreamNotFound) {
        // A real answer: the declaring component is not deployed here.
        // Handle it — do not create the stream to make the error go away.
        return nil
    }
    return err
}

This is not style. EnsureStream is get-or-create: a reader that calls it with its own configuration either creates the stream with limits it does not own, or binds an existing one and has its declaration silently discarded — after which the stream's limits are decided permanently by boot order. Since the framework's provisioner reconciles retention drift, two components declaring one stream differently is worse than that: each repairs the other's value on every boot, forever. Neither can detect it locally, because a caller sees its own declaration and the live stream, never the other declaration.

EnsureStream reports a divergence between what you declared and what a bound stream actually carries (see DiffDeclaredStream). A report that returns on every boot, with the observed value alternating, is that flap.

JetStream Operations

Working with JetStream streams and consumers: 

// Create a stream. An ordinary stream MUST declare a finite MaxAge and a
// finite MaxBytes: 0 and -1 both mean unlimited to JetStream, so neither is a
// declaration, and creation is refused without them. Set Discard explicitly
// too — it cannot be required (DiscardOld is the field's zero value) but at a
// finite MaxBytes it decides whether the oldest messages are evicted or the
// newest are refused.
stream, err := client.CreateStream(ctx, jetstream.StreamConfig{
    Name:     "EVENTS",
    Subjects: []string{"events.>"},
    MaxAge:   24 * time.Hour,
    MaxBytes: 1 << 30,
    Discard:  jetstream.DiscardOld,
})

// 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 (
	// KVStreamPrefix is the NATS naming convention for a KV bucket's backing
	// stream: KV_<bucket> (nats.go jetstream/kv.go, kvBucketNamePre = "KV_").
	KVStreamPrefix = "KV_"

	// ObjectStoreStreamPrefix is the NATS naming convention for an
	// ObjectStore's backing stream: OBJ_<bucket> (nats.go
	// jetstream/object.go, objNameTmpl = "OBJ_%s").
	ObjectStoreStreamPrefix = "OBJ_"
)

Backing-stream naming conventions. NATS implements every KV bucket and every ObjectStore as an ordinary JetStream stream with a reserved name prefix, so a bucket's lifecycle-eviction config (MaxAge/MaxBytes/Discard) is reachable through the plain stream API. That reachability is what the retention reconcilers here and in storage/objectstore depend on — and it is also what lets an unfiltered provisioning seam stamp age eviction onto authoritative graph state, which is why every such seam refuses these prefixes outright.

These are the ONE definition of each prefix. The KV and ObjectStore retention paths and every provisioning guard read them from here so the convention cannot fork into per-package literals.

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 (
	// OvercommitmentUnavailableUnboundedLimit means the server reports no
	// limit for the tier. The resources are still named and their declared sum
	// still reported; only the verdict is withheld.
	OvercommitmentUnavailableUnboundedLimit = "account-limit-unbounded"
	// OvercommitmentUnavailableUnknownLimit means the limit could not be
	// determined — AccountInfo was unreadable, or the value the server returned
	// is ambiguous.
	OvercommitmentUnavailableUnknownLimit = "account-limit-unknown"
)

Reasons an over-commitment comparison is not applicable. They are distinct because they resolve differently: the first is a deliberate operator choice and needs no action, the second is a gap in what this process can read.

View Source 
const (
	// GrowthUnavailableNoPriorObservation means fewer than two observations of
	// this resource exist yet — a new resource, or a fresh report bucket.
	GrowthUnavailableNoPriorObservation = "no-prior-observation"
	// GrowthUnavailableUnknownUsage means the current usage could not be read,
	// so there is nothing to difference.
	GrowthUnavailableUnknownUsage = "unknown-usage"
	// GrowthUnavailableObservationsTooClose means observations exist but none
	// is separated from the current one by MinGrowthSampleInterval.
	GrowthUnavailableObservationsTooClose = "observations-too-close"
)

Reasons a growth rate is unavailable. They are distinct because they resolve differently: the first resolves on the next collection, the second needs the resource to become describable, and the third needs the collection interval (or the publication cadence) to be wider than MinGrowthSampleInterval.

View Source 
const (
	// DefaultStorageInventoryInterval is how often the account is enumerated
	// when the caller does not choose.
	DefaultStorageInventoryInterval = time.Minute
	// DefaultStorageInventoryTimeout bounds ONE collection, both listings
	// included. It is the collector's own bound, not the caller's, so an
	// unbounded caller context cannot make a collection unbounded.
	DefaultStorageInventoryTimeout = 15 * time.Second
)

Collection defaults. Both are configuration: every process polling account-wide multiplies cost by deployment size, so an operator running many instances turns the interval down rather than losing the view entirely.

View Source 
const (
	// DefaultWarningHeadroom is the free fraction at or below which a resource
	// reports warning.
	DefaultWarningHeadroom = 0.25
	// DefaultHighHeadroom is the free fraction at or below which a resource
	// reports high.
	DefaultHighHeadroom = 0.15
	// DefaultCriticalHeadroom is the free fraction at or below which a resource
	// reports critical. It is ALSO the level the time projection targets.
	DefaultCriticalHeadroom = 0.05

	// DefaultWarningHorizon is the projected time-to-threshold at or below
	// which a resource reports warning.
	DefaultWarningHorizon = 72 * time.Hour
	// DefaultHighHorizon is the projected time-to-threshold at or below which a
	// resource reports high.
	DefaultHighHorizon = 24 * time.Hour
	// DefaultCriticalHorizon is the projected time-to-threshold at or below
	// which a resource reports critical.
	DefaultCriticalHorizon = 4 * time.Hour
)

Default pressure thresholds. Headroom bands are the FRACTION of the configured bound still free; horizon bands are the projected time until the resource crosses the critical-headroom level.

The defaults are deliberately generous on time and tight on space: a three-day warning horizon is roughly the lead time needed to plan and apply a capacity change, while 25% free is late enough that it is not noise.

View Source 
const (
	// ProjectionUnavailableUnknownCapacity means the limit or usage could not
	// be read. Projecting from it would fabricate a number.
	ProjectionUnavailableUnknownCapacity = "unknown-capacity"
	// ProjectionUnavailableUnbounded means the resource declares no bound.
	// There is nothing to have headroom against and nothing to run out of.
	ProjectionUnavailableUnbounded = "unbounded"
	// ProjectionUnavailableUnknownGrowth means the rate has not been measured
	// yet, so no time can be projected. Headroom is still reported.
	ProjectionUnavailableUnknownGrowth = "unknown-growth-rate"
	// ProjectionUnavailableNotGrowing means the rate was measured and is zero
	// or negative: the threshold is never reached. Distinct from
	// unknown-growth-rate, because "measured, and not growing" is a finding an
	// operator can rely on.
	ProjectionUnavailableNotGrowing = "not-growing"
)

Reasons a projection is unavailable. Each names a different missing input, so an operator reading a suppressed projection can tell whether to fix the server, the declaration, or simply wait for the next collection.

View Source 
const (
	// PressureUnavailableUnknownCapacity means the capacity could not be read.
	// Reporting normal here would manufacture confidence about a resource
	// nobody can measure.
	PressureUnavailableUnknownCapacity = "unknown-capacity"
	// PressureUnavailableUnbounded means the capacity handed to AssessPressure
	// declares no bound, so neither band has an input.
	//
	// For a RESOURCE it is not the final word: PressureAgainstAccountTier
	// re-evaluates it against the only ceiling it has, and the more specific
	// unbounded-* reasons below are what a resource ends up carrying. This value
	// survives on a TIER row — AssessPressure over an unbounded account limit —
	// and as the fallback when a bounded tier reports no reason of its own.
	PressureUnavailableUnbounded = "unbounded"
	// PressureUnavailableUnboundedNoTierCeiling means the resource declares no
	// bound AND its storage tier's account limit is itself unbounded. Nothing
	// anywhere constrains it, so there is genuinely nothing to project — as
	// distinct from a ceiling that exists and could not be read.
	PressureUnavailableUnboundedNoTierCeiling = "unbounded-no-account-tier-ceiling"
	// PressureUnavailableUnboundedTierUnknown means the resource declares no
	// bound and its tier's account limit could not be read. The ceiling may well
	// exist; this process cannot see it, which is a gap rather than a licence to
	// report the resource as fine.
	PressureUnavailableUnboundedTierUnknown = "unbounded-account-tier-unknown"
	// PressureUnavailableUnboundedTierUnfiled means the resource declares no
	// bound and could not be filed under any tier, so no account ceiling applies
	// to it that this process can name.
	//
	// NOT REACHABLE from the production collector: DeriveAccountReport emits a row
	// for every tier any resource reports, creating the unknown-tier row on demand,
	// so a collected resource always finds one. It exists as the fail-closed arm
	// for a hand-built inventory, and reporting no state is the right answer there
	// rather than borrowing an arbitrary tier's.
	PressureUnavailableUnboundedTierUnfiled = "unbounded-account-tier-unfiled"
)

Reasons no pressure state exists for a resource.

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 DefaultReportConsumerRetryBackoff = 5 * time.Second

DefaultReportConsumerRetryBackoff is how long the consumer waits before re-establishing a watch that could not be created or that ended.

View Source 
const DefaultRequestHandlerTimeout = 30 * time.Second

DefaultRequestHandlerTimeout bounds a single inbound request-handler invocation (SubscribeForRequests). It caps how long a handler may run before its context is cancelled, so a wedged or pathologically slow handler cannot pin a NATS delivery goroutine indefinitely. This is the CI/default value and MUST NOT change without weighing every request handler in the tree; slow-by-design handlers (e.g. an LLM answer-synthesis path that legitimately needs >30s) raise it per deployment via WithRequestHandlerTimeout or the SEMSTREAMS_NATS_REQUEST_HANDLER_TIMEOUT environment variable rather than editing this constant.

View Source 
const DefaultRequestTimeout = 5 * time.Second

DefaultRequestTimeout is the default timeout for request/reply operations.

View Source 
const ErrorCodeBucketNotReady = "index_not_ready"

ErrorCodeBucketNotReady is the classified error code OpenFrameworkBucket returns for an absent must-exist bucket. Its value deliberately equals graph.ErrorCodeIndexNotReady (the code graph/query already emits for a not-sound-to-read index) so classified consumers handle "the bucket's owner has not provisioned it yet" with the same retry-with-backoff posture; a cross-pin test in graph asserts the two can never drift.

View Source 
const MinGrowthSampleInterval = 5 * time.Second

MinGrowthSampleInterval is the shortest interval between two observations that yields a rate worth publishing.

Below it the measurement is dominated by sampling jitter rather than growth: two publications a second apart with one 50 MB write between them read as 50 MB/s and project exhaustion in seconds. That is the false-critical class this capability exists to remove, so a too-close pair reports UNKNOWN — and the caller keeps the older baseline rather than advancing it, so the series converges on a usable interval instead of resetting to "now" forever.

View Source 
const SkipReasonStaleInventory = "inventory is stale; last-good is not a new observation"

SkipReasonStaleInventory is the SkipReason for an inventory whose most recent collection did not succeed.

View Source 
const StorageAccountReportKey = "_account.tiers"

StorageAccountReportKey is the reserved key carrying the per-tier account report (AccountReport) rather than a resource row.

It is RESERVED by construction, not by convention. Every resource key is exactly ONE key token: a JetStream stream name may not contain a dot, and StorageReportKey's fallback opaque token is a single token too. A key containing a dot is therefore unreachable from any resource name, however hostile — which is what lets one bucket carry two row kinds with no possibility of one addressing the other. A consumer discriminates on the key.

View Source 
const StorageReportGrowthObservations uint8 = 2

StorageReportGrowthObservations is how many observations of a resource the rate derivation needs: two, the current one and one retained prior one. It is the FLOOR on the report bucket's declared History depth, not the depth that bucket should carry.

The distinction is worth stating because the obvious argument for depth is wrong. A publisher seeds from the history BEFORE it writes, so the row it is about to compact is the row it just read: a single restart recovers its baseline at any depth, including 1. Depth becomes load-bearing in the case the design actually cares about — a crash loop, a deploy loop, or several processes publishing account-wide, where the newest retained rows are all too close together to measure against (MinGrowthSampleInterval) and the walk has to reach back PAST them. At depth 1 there is nothing to reach back to and the rate stays unknown for exactly as long as the loop lasts.

It is exported as a CROSS-PIN so a catalog editor who reconciles the STORAGE_REPORT row's History below the floor fails a test in graph rather than silently blanking every rate in the account.

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 ErrBackingStreamNotProvisionable = errors.New(
	"stream provisioning governs ordinary streams only; " +
		"KV and ObjectStore backing streams are not provisionable here")

ErrBackingStreamNotProvisionable is returned when a stream-provisioning seam is handed a KV or ObjectStore backing-stream name. Sentinel so a boot path (or a test) can classify the refusal distinctly from a NATS-side create failure.

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.

View Source 
var ErrStreamBoundsUndeclared = errors.New("ordinary stream bounds are not declared")

ErrStreamBoundsUndeclared is the ONE identity for the bounds requirement, whichever seam refuses. `config` shares this value rather than defining its own, so a caller — including a sister repo — can test for the requirement without knowing whether the declarative path or the programmatic one caught it.

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 CheckOrdinaryStreamName 

func CheckOrdinaryStreamName(name, source string) error

CheckOrdinaryStreamName is the fail-closed name guard shared by every stream-provisioning seam. Stream provisioning governs ORDINARY streams — JetStream streams SemStreams creates to carry time-shaped events. NATS also implements every KV bucket and every ObjectStore as a JetStream stream (KVStreamPrefix+<bucket> and ObjectStoreStreamPrefix+<bucket>), which puts those resources within reach of the plain stream API. They are not ordinary streams: KV buckets are acquired through the bucket descriptor catalog's acquisition seam or by the product component that owns them, content stores through the content-store constructor, and both retention contracts belong to graph-retention (ADR-068/073).

The hazard is measured, not theoretical. Handing an EXISTING OBJ_ backing stream to a reconciling provisioner switches on age eviction, flips Discard to delete-oldest, AND replaces the store's real chunk/meta subjects — silently, returning nil. On a get-or-create seam the damage is name-squatting instead: a stream created under a bucket's reserved name carrying a foreign TTL and the wrong subjects, which the bucket's later catalog acquisition then collides with. Both shapes are refused here, at the seam.

The rule is the NAME PREFIX, not descriptor-catalog membership, because every downstream safety net has a hole: ReconcileNoLifecycleRetention clears only MaxAge/MaxBytes and never a discard policy; a descriptor declared retention-unmanaged reconciles nothing at all; and product or sister-repo buckets outside the catalog have neither an acquisition seam nor a pre-start backstop to repair a stamp a provisioner applied. A guard covering only catalog members would leave all three open.

The source argument names where the declaration came from — an operator config key, a component port, or the calling seam — so the operator can find and delete it. This function is pure and I/O-free, so it can run at config validation, before any JetStream call, and before any connection or circuit-breaker check.

func CheckStreamBounds 

func CheckStreamBounds(cfg jetstream.StreamConfig, source string) error

CheckStreamBounds refuses to CREATE an ordinary stream that declares no finite byte and age bounds.

It is pure and I/O-free, so a caller can run it at configuration-validation time rather than discovering the refusal at boot. Backing streams are not its business: a KV or ObjectStore backing stream is refused outright by CheckOrdinaryStreamName, and its retention contract belongs to graph-retention (ADR-068/073), so this returns nil for one rather than demanding bounds that would be the wrong requirement entirely.

source names the caller so an operator can find what to edit.

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 DivergenceLabels 

func DivergenceLabels(divergences []StreamFieldDivergence) []string

DivergenceLabels renders divergences for a log attribute.

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 EnsureFrameworkBucket 

func EnsureFrameworkBucket(ctx context.Context, c *Client, spec BucketSpec) (jetstream.KeyValue, error)

EnsureFrameworkBucket is the OWNER acquisition seam: create-or-open the bucket, reconcile the live bucket to the declared policy (retention per Kind; History to the declared value), verify by re-read, and return the handle. Any failure returns to the caller's Start, which the composition root fails closed (the component-start barrier) — that composition is why no post-start boot sweep exists anymore.

Reconciling at acquisition (rather than at boot only) is what closes the post-boot-cutoff class: a dynamic component add/edit re-acquires its buckets through this seam and reconciles them right there.

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 OpenFrameworkBucket 

func OpenFrameworkBucket(ctx context.Context, c *Client, spec BucketSpec) (jetstream.KeyValue, error)

OpenFrameworkBucket is the READER acquisition seam: bind must-exist. It NEVER creates (a reader that creates is an emitter of divergent configuration — the #714 class) and NEVER reconciles (a reader that "fixes" stream config is the same bug class). An absent bucket yields a classified not-ready error naming the catalog Owner, so the operator reads "wait for / deploy the owner", not "the reader is broken".

func ReconcileNoLifecycleRetention 

func ReconcileNoLifecycleRetention(
	ctx context.Context, js jetstream.JetStream, bucket string, logger *slog.Logger,
) error

ReconcileNoLifecycleRetention is the boot-time D1 guard for a framework-owned KV bucket — ADR-068's "no reference-blind lifecycle retention on state the live graph references" invariant, applied to the derived-KV plane (framework-owned-bucket-guards; #622). It is the KV analogue of the shipped ObjectStore guard (storage/objectstore/retention.go) and runs TWO steps IN ORDER on the backing stream (KV_<bucket>):

  1. Reconcile (strip-and-log). If the backing stream carries a binding MaxAge/MaxBytes (e.g. a foreign 7-day TTL from a process that won the get-or-create race, as in #610/#611, or an out-of-band NATS edit), clear it via UpdateStream and emit a WARN naming the bucket and the removed retention. Stripping stops FUTURE time/size eviction and deletes no stored key, so it self-heals legacy buckets that a create-or-get path would otherwise never reconcile.

  2. Assert (fail-closed). Re-read the backing stream fresh and run the pure CheckNoLifecycleRetention. If retention is STILL binding (the UpdateStream was denied, or a concurrent writer re-set it), return a wrapped fatal ErrGraphBucketRetention so startup fails closed rather than proceeding to silently expire graph state a day later.

The strip trigger and the final assert share ONE predicate (CheckNoLifecycleRetention) with the ObjectStore guard, so KV and ObjectStore can never diverge on what "binding" means.

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 StorageReportKey 

func StorageReportKey(resource string) (string, error)

StorageReportKey is the report bucket key for one resource name.

An ordinary name IS its own key, so `nats kv get` reads naturally and an operator can address a resource by the name they already know. JetStream accepts stream names that are not legal KV keys, though — `$` and `+` among others are fine in a stream name and illegal in a key — and the resources most likely to be a growth problem are exactly the ones nobody in this process created. Dropping them would rebuild the silent omission the inventory exists to end, so an illegal name is addressed through the repository's opaque key codec instead. The row still carries the real name.

A name that is ITSELF a canonical opaque token is also encoded, so no two resources can ever collide on one key.

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 AccountLimitReader 

type AccountLimitReader interface {
	AccountInfo(ctx context.Context) (*jetstream.AccountInfo, error)
}

AccountLimitReader is the narrow JetStream capability the account comparison needs. jetstream.JetStream satisfies it.

It is deliberately SEPARATE from StreamLister rather than folded into it. Account limits are an optional enrichment: a collector that cannot read them still publishes a complete resource inventory with the comparison reported unknown, and widening the listing interface would make every test fake and every future lister implementation carry a method the inventory can do without.

type AccountReport 

type AccountReport struct {
	// CollectedAt is when the account was read. Stamped by the publisher from
	// the inventory this report was derived with, so it always matches the
	// resource rows it was computed alongside.
	CollectedAt time.Time `json:"collected_at"`

	// ProducedBy names the process that produced this report.
	ProducedBy string `json:"produced_by"`

	// Tiers holds file and memory always, and the unknown tier only when some
	// resource could not be filed under either. An always-present unknown row
	// would publish a permanently empty comparison for a limit that does not
	// exist.
	Tiers []TierComparison `json:"tiers"`

	// LimitsUnavailable names why the account limits could not be read. Empty
	// when they were.
	LimitsUnavailable string `json:"limits_unavailable,omitempty"`
}

AccountReport is the account-level half of the storage report: what each storage tier's ceiling is, and whether the declarations filed against it fit.

func DeriveAccountReport 

func DeriveAccountReport(resources []StorageResource, limits AccountTierLimits) AccountReport

DeriveAccountReport compares the inventory's declared bounds against the account limits, WITHIN each tier.

It is derived at collection time, where both inputs are in hand, so the published report carries the verdict rather than leaving every operator surface to recompute it from the rows and risk disagreeing.

func (AccountReport) TierFor 

func (r AccountReport) TierFor(tier StorageTier) (TierComparison, bool)

TierFor returns one tier's comparison. ok is false when the report carries no row for that tier, which is a real answer rather than a zero comparison.

type AccountTierLimits 

type AccountTierLimits struct {
	// Known reports that AccountInfo was read. False means every tier's limit
	// reports unknown — never unbounded.
	Known bool

	// Unavailable names why the limits could not be read. Empty when Known.
	Unavailable string

	// MaxMemory and MaxStore are the raw account ceilings as AccountLimits
	// reports them, including its -1 unlimited sentinel.
	MaxMemory int64
	MaxStore  int64

	// MemoryUsed and StoreUsed are the account's own per-tier usage.
	MemoryUsed int64
	StoreUsed  int64
}

AccountTierLimits is the server's own per-tier accounting: the ceiling for each tier and the usage the server measures against it.

Usage comes from the ACCOUNT rather than from summing the inventory's per-resource usage. The account number is what the limit is actually enforced against, and it is the number `nats account info` shows; a sum over resources drifts from it (replication, per-stream overhead) and would produce a headroom an operator cannot reconcile with the server.

func AccountTierLimitsFrom 

func AccountTierLimitsFrom(info *jetstream.AccountInfo) AccountTierLimits

AccountTierLimitsFrom reads one AccountInfo response. A nil response is UNKNOWN rather than a zero-valued account: a zeroed limit set would classify as an account bounded at zero bytes and report every tier permanently over-committed.

func UnknownAccountTierLimits 

func UnknownAccountTierLimits(reason string) AccountTierLimits

UnknownAccountTierLimits is the explicit unreadable account, naming why.

type AttributionState 

type AttributionState string

AttributionState discriminates why a resource does or does not name an owner.

Unattributed and not-applicable are DIFFERENT facts and only one of them is a finding. Logical ownership is defined for KV buckets, through the descriptor catalog; there is no owner registry for ordinary streams or ObjectStores, and the inventory enumerates the ACCOUNT, so a resource another process declared has no declaration this process could read. Reporting both as an empty owner would say "the framework has no owner concept here" and "this bucket escaped the catalog" in the same breath. 

const (
	// AttributionAttributed means the descriptor catalog declares an owner for
	// this resource's bucket, and Owner carries it.
	AttributionAttributed AttributionState = "attributed"
	// AttributionUnattributed means the resource IS a KV bucket, so an owner is
	// meaningful, but the catalog declares none. This is the finding state: a
	// bucket outside framework ownership, reported rather than omitted or
	// force-fit.
	AttributionUnattributed AttributionState = "unattributed"
	// AttributionNotApplicable means ownership is not defined for this kind of
	// resource at all. Not a finding.
	AttributionNotApplicable AttributionState = "not-applicable"
)

Attribution states.

type BucketClass 

type BucketClass string

BucketClass classifies what kind of state a framework bucket holds. It is descriptive (operator/reviewer-facing); no enforcement derives from it. 

const (
	// ClassAuthoritative is canonical domain state (ENTITY_STATES).
	ClassAuthoritative BucketClass = "authoritative"
	// ClassDerived is state rebuilt from authoritative state (the indexes).
	ClassDerived BucketClass = "derived"
	// ClassOperational is framework-internal correctness state (redelivery
	// stamps, readiness envelopes, ownership epochs).
	ClassOperational BucketClass = "operational"
	// ClassDiagnostic is observability-only state nothing in production reads.
	ClassDiagnostic BucketClass = "diagnostic"
)

Bucket classes.

type BucketSpec 

type BucketSpec struct {
	Name        string
	Owner       string // the component/subsystem that provisions and writes it
	Description string // stamped on the bucket so `nats kv ls` self-describes

	Class     BucketClass
	Retention RetentionPolicy
	Write     WritePolicy
	Posture   CreatePosture

	History  uint8 // KV history depth (stream MaxMsgsPerSubject); 0 = 1
	Replicas int
}

BucketSpec is one framework bucket's full declaration: identity, ownership, policy, and bucket configuration. The catalog (graph.KVCatalog) is a slice of these; every enforcement set the framework uses is a derived view.

func (BucketSpec) Validate 

func (s BucketSpec) Validate() error

Validate fails closed on a descriptor this binary cannot enforce. EVERY discriminated field is checked against its explicit allowed arms with a fail-closed default: an unknown RetentionKind (a newer catalog on an older binary) must be an invalid-policy error, never a silently unapplied policy — and an empty or typoed Write policy must never validate, because the derived owned-bucket write guard filters on it and a zero value would silently drop the bucket OUT of the guard set (fail-open in the exact foundation meant to replace review-maintained integrity). Class and Posture get the same treatment: a descriptor grammar that tolerates zero values is a hand list with extra steps.

type Capacity 

type Capacity struct {
	State CapacityState `json:"state"`

	// ConfiguredLimit is the declared finite bound. Non-nil ONLY when State is
	// CapacityBounded.
	ConfiguredLimit *int64 `json:"configured_limit,omitempty"`

	// Used is observed usage. Nil when State is CapacityUnknown, because an
	// unreadable usage must not read back as zero.
	Used *int64 `json:"used,omitempty"`
}

Capacity is one bounded dimension of a resource — a configured limit paired with observed usage.

The numeric fields are pointers on purpose. An absent value must be ABSENT, never a zero that a downstream reader can mistake for a real measurement: "limit 0" is the exact shape that turns an unreadable resource into a reported-healthy one. Read them through Limit and Usage, which return an ok flag, and build them through NewCapacity, which is the only classifier.

func NewCapacity 

func NewCapacity(limit, used int64, known bool) Capacity

NewCapacity classifies one capacity axis. It is the ONE place a limit/usage pair becomes a state, so no caller can invent a fourth interpretation.

known reports whether the resource's configuration AND state were readable. They arrive together in a single listing entry today; a future collector that reads them from separate calls MUST pass false when either fails, because the requirement is "limit OR usage cannot be determined".

A non-positive limit is JetStream's unlimited encoding (both 0 and the -1 sentinel appear in practice) and yields CapacityUnbounded — never a bounded zero.

func UnknownCapacity 

func UnknownCapacity() Capacity

UnknownCapacity is the explicit unreadable capacity.

func (Capacity) Bounded 

func (c Capacity) Bounded() bool

Bounded reports whether headroom and time-to-threshold are projectable for this axis at all. Both an unbounded and an unknown capacity answer false, for different reasons the State field keeps distinct.

func (Capacity) Limit 

func (c Capacity) Limit() (int64, bool)

Limit returns the configured finite bound. ok is false unless the capacity is bounded, so an unbounded or unknown resource can never yield a limit number.

func (Capacity) Usage 

func (c Capacity) Usage() (int64, bool)

Usage returns observed usage. ok is false when the capacity is unknown.

type CapacityState 

type CapacityState string

CapacityState discriminates the three — not two — states a capacity axis can be in. Collapsing any pair produces the phantom-signal class this capability exists to remove: an unreadable limit is not an absent limit, and an absent limit is not a safe one. 

const (
	// CapacityBounded means a finite configured limit was read. Only this state
	// carries a limit value, and only this state supports headroom or
	// time-to-threshold projection.
	CapacityBounded CapacityState = "bounded"
	// CapacityUnbounded means the resource is deliberately unlimited. Usage is
	// still observable; headroom is not, because there is no bound to have
	// headroom against.
	CapacityUnbounded CapacityState = "unbounded"
	// CapacityUnknown means the limit or the usage could not be determined.
	// It is NOT unlimited, NOT zero, and NOT healthy, and it suppresses
	// projection rather than emitting a fabricated one.
	CapacityUnknown CapacityState = "unknown"
)

Capacity states.

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) AccountStreamLister 

func (m *Client) AccountStreamLister() (StreamLister, error)

AccountStreamLister resolves this client's live JetStream context as the narrow account-listing capability the storage inventory needs. Pass the method value itself as a StreamListerSource so resolution happens per collection.

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) OutstandingWork 

func (m *Client) OutstandingWork(
	ctx context.Context, streamName, consumerName string,
) (uint64, error)

OutstandingWork returns the bound consumer's TOTAL outstanding messages — undelivered (pending) plus delivered-but-unacknowledged.

It returns one number because the total is the only sound one. The sum is invariant to which of the two underlying counters currently holds a message and neither half is: a message moves between them continuously — delivered (pending -> ack-pending), negatively acknowledged back (ack-pending -> pending), redelivered again — so either counter read alone oscillates while real outstanding work is steady. Only the total is monotone with respect to work actually outstanding. Reading pending alone would additionally under-report by the whole in-process lane queue, which is delivered-but-unacked. The halves are therefore summed HERE rather than returned, so no caller can gate on one of them.

ZERO MEANS NO OUTSTANDING WORK, NOT THAT EVERY MESSAGE WAS APPLIED. A message that exhausts MaxDeliver is parked and leaves the count entirely, so it is invisible here. Callers must not read zero as evidence of completeness or use it to license an authoritative-absence claim (gh#742 owns operator visibility for parked messages).

This deliberately returns a count rather than the jetstream.Consumer handle. Handing back the handle would leak consume/fetch capability to callers that need only a number, and would put Info().AckFloor one field away from every holder — and the ack floor is MEASURED-UNUSABLE for catch-up: against both deployed server versions it does not advance past a MaxDeliver-exhausted message, then advances past it on an unrelated later acknowledgement, so it reads not-caught-up while idle and falsely-covered under traffic. Keeping the handle private seals "never floor-derived" at the seam instead of leaving it to review vigilance.

It reads the UNCONDITIONAL consumer bookkeeping, not the metrics registry. The only other retained handle lives in jetstreamMetrics.consumers, which is nil unless WithMetrics was given a non-nil registry — a readiness producer sourced from there would silently degrade in any deployment running without metrics, which is the phantom-signal class this repo hunts.

An unbound consumer is an ERROR, never (0, nil): unknown backlog must not be representable as empty backlog. Mapping the error to a degraded readiness state is caller policy.

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 WithRequestHandlerTimeout 

func WithRequestHandlerTimeout(d time.Duration) ClientOption

WithRequestHandlerTimeout sets the per-message timeout applied to a SubscribeForRequests handler invocation. Zero or negative leaves the current value (env-resolved default, DefaultRequestHandlerTimeout=30s) untouched. An explicit option wins over the SEMSTREAMS_NATS_REQUEST_HANDLER_TIMEOUT env var. Raise it only for deployments running slow-by-design handlers (e.g. LLM answer synthesis).

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 CreatePosture 

type CreatePosture string

CreatePosture declares how a bucket comes into existence. 

const (
	// PostureOwnerCreates: the declared owner provisions the bucket through
	// EnsureFrameworkBucket; readers must-exist via OpenFrameworkBucket.
	PostureOwnerCreates CreatePosture = "owner-creates"
	// PostureReaderMustExist: no in-process owner provisions it; every binding
	// is must-exist. No shipped descriptor uses this posture today — it exists
	// because the catalog grammar declares both arms.
	PostureReaderMustExist CreatePosture = "reader-must-exist"
)

Create postures.

type Growth 

type Growth struct {
	State GrowthState `json:"state"`

	// BytesPerSecond is the measured rate. Non-nil ONLY when State is
	// GrowthKnown. Negative means the resource is shrinking.
	BytesPerSecond *float64 `json:"bytes_per_second,omitempty"`

	// ObservedOver is the interval between the two observations the rate was
	// measured across. Published so a consumer can weigh a rate measured over
	// one minute against one measured over a day; a rate with no interval is
	// not interpretable.
	ObservedOver time.Duration `json:"observed_over_ns,omitempty"`

	// ObservedFrom is when the baseline observation was taken. A POINTER, like
	// the numeric absences: `omitempty` does nothing for a time.Time, so a
	// value field would publish "observed_from":"0001-01-01T00:00:00Z" on every
	// unknown rate, and a consumer could read that zero as a real baseline.
	ObservedFrom *time.Time `json:"observed_from,omitempty"`

	// Unavailable names why the rate is unknown. Empty when State is
	// GrowthKnown.
	Unavailable string `json:"unavailable,omitempty"`
}

Growth is a resource's observed rate of change.

BytesPerSecond is a pointer for the same reason the capacity numbers are: an absent rate must be ABSENT, never a zero a downstream reader can mistake for "measured, and not growing".

func DeriveGrowth 

func DeriveGrowth(current Observation, prior []Observation) Growth

DeriveGrowth measures the rate between the current observation and the most recent usable prior one.

prior may be in any order and may contain observations that are unusable (from the future, by clock skew across producers, or too close to the current one to measure). The NEWEST usable one is the baseline: capacity planning needs the current rate, and averaging in an old sample dilutes exactly the recent acceleration an operator needs to see.

func UnknownGrowth 

func UnknownGrowth(reason string) Growth

UnknownGrowth is the explicit unmeasured rate, naming why.

func (Growth) Rate 

func (g Growth) Rate() (float64, bool)

Rate returns the measured rate in bytes per second. ok is false unless the rate is known, so an unknown rate can never yield a number.

type GrowthState 

type GrowthState string

GrowthState discriminates a known rate from an absent one. There is no third state: a rate is either measured across two observations or it is unknown. 

const (
	// GrowthKnown means a rate was measured across two observations. Zero and
	// negative rates are KNOWN — a stable resource and a shrinking one are
	// facts, not absences.
	GrowthKnown GrowthState = "known"
	// GrowthUnknown means no usable pair of observations exists. It suppresses
	// time-to-threshold rather than being treated as zero growth, because "not
	// growing" and "not measured" license different operator decisions.
	GrowthUnknown GrowthState = "unknown"
)

Growth states.

type InventoryPublisher 

type InventoryPublisher interface {
	Publish(ctx context.Context, inv StorageInventory) (PublishResult, error)
}

InventoryPublisher publishes one collection's inventory. *StorageReportPublisher satisfies it; the interface keeps the collector from depending on the report's whole surface.

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.

Use UpdateWithRetryRev when the caller must ATTRIBUTE the write to itself: this variant discards the committed revision, and re-reading it afterwards is not equivalent — another writer can commit in between, and the re-read then returns that writer's revision.

func (*KVStore) UpdateWithRetryRev 

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

UpdateWithRetryRev is UpdateWithRetry that also returns the EXACT KV revision the committed write produced.

This exists because a post-hoc `Get` is not a substitute. Between the CAS and the re-read, another writer can commit; the re-read then reports THAT writer's revision, and any caller that attributes the reported revision to its own write is now holding someone else's. The rule engine's per-rule feedback-loop tracker does exactly that attribution, and `shouldSkipRule` consumes the recorded revision once and returns true — so a mis-attributed revision makes the rule silently DROP the external writer's genuine change.

Note this is a strictly different safety property from the readiness consumer's `IndexedRevision >= myRev` check, which tolerates over-reporting because revisions are monotonic. Two consumers, two properties; do not generalize from the tolerant one.

On any non-nil error the returned revision is 0 — nothing committed.

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 Observation 

type Observation struct {
	At    time.Time `json:"at"`
	Bytes int64     `json:"bytes"`
}

Observation is one measurement of a resource's size at one moment. At is when the size was READ from the account, not when the row was written: the write happens later by the publication latency, and using it would inflate every interval by that latency.

type OvercommitmentState 

type OvercommitmentState string

OvercommitmentState is the verdict on one tier's declared-versus-limit comparison. 

const (
	// OvercommitmentWithin means the declared bounds in this tier fit inside
	// the account limit for the tier. It says nothing about the tier's
	// unbounded resources, whose declarations are absent from the sum by
	// construction — read UnboundedResources alongside it.
	OvercommitmentWithin OvercommitmentState = "within-limit"
	// OvercommitmentOver means the declared bounds in this tier exceed the
	// account limit for the tier. Every resource can still be within its own
	// bound: this is the failure mode a per-resource view cannot see.
	OvercommitmentOver OvercommitmentState = "over-committed"
	// OvercommitmentNotApplicable means there is no comparison to make,
	// because the account limit for this tier is unbounded or unreadable. It
	// is NOT a passing verdict, and Unavailable says which case it is.
	OvercommitmentNotApplicable OvercommitmentState = "not-applicable"
)

Over-commitment states.

type OwnerResolver 

type OwnerResolver func(bucket string) string

OwnerResolver resolves a KV bucket name to its declared logical owner, returning "" for a bucket it does not declare.

This is a FUNCTION, not a map, and that is the point: attribution must be a read of the one descriptor catalog at collection time. graph.OwnerOf satisfies it directly. A retained copy could disagree with the acquisition seam about who owns a bucket, and would keep reporting a former owner after the catalog dropped the row.

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 Pressure 

type Pressure struct {
	// Evaluated reports whether a state could be derived at all.
	Evaluated bool `json:"evaluated"`

	// State is the worse of the two bands. Empty when Evaluated is false.
	State PressureState `json:"state,omitempty"`

	// RaisedBy names the input that produced State.
	RaisedBy PressureInput `json:"raised_by,omitempty"`

	// FromHeadroom and FromTimeToThreshold are the individual bands, published
	// so a consumer can see the input that did NOT win.
	// FromTimeToThreshold is empty when no projection was available.
	//
	// They describe whichever ceiling EvaluatedAgainst names. On a row whose basis
	// is the account tier these are the TIER's bands, not this resource's — the
	// resource has no bound of its own to produce any — so they will not agree with
	// the row's own Projection, which reports headroom unavailable. Read the basis
	// first; the numbers behind these bands are on the account row.
	FromHeadroom        PressureState `json:"from_headroom,omitempty"`
	FromTimeToThreshold PressureState `json:"from_time_to_threshold,omitempty"`

	// EvaluatedAgainst names the ceiling State was derived from. Set whenever
	// Evaluated, so no consumer has to infer the basis from the resource's
	// capacity state.
	EvaluatedAgainst PressureBasis `json:"evaluated_against,omitempty"`

	// Unavailable names why no state was derived. Empty when Evaluated.
	Unavailable string `json:"unavailable,omitempty"`
}

Pressure is one resource's derived pressure state.

Evaluated is explicit rather than inferred from an empty State, so a consumer cannot read "no state" as normal. A resource whose capacity is unknown or unbounded has NO pressure state at all — reporting normal for it would be the phantom-signal class this capability exists to remove.

func AssessPressure 

func AssessPressure(
	capacity Capacity, projection Projection, thresholds ResolvedPressureThresholds,
) Pressure

AssessPressure derives the pressure state from a resource's projection.

It returns a value and touches nothing else: no write is rejected, no component is throttled, no readiness gate is failed, and no retention is applied as a consequence of what it returns.

func PressureAgainstAccountTier 

func PressureAgainstAccountTier(tier TierComparison, found bool) Pressure

PressureAgainstAccountTier evaluates an UNBOUNDED resource against the only ceiling it has: the account limit of the storage tier it lives in.

This exists so that declaring a stream archival cannot remove it from the surface that would warn about it. Capacity matters MORE for a resource that can never evict — capacity is then the only lever an operator has — so reporting it permanently unevaluable would put the least recoverable resources in the account behind the one field every alert rule keys on.

The state is the TIER's, unmodified and deliberately so. The ceiling is shared, the usage measured against it is the account's own, and the rate is the tier's: every resource sharing a filling tier is genuinely in the same trouble. Scaling the tier's verdict by this resource's share would be a fabrication — it would report a stream as calm because it is individually small, while the ceiling that governs it fills from elsewhere.

found reports whether the resource could be filed under a tier at all. It is a parameter rather than a zero-value check because an absent tier row and a tier row with no ceiling are different findings.

func (Pressure) AgainstAccountTier 

func (p Pressure) AgainstAccountTier() Pressure

AgainstAccountTier restates an evaluated pressure as having come from an account tier ceiling rather than a resource's own bound.

It only relabels the basis. The bands are whatever the comparison produced, and an unevaluated pressure passes through untouched: there is no state to attribute to any ceiling.

type PressureBasis 

type PressureBasis string

PressureBasis names WHICH CEILING a pressure state was evaluated against.

It exists because this capability now evaluates two different ceilings and an operator acts on them differently. A resource at high pressure against its own declared bound is fixed by raising that bound or letting retention do its job. A resource at high pressure against its ACCOUNT TIER has no bound of its own to raise: the tier is filling, the finding is about every resource sharing it, and the levers are account capacity or deleting data. Publishing the state without the basis would leave those indistinguishable in the one field every alert rule reads. 

const (
	// PressureBasisOwnBound means the state came from the resource's own
	// declared limit.
	PressureBasisOwnBound PressureBasis = "own-bound"
	// PressureBasisAccountTier means the state came from the account limit of
	// the storage tier the resource lives in. For an unbounded resource that is
	// the only ceiling there is; for a tier row it is the tier's own limit.
	PressureBasisAccountTier PressureBasis = "account-tier"
)

Pressure bases.

type PressureInput 

type PressureInput string

PressureInput names which input raised a pressure state, so an operator can tell a capacity problem from a rate problem. 

const (
	// PressureInputNone means nothing raised the state above normal.
	PressureInputNone PressureInput = "none"
	// PressureInputHeadroom means proportional headroom raised it.
	PressureInputHeadroom PressureInput = "headroom"
	// PressureInputTimeToThreshold means the projection raised it — the case
	// proportional headroom alone would have missed.
	PressureInputTimeToThreshold PressureInput = "time-to-threshold"
	// PressureInputBoth means both inputs landed on the same band. Reported
	// rather than tie-broken: an arbitrary winner would misreport the cause.
	PressureInputBoth PressureInput = "both"
)

Pressure inputs.

type PressureState 

type PressureState string

PressureState is how worried an operator should be about one resource. 

const (
	// PressureNormal means neither input is inside a configured band.
	PressureNormal PressureState = "normal"
	// PressureWarning means an input crossed the widest band.
	PressureWarning PressureState = "warning"
	// PressureHigh means an input crossed the middle band.
	PressureHigh PressureState = "high"
	// PressureCritical means an input crossed the tightest band. It still
	// rejects nothing: pressure is report-only in this capability.
	PressureCritical PressureState = "critical"
)

Pressure states, worst last.

type Projection 

type Projection struct {
	// HeadroomBytes is the distance from current usage to the configured
	// bound. It can be NEGATIVE for a resource already over its bound, which is
	// reported rather than clamped.
	HeadroomBytes *int64 `json:"headroom_bytes,omitempty"`

	// HeadroomFraction is HeadroomBytes as a fraction of the bound.
	HeadroomFraction *float64 `json:"headroom_fraction,omitempty"`

	// HeadroomUnavailable names why headroom is absent. Empty when present.
	HeadroomUnavailable string `json:"headroom_unavailable,omitempty"`

	// ThresholdBytes is the usage level the time projection targets: the
	// critical-headroom level, not the bound itself. Capacity has to be
	// corrected BEFORE the last byte fits.
	ThresholdBytes *int64 `json:"threshold_bytes,omitempty"`

	// TimeToThreshold is the projected time until usage reaches
	// ThresholdBytes at the observed rate. Zero means the resource is already
	// there; it is never negative.
	TimeToThreshold *time.Duration `json:"time_to_threshold_ns,omitempty"`

	// TimeToThresholdUnavailable names why the time projection is absent.
	TimeToThresholdUnavailable string `json:"time_to_threshold_unavailable,omitempty"`
}

Projection is what the report can say about a resource's future.

Headroom and time-to-threshold are suppressed INDEPENDENTLY, each with its own reason: headroom needs a readable bound, while a time projection additionally needs a measured, positive rate. Suppressing them together would discard the half that is knowable.

The numeric fields are pointers so an absent projection is ABSENT rather than a zero that reads as "no headroom left" or "exhausted now".

func Project 

func Project(capacity Capacity, growth Growth, thresholds ResolvedPressureThresholds) Projection

Project derives headroom and time-to-threshold for one resource.

type PublishResult 

type PublishResult struct {
	Published int
	Deleted   int

	// AccountPublished reports that the per-tier account row was written. It is
	// a separate field rather than part of Published so that count keeps meaning
	// "resources", which is what every caller logging it reads it as.
	AccountPublished bool

	// Skipped reports that nothing was written at all, with SkipReason saying
	// why. It is not an error: declining to republish a stale inventory is the
	// correct behavior, not a failure.
	Skipped    bool
	SkipReason string
}

PublishResult is what one publication did, for the caller that logs it.

It deliberately does NOT carry the published rows. Every operator-facing surface must be a CONSUMER of the bucket rather than a second report-producing path, and handing an in-process caller the derived values directly is exactly the shortcut that would let one surface disagree with another.

type ReportStore 

type ReportStore interface {
	Put(ctx context.Context, key string, value []byte) (uint64, error)
	Delete(ctx context.Context, key string, opts ...jetstream.KVDeleteOpt) error
	History(ctx context.Context, key string, opts ...jetstream.WatchOpt) ([]jetstream.KeyValueEntry, error)
	ListKeys(ctx context.Context, opts ...jetstream.WatchOpt) (jetstream.KeyLister, error)
}

ReportStore is the narrow KV capability the report publisher needs. jetstream.KeyValue satisfies it.

History rather than Get: the seed read has to be able to walk PAST an observation too close to the current one to measure (several processes may publish account-wide, so the newest revisions can all land inside one collection interval), and the retained series is what makes that possible.

type ReportWatchStore 

type ReportWatchStore interface {
	WatchAll(ctx context.Context, opts ...jetstream.WatchOpt) (jetstream.KeyWatcher, error)
}

ReportWatchStore is the narrow KV capability the report consumer needs. jetstream.KeyValue satisfies it.

WatchAll rather than a range read: on restart a consumer needs the current report immediately (KV re-delivers every current value, which is correct recovery), and several surfaces should each react rather than one dequeuing. Deletes are NOT filtered out — a reclaimed row must retract its metric series rather than leave a resource reporting forever after it is gone.

type ResolvedPressureThresholds 

type ResolvedPressureThresholds struct {
	WarningHeadroom  float64
	HighHeadroom     float64
	CriticalHeadroom float64

	WarningHorizon  time.Duration
	HighHorizon     time.Duration
	CriticalHorizon time.Duration
}

ResolvedPressureThresholds is the applied form: defaults filled, durations parsed, ordering validated. Only Resolve produces one, so no evaluation can run against a threshold set nobody checked.

type ResourceKind 

type ResourceKind string

ResourceKind classifies a physical JetStream stream name by the reserved backing-stream namespace it occupies. It is the naming convention read two ways: the provisioning guard refuses everything that is not an ordinary stream, and the storage inventory reports every kind. 

const (
	// ResourceOrdinaryStream is a stream SemStreams provisions to carry
	// time-shaped events — the only kind stream provisioning governs.
	ResourceOrdinaryStream ResourceKind = "stream"
	// ResourceKeyValue is a KV bucket's backing stream (KVStreamPrefix).
	ResourceKeyValue ResourceKind = "kv"
	// ResourceObjectStore is an ObjectStore's backing stream
	// (ObjectStoreStreamPrefix).
	ResourceObjectStore ResourceKind = "objectstore"
)

Resource kinds.

func ClassifyBackingStream 

func ClassifyBackingStream(stream string) (ResourceKind, string)

ClassifyBackingStream maps a physical JetStream stream name to the logical resource it backs: the kind, and for a backing stream the bucket name behind it (empty for an ordinary stream, which has no bucket).

EXACTLY ONE leading prefix is stripped. A product bucket may legitimately be named "KV_FOO", whose backing stream is then "KV_KV_FOO" and whose real name is "KV_FOO" — not "FOO". Stripping greedily would attribute that resource to a different bucket entirely, which is why the recovery lives here, once, and is read by both the provisioning guard's diagnostic and the inventory's owner attribution: the two can disagree about a bucket's name only if they compute it separately, so they do not.

The rule is the prefix and nothing else — no catalog lookup, no membership test — so a product or sister-repo bucket outside the framework catalog classifies identically to a framework one.

type ResourceReport 

type ResourceReport struct {
	// Resource is the inventory row verbatim — name, kind, attribution, owner,
	// tier, and the capacity states. The report is a VIEW of the inventory, not
	// a second opinion about any of it.
	Resource StorageResource `json:"resource"`

	// CollectedAt is when this resource was read from the account. Every key
	// carries its own, because a consumer ranging the bucket sees a mix of
	// revisions rather than an atomic snapshot and needs to see the spread.
	CollectedAt time.Time `json:"collected_at"`

	// ProducedBy names the process that collected and published this row. A
	// fleet of processes each polling account-wide is unreconcilable without
	// it.
	ProducedBy string `json:"produced_by"`

	// Growth is the observed rate of change across successive observations.
	Growth Growth `json:"growth"`

	// Projection is headroom and time-to-threshold, each suppressed on its own
	// terms rather than fabricated.
	Projection Projection `json:"projection"`

	// Pressure is the derived state. Report-only: no write is rejected, no
	// component is throttled, no readiness gate is failed, and no retention is
	// applied because of what it says.
	Pressure Pressure `json:"pressure"`
}

ResourceReport is one published row: what the collector saw about one resource, and what it derived from it.

The row is plain JSON with no BaseMessage wrapper, following the GRAPH_STATUS readiness envelope: the payload registry governs polymorphic publishes on subjects, where a receiver must discriminate a type it did not choose. This is a KV value on a key whose type is fixed by the contract, and wrapping it would break every consumer's plain decode for nothing.

type RetentionKind 

type RetentionKind string

RetentionKind discriminates a bucket's declared retention policy. Every switch over it MUST carry a default arm that fails closed: a Kind this binary does not know (a newer catalog on an older binary) is an invalid policy, never a silent no-op. 

const (
	// RetentionNoLifecycle: no NATS TTL (MaxAge) or binding MaxBytes, ever —
	// correctness-critical no-eviction state (ADR-068). Acquisition strips a
	// foreign retention in place or fails closed.
	RetentionNoLifecycle RetentionKind = "no-lifecycle"
	// RetentionBoundedTTL: the declared TTL IS the contract (e.g.
	// OWNER_PRESENCE liveness). Acquisition converges MaxAge TO the declared
	// TTL — preserved, never stripped.
	RetentionBoundedTTL RetentionKind = "bounded-ttl"
	// RetentionUnmanaged: the framework guarantees no retention posture — an
	// explicit catalog fact (COMPONENT_STATUS), not an omission. Acquisition
	// does not reconcile retention.
	RetentionUnmanaged RetentionKind = "unmanaged"
)

Retention kinds. All three are populated by shipped catalog rows; adding a future kind (e.g. bounded-storage's DiscardNew ceiling) is a new constant + params on RetentionPolicy + a reconcile arm — no shape change.

type RetentionPolicy 

type RetentionPolicy struct {
	Kind RetentionKind
	// TTL is the declared bucket TTL. Meaningful (and required non-zero) for
	// RetentionBoundedTTL only.
	TTL time.Duration
}

RetentionPolicy is a discriminated Kind plus its parameters — a struct, not a bare enum, so a future kind adds params without reshaping every consumer.

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 StorageInventory 

type StorageInventory struct {
	// ProducedBy names the process that collected this inventory. Never empty,
	// because a fleet of processes each polling account-wide produces reports
	// that cannot be reconciled without it.
	ProducedBy string `json:"produced_by"`

	// CollectedAt is when Resources were read from the account. Zero when no
	// collection has ever succeeded.
	CollectedAt time.Time `json:"collected_at"`

	// Resources is the complete account listing, deduplicated and sorted by
	// physical name. It is never a partial listing: a walk that fails part way
	// through leaves the previous result in place rather than reporting a
	// subset of the account as if it were all of it.
	Resources []StorageResource `json:"resources"`

	// Account is the per-tier declared-versus-account-limit comparison over
	// exactly these Resources. It travels WITH them rather than being computed
	// by each consumer, so no operator surface can disagree with another about
	// whether a tier is over-committed. An account whose limits could not be
	// read still yields a report — with every tier's limit unknown, which is
	// distinct from unbounded and from satisfied.
	Account AccountReport `json:"account"`

	// Stale reports that the most recent collection attempt did not succeed, so
	// Resources and CollectedAt describe an earlier moment. It is also true
	// before the first successful collection, when Resources is legitimately
	// empty but the account is NOT known to be empty.
	Stale bool `json:"stale"`

	// StaleSince is when the most recent failed attempt happened. A POINTER for
	// the same reason the capacity numbers are: `omitempty` is a no-op on a
	// time.Time, so a value field would publish a zero timestamp on a healthy
	// inventory and invite a consumer to read it as a real failure time.
	StaleSince *time.Time `json:"stale_since,omitempty"`

	// StaleReason explains the failure in operator terms.
	StaleReason string `json:"stale_reason,omitempty"`
}

StorageInventory is one account-wide collection result.

CollectedAt is when the RESOURCES were read, not when the report was rendered: a degraded inventory keeps the timestamp its data actually came from, so an operator reading a stale report can tell how stale it is.

type StorageInventoryCollector 

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

StorageInventoryCollector enumerates account storage on an interval and publishes the last good result.

Collect performs all of its I/O outside the publication lock and takes the write lock only to swap the finished snapshot, so Latest never waits behind a collection. Nothing here belongs on a component's Start or health path.

func NewStorageInventoryCollector 

func NewStorageInventoryCollector(
	source StreamListerSource,
	cfg StorageInventoryConfig,
) (*StorageInventoryCollector, error)

NewStorageInventoryCollector builds a collector. It fails closed on a missing lister source or owner resolver rather than degrading to a silently unattributed inventory.

func (*StorageInventoryCollector) Collect 

func (c *StorageInventoryCollector) Collect(parent context.Context) (StorageInventory, error)

Collect enumerates the account once, bounded by the configured timeout, and publishes the result. Collections are serialized, so two overlapping callers cannot publish out of order and walk CollectedAt backwards.

On failure it returns the last good inventory marked stale ALONGSIDE the error, so a caller that drops the error still cannot mistake a failed collection for an empty account. A failure caused by the CALLER's context ending — a graceful shutdown — leaves the last good result unmarked, because "context canceled" is not a storage finding.

func (*StorageInventoryCollector) Latest 

func (c *StorageInventoryCollector) Latest() StorageInventory

Latest returns the most recent inventory without doing any I/O. It never blocks on a collection and never fails, so a health check, a readiness evaluation, or an operator report can call it freely.

The returned Resources slice is a COPY. The published slice is built with spare capacity, so handing out the same backing array would let two callers appending to their own results write the same slots.

func (*StorageInventoryCollector) Run 

func (c *StorageInventoryCollector) Run(ctx context.Context)

Run collects on the configured interval until ctx ends. Call it in its own goroutine — never from a component's Start or from health evaluation.

The first collection happens immediately rather than one interval later, so a freshly booted process has a report to serve; because Run is already asynchronous, that cannot delay anything.

type StorageInventoryConfig 

type StorageInventoryConfig struct {
	// Interval is how often Run enumerates the account. Defaults to
	// DefaultStorageInventoryInterval.
	Interval time.Duration

	// Timeout bounds one collection. Defaults to
	// DefaultStorageInventoryTimeout.
	Timeout time.Duration

	// ProducedBy names this process in the report. Defaults to host/pid.
	ProducedBy string

	// OwnerResolver attributes KV resources. REQUIRED: with no resolver every
	// KV resource would report unattributed, which reads as "nothing is
	// framework-owned" rather than as "attribution was never wired", so the
	// constructor fails closed instead of defaulting.
	OwnerResolver OwnerResolver

	// Publisher writes the report after each SUCCESSFUL collection Run makes.
	// Optional: a collector without one keeps its in-process inventory and
	// publishes nothing, which is what the attribution unit tests want.
	//
	// It hangs off the collector rather than running its own timer because the
	// publication cadence IS the observation cadence — the growth series is
	// Δbytes over Δt across published observations, and a second timer could
	// drift from the interval the operator configured.
	Publisher InventoryPublisher

	// Logger receives collection-failure warnings. Defaults to slog.Default().
	Logger *slog.Logger
}

StorageInventoryConfig configures account storage collection.

type StoragePressureThresholds 

type StoragePressureThresholds struct {
	// WarningHeadroom is the free fraction (0..1) at or below which a resource
	// reports warning.
	WarningHeadroom float64 `` /* 165-byte string literal not displayed */

	// HighHeadroom is the free fraction at or below which a resource reports
	// high.
	HighHeadroom float64 `` /* 159-byte string literal not displayed */

	// CriticalHeadroom is the free fraction at or below which a resource
	// reports critical. It is also the level the time projection targets, so
	// "time to threshold" means "time until this fraction is all that is left".
	CriticalHeadroom float64 `` /* 210-byte string literal not displayed */

	// WarningHorizon is the projected time-to-threshold at or below which a
	// resource reports warning.
	WarningHorizon string `` /* 177-byte string literal not displayed */

	// HighHorizon is the projected time-to-threshold at or below which a
	// resource reports high.
	HighHorizon string `` /* 171-byte string literal not displayed */

	// CriticalHorizon is the projected time-to-threshold at or below which a
	// resource reports critical.
	CriticalHorizon string `` /* 178-byte string literal not displayed */
}

StoragePressureThresholds is the OPERATOR-FACING pressure configuration.

Durations are strings ("72h", "90m") following this repository's operator configuration convention, parsed by Resolve rather than at decode time, so a malformed edit is a reported configuration error rather than a decode failure that takes a component down.

Every field is optional and an omitted field takes its documented default. An out-of-range or incoherent field is an ERROR, never a silent fallback: a silently defaulted threshold applies a number the operator did not choose and is indistinguishable from a working edit.

func DefaultStoragePressureThresholds 

func DefaultStoragePressureThresholds() StoragePressureThresholds

DefaultStoragePressureThresholds returns the shipped defaults in their operator-authored form, so a caller that wants them deliberately can say so in one line rather than passing a zero value and hoping.

func (StoragePressureThresholds) Resolve 

func (t StoragePressureThresholds) Resolve() (ResolvedPressureThresholds, error)

Resolve applies defaults, parses the horizon durations, and validates the whole set. It fails on any incoherence rather than repairing it, because a repaired threshold is a number the operator did not choose.

type StorageReportConfig 

type StorageReportConfig struct {
	// Thresholds resolves the pressure thresholds per evaluation. REQUIRED:
	// see ThresholdSource.
	Thresholds ThresholdSource

	// Logger receives publication and threshold warnings. Defaults to
	// slog.Default().
	Logger *slog.Logger
}

StorageReportConfig configures the report publisher.

type StorageReportConsumer 

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

StorageReportConsumer maintains the in-process view of the published report.

func NewStorageReportConsumer 

func NewStorageReportConsumer(
	store ReportWatchStore, cfg StorageReportConsumerConfig,
) (*StorageReportConsumer, error)

NewStorageReportConsumer builds a consumer over the report bucket.

func (*StorageReportConsumer) Run 

func (c *StorageReportConsumer) Run(ctx context.Context)

Run watches the report bucket until ctx ends, re-establishing the watch when it cannot be created or when it ends.

The retry is not optional resilience. A watch that dies leaves every operator surface frozen on its last values, and a frozen report is indistinguishable from a calm account — which is the exact failure this capability exists to end. Call Run in its own goroutine.

func (*StorageReportConsumer) Snapshot 

func (c *StorageReportConsumer) Snapshot() StorageReportSnapshot

Snapshot returns the current view without doing any I/O. Safe to call from a health check, an HTTP handler, or a metrics scrape.

type StorageReportConsumerConfig 

type StorageReportConsumerConfig struct {
	// Observer receives each applied change. Optional: a consumer without one
	// still maintains its snapshot.
	Observer StorageReportObserver

	// RetryBackoff is the wait before re-establishing a watch. Defaults to
	// DefaultReportConsumerRetryBackoff.
	RetryBackoff time.Duration

	// Logger receives watch failures. Defaults to slog.Default().
	Logger *slog.Logger
}

StorageReportConsumerConfig configures the report consumer.

type StorageReportObserver 

type StorageReportObserver interface {
	ObserveResource(row ResourceReport)
	ForgetResource(row ResourceReport)
	ObserveAccount(row AccountReport)
}

StorageReportObserver receives each change the consumer applies.

It exists so a metrics surface can update ONE series per event instead of rebuilding every series on every tick. ForgetResource carries the whole last known row rather than a key, because a resource whose name is not a legal KV key is addressed by an opaque token — a bare key cannot say which label set a gauge was registered under.

Implementations are called from the consumer's watch goroutine and must not block on it.

type StorageReportPublisher 

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

StorageReportPublisher writes the account report to its KV bucket.

Safe for concurrent use, though one publication at a time is the intended shape: the baseline map is the growth series' in-process cache and two overlapping publications would race to advance it.

func NewStorageReportPublisher 

func NewStorageReportPublisher(store ReportStore, cfg StorageReportConfig) (*StorageReportPublisher, error)

NewStorageReportPublisher builds a publisher. It fails closed on a missing store or threshold source rather than evaluating every resource in the account against numbers no operator chose.

func (*StorageReportPublisher) Publish 

func (p *StorageReportPublisher) Publish(ctx context.Context, inv StorageInventory) (PublishResult, error)

Publish writes one row per inventoried resource and deletes the key of any resource the inventory no longer names.

A STALE inventory publishes nothing. Last-good is what the collector serves when a collection fails, and republishing it as a fresh observation would inject a duplicate sample into the growth series and move every row's collection timestamp forward onto data that is not new.

A resource that fails to publish does not stop the rest of the report: the failures are joined and returned, naming the resources they lost, so a partial publication is REPORTED rather than silently shrinking the report.

type StorageReportSnapshot 

type StorageReportSnapshot struct {
	// Resources is every published resource row, sorted by resource name.
	Resources []ResourceReport

	// Account is the per-tier comparison, valid only when AccountKnown.
	Account      AccountReport
	AccountKnown bool

	// Synced reports that the watch has delivered every current value at least
	// once. Before it, an empty snapshot means "not read yet" rather than "the
	// account holds nothing" — the same distinction the inventory's Stale flag
	// keeps.
	Synced bool

	// UpdatedAt is when the last change was applied.
	UpdatedAt time.Time

	// PressureCounts tallies the EVALUATED rows by state.
	PressureCounts map[PressureState]int

	// NotEvaluated counts rows carrying no pressure state at all — unbounded or
	// unknown capacity. Counted rather than folded into normal: a surface that
	// filtered on "state != normal" would make exactly the unbounded resources
	// invisible.
	NotEvaluated int

	// WorstPressure is the worst evaluated state, or empty when nothing was
	// evaluated. Empty is NOT normal: an account whose every row declined to
	// evaluate has no pressure verdict, and reporting one would manufacture it.
	WorstPressure PressureState
}

StorageReportSnapshot is the whole published report as one in-process view.

The counts are a TALLY of what the rows say, never a re-evaluation: pressure is derived at publication, and a consumer that recomputed it would be the second producer this design exists to prevent.

type StorageResource 

type StorageResource struct {
	// Name is the physical JetStream stream name. Always present: a listing
	// entry the collector cannot name fails the collection rather than becoming
	// an unlookupable row.
	Name string `json:"name"`

	// Kind is the logical resource the physical name backs. Derived from the
	// name alone, so it is available even for a resource the server declines to
	// describe.
	Kind ResourceKind `json:"kind"`

	// Bucket is the logical bucket name behind a backing stream, with exactly
	// one reserved prefix stripped. Empty for an ordinary stream.
	Bucket string `json:"bucket,omitempty"`

	// Attribution says whether an owner is defined, undeclared, or not a
	// meaningful question for this kind of resource.
	Attribution AttributionState `json:"attribution"`

	// Owner is the logical owner as the descriptor catalog declares it.
	// Non-empty only when Attribution is AttributionAttributed.
	Owner string `json:"owner,omitempty"`

	// Tier is the storage tier this resource's usage counts against.
	Tier StorageTier `json:"tier"`

	// Bytes and Messages are the two capacity axes JetStream bounds.
	Bytes    Capacity `json:"bytes"`
	Messages Capacity `json:"messages"`
}

StorageResource is one account storage resource as the inventory sees it.

func (StorageResource) Attributed 

func (r StorageResource) Attributed() bool

Attributed reports whether the resource resolved to a declared owner. It reads the Attribution state rather than testing Owner for emptiness, so a not-applicable resource is never mistaken for an escaped bucket.

func (StorageResource) Undescribable 

func (r StorageResource) Undescribable() bool

Undescribable reports whether the server declined to describe this resource — it appeared in the account's name listing but the info listing omitted it, which the server does for any stream carrying an offline reason.

Derived rather than stored so a hand-built row cannot claim otherwise: such a resource is exactly the one with no readable tier and no readable capacity on either axis.

type StorageTier 

type StorageTier string

StorageTier is the JetStream storage tier backing a resource. JetStream keeps SEPARATE memory and file account limits and this repository's own streams span both, so the tier is recorded per resource and must never be guessed: summing across tiers produces a number that means nothing. 

const (
	// TierFile is file-backed storage.
	TierFile StorageTier = "file"
	// TierMemory is memory-backed storage.
	TierMemory StorageTier = "memory"
	// TierUnknown is a resource whose tier could not be read, because the
	// server declined to describe it. It is reported rather than defaulted:
	// defaulting would silently file the resource under one account limit's
	// comparison.
	TierUnknown StorageTier = "unknown"
)

Storage tiers.

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. REQUIRED and must be positive:
	// auto-create is stream provisioning, and an ordinary stream declares finite
	// bounds (see CheckStreamBounds). There is no framework default — one would be
	// a bound nobody chose, indistinguishable in the operator surface from one
	// somebody did, which is what the bounds requirement exists to end.
	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. REQUIRED and must be positive, for the
	// same reason as MaxAge: JetStream reads 0 and -1 alike as unlimited, so
	// neither is a declaration.
	MaxBytes int64

	// Discard is what happens at the ceiling: jetstream.DiscardOld evicts the
	// oldest, jetstream.DiscardNew refuses the write.
	//
	// It exists here so an auto-create path can carry a declaration's discard
	// policy. Without it this struct could express a bound but not what happens
	// when the bound is reached, so a caller recreating a stream from an operator's
	// declaration silently substituted DiscardOld — the zero value — for whatever
	// they chose. It cannot be REQUIRED (DiscardOld being the zero value is exactly
	// why), but it can at least be expressible.
	Discard jetstream.DiscardPolicy

	// 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 the auto-create defaults for the fields that HAVE defaults: storage tier, retention policy and replica count.

It deliberately declares NO bounds. It used to return MaxAge 7 days with MaxBytes unset — the exact silent framework default the bounds requirement removed from the configuration path, still handing out a retention window nobody chose and no size ceiling at all. A caller that auto-creates must state its own bounds; CheckStreamBounds refuses the creation otherwise.

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 StreamFieldDivergence 

type StreamFieldDivergence struct {
	// Field is the jetstream.StreamConfig field name, so it can be found in code.
	Field string
	// Declared is what the caller asked for.
	Declared string
	// Observed is what the live stream actually carries.
	Observed string
}

StreamFieldDivergence is one field a caller declared that the live stream does not match.

The values are rendered strings rather than typed values because the whole point is to be read by a person: a duration, a byte count and a subject list have nothing in common except that an operator needs to see both sides.

func DiffDeclaredStream 

func DiffDeclaredStream(declared, observed jetstream.StreamConfig) []StreamFieldDivergence

DiffDeclaredStream reports the DECLARED fields, of those it compares, whose live value differs.

The compared set is the limits and the routing — the fields whose divergence changes what the stream does with messages. It is NOT every field of jetstream.StreamConfig: Description, AllowDirect, DenyDelete, DenyPurge, AllowRollup, Compression, Metadata, SubjectTransform, RePublish, Placement, Mirror, Sources, ConsumerLimits and FirstSeq are not compared, so a caller that declares one of those and binds a stream without it is not told. AllowDirect is the one with real operational weight (a bound stream without it silently refuses direct gets); the rest are descriptive or advanced-topology fields whose divergence does not change delivery. Widen the set rather than assuming it is already complete.

It is exported so a caller CAN do more than have it logged — treating a subject divergence as fatal, say, since a stream that does not capture the caller's subject makes every publish fail. Nothing in this repository does yet: gated-dag makes exactly that check, but through its own wildcard-aware subjectCovered rather than this set comparison. The framework reports; an owner that needs to act has the values to act on.

func (StreamFieldDivergence) String 

func (d StreamFieldDivergence) String() string

String renders one divergence for a diagnostic.

type StreamLister 

type StreamLister interface {
	ListStreams(context.Context, ...jetstream.StreamListOpt) jetstream.StreamInfoLister
	StreamNames(context.Context, ...jetstream.StreamListOpt) jetstream.StreamNameLister
}

StreamLister is the narrow JetStream capability the inventory needs: the two paged account listings. jetstream.JetStream satisfies it.

BOTH are required, and neither is a per-resource round-trip. ListStreams returns configuration and state together, which is what keeps the inventory off an N+1 describe path. StreamNames is the completeness check: the server excludes an offline stream from the info listing but not from the name listing, so without the second listing the inventory silently omits exactly the resources nobody can read.

The interface is this narrow so the collector's classification, ordering, reconciliation, degradation, and timeout behavior are all testable without a server, while the production path still drives both real listings through an integration test.

type StreamListerSource 

type StreamListerSource func() (StreamLister, error)

StreamListerSource yields the account lister for ONE collection. It is resolved per collection rather than captured at construction, so a collector built before its client connects starts working when the client does, and one that outlives a reconnect never holds a stale JetStream context.

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. The stream carries declared bounds; see testStreamConfig.

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 ThresholdSource 

type ThresholdSource func() StoragePressureThresholds

ThresholdSource yields the pressure thresholds for ONE evaluation.

It is a FUNCTION, and that is the whole point. SemStreams is flow-based and its components are runtime-reconfigurable — watchConfigUpdates is launched after the boot barrier specifically so post-boot component edits reach running components. A threshold captured into a value at composition-root construction would apply the stale number successfully and silently after an operator edit: nothing errors, the report just answers the wrong question. Resolving at the seam, from live configuration, is the same discipline the bucket catalog applied at the acquisition seam.

type TierComparison 

type TierComparison struct {
	// Tier is the storage tier this row compares within. Memory and file have
	// separate account limits; the unknown tier has none at all.
	Tier StorageTier `json:"tier"`

	// Limit is the account ceiling for this tier paired with the account's
	// usage against it — bounded, unbounded, or unknown, on the same three-state
	// model every capacity in this capability uses.
	Limit Capacity `json:"limit"`

	// DeclaredBytes is the sum of the configured bounds of the BOUNDED
	// resources in this tier. Unbounded and unknown-capacity resources
	// contribute nothing, which makes this a FLOOR rather than a total whenever
	// UnboundedResources or UnknownResources is non-zero.
	DeclaredBytes int64 `json:"declared_bytes"`

	// The resource counts behind DeclaredBytes, published so a consumer can see
	// how much of the tier the sum actually covers.
	BoundedResources   int `json:"bounded_resources"`
	UnboundedResources int `json:"unbounded_resources"`
	UnknownResources   int `json:"unknown_resources"`

	// State is the verdict. Report-only: nothing rejects, throttles, or
	// degrades because a tier is over-committed.
	State OvercommitmentState `json:"state"`

	// Unavailable names why State is not applicable. Empty otherwise.
	Unavailable string `json:"unavailable,omitempty"`

	// Growth, Projection and Pressure are the tier ceiling's own capacity
	// picture, measured against the account's usage of this tier rather than
	// against any resource's declaration.
	//
	// They are here because a tier limit is a CEILING like any other, and an
	// unbounded resource has no other one. Over-commitment answers a different
	// question — "do the declarations filed against this tier fit inside it" —
	// and answers it about declared bounds, so a tier can be comfortably
	// within-limit while the bytes actually stored are about to exhaust it. Both
	// are published because they fail independently.
	//
	// The rate is measured across the account row's own retained history, which
	// carries this tier's usage at every past collection (storage_report.go). It
	// is the tier's rate, not the sum of any subset of resource rates: resources
	// appear and disappear between collections, and a sum over the ones that
	// happen to be present would move for reasons that are not growth.
	Growth     Growth     `json:"growth"`
	Projection Projection `json:"projection"`
	Pressure   Pressure   `json:"pressure"`
}

TierComparison is one storage tier's account-level capacity picture.

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

type WritePolicy 

type WritePolicy string

WritePolicy declares who may write a framework bucket. The framework-owned write-guard set is DERIVED from this field (Write == WriteOwnerOnly), never hand-maintained. 

const (
	// WriteOwnerOnly means only the declared owner writes; a generic KV writer
	// (rule update_kv) is rejected at load and at runtime.
	WriteOwnerOnly WritePolicy = "owner-only"
	// WriteOpen means any component may write (e.g. COMPONENT_STATUS).
	WriteOpen WritePolicy = "open"
)

Write policies.

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.