nats

package module
v0.7.0 Latest Latest
Warning

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

 Go to latest Published: Aug 22, 2026 License: Apache-2.0 Imports: 39 Imported by: 0

README

nats

JetStream client for Go. Telemetry is optional via tel.

Designed for teams that want explicit JetStream topology ownership and predictable runtime behavior in production services.

Module: github.com/gopherust-io/nats

OpenSSF Scorecard

Latest stable release: see GitHub Releases.

Quick links: Architecture · Getting started · API reference · Compatibility

Mental model: the app connects and binds. Ops owns streams (CLI/platform). If no stream captures the subject, the message is not retained—that is topology, not a client bug.

go get github.com/gopherust-io/nats@latest
# local broker: nats-server -js  — or labs in nats-console (make nats-up)

# production topology — not NewClient
nats stream add ORDERS --subjects 'orders.>' --storage file --retention workqueue

Labs may call Streams() / Consumers() / SetupWorker explicitly. See API reference. cfg.Stream on NewClient does not create anything.

Worker sketch

package main

import (
	"context"
	"log"

	libnats "github.com/gopherust-io/nats"
	"github.com/gopherust-io/tel"
	natspkg "github.com/nats-io/nats.go"
)

func main() {
	ctx := context.Background()

	telem := tel.NewWithConfig(tel.DefaultDebugConfig()) // prod: DefaultConfig + collector
	_ = telem.Start(ctx)
	defer telem.Shutdown(ctx)
	ctx = tel.WrapContext(ctx, telem)

	cfg := libnats.DefaultConfig()
	cfg.Conn.Address = "nats://127.0.0.1:4222"

	client, err := libnats.NewClient(ctx, &cfg)
	if err != nil {
		log.Fatal(err)
	}
	defer client.Connector().Shutdown()

	// Stream must already exist (CLI above). Then:
	if err := client.Publisher().PublishJSON(ctx, "orders.created", map[string]any{"id": "42"}); err != nil {
		log.Fatal(err)
	}

	handler := func(ctx context.Context, msg *natspkg.Msg) error {
		// nil → Ack; non-nil → Nak / retry
		return nil
	}

	_, err = client.Consumer().QueueSubscribeBound(
		ctx,
		"ORDERS",           // stream
		"orders-processor", // durable
		"orders-workers",   // queue
		"orders.>",         // filter
		handler,
	)
	if err != nil {
		log.Fatal(err)
	}

	select {}
}

Publish variants: PublishBytes, PublishWithMsgID (requires stream DuplicateWindow + stable IDs). Pick one codec per stream and stay consistent.

Pull

Create the pull durable first (CLI or Consumers().CreateOrUpdateConsumer), then:

pull, err := client.Consumer().Pull("ORDERS", "orders-puller")
if err != nil {
	return err
}
return pull.Process(ctx, handler)

Request / Reply (core NATS)

JetStream is not involved. Requester calls Request*; Responder uses core subscribe (no auto Ack/Nak). Handler must Respond*. Codecs match Publisher: Bytes, JSON, MessagePack, Proto.

_, err = client.Responder().Subscribe(ctx, "echo", func(_ context.Context, msg *natspkg.Msg) error {
	return libnats.RespondJSON(msg, map[string]any{"ok": true})
})

var resp map[string]any
err = client.Requester().RequestJSONInto(ctx, "echo", map[string]any{"ping": 1}, &resp)

Also: RequestBytes / RequestMsgPackInto / RequestProtoInto, and RespondBytes / RespondMsgPack / RespondProto. Queue groups: Responder().QueueSubscribe. Default request timeout is 2s when ctx has no deadline (RequesterConfig.Timeout).

Do not request on subjects captured by a JetStream stream—the server PubAck is delivered to the reply inbox. Auth that blocks _INBOX.> also breaks requests — see devops.

Start from DefaultConfig(), then apply knobs for your workload. Stream topology is always explicit (CreateOrUpdateStream / ops CLI) — the client factory does not invent it.

cfg := libnats.DefaultConfig()

// Job queue client (competing workers):
cfg.RuntimeConsumer.WorkerPoolEnabled = true
cfg.RuntimeConsumer.WorkerPoolSize = 8
cfg.RuntimeConsumer.WorkerBufferSize = 256
cfg.RuntimeConsumer.AckWait = 45 * time.Second
cfg.RuntimeConsumer.PendingMsgLimit = 1000
cfg.RuntimeConsumer.PendingMsgBuffer = 10 << 20 // 10 MiB
cfg.Backpressure.Mode = libnats.BackpressureNak
cfg.Backpressure.MaxAckPending = 1000

// Job stream (HA cluster):
stream := libnats.StreamConfig{
    Name: "ORDERS", Subjects: []string{"orders.>"},
    Storage: libnats.FileStorage, Retention: libnats.WorkQueuePolicy,
    Replicas: 3, Discard: libnats.DiscardOld,
}

More recipes (fan-out, local/dev, max QPS): consumer tuning · performance · devops.

vs nats.go

This module is an opinionated façade over legacy nats.go JetStream (JetStreamContext): bind-first consume, codecs, optional tel observability, and ops helpers (idempotency, dlq, shadow, …). Escape hatches: concrete client Conn() / JetStream().

It is not a faster protocol client. Competitive benches measure wrapper tax on the max-throughput path against equivalent legacy JetStream calls—not broker throughput.

Benchmarks

Codec comparison benchmarks

Internal codec matrix (make bench-codec): prefer Protobuf / PublishProto or raw PublishBytes on hot paths; JSON is for interoperability. See docs/performance.md.

Payload compression

Large bodies (>32 KiB): PayloadCompressionAuto (br→gzip), or forced PayloadCompressionGzip / PayloadCompressionBrotli on publish, request, and Responder().Respond* (package-level Respond* stay uncompressed). Pure + pub/sub + request/reply benches: make bench-payload — docs/performance.md § Payload compression.

Wrapper tax vs legacy nats.go

Same embedded JetStream broker, bytes payload, metrics/tracing off (max-throughput recipe), worker pool off. Sample (darwin/arm64, Apple M4 Pro; directional):

Path gopherust nats.go note
Sync Publish + PubAck ~71 µs, 36 allocs ~55 µs, 34 allocs small wrapper tax
Async Publish + wait ~51 µs, 42 allocs ~56 µs, 42 allocs ≈ parity
Push consume + Ack ~62 µs, 67 allocs ~65 µs, 64 allocs ≈ parity
Pull fetch + Ack ~112 µs, 105 allocs ~104 µs, 91 allocs small tax 
make bench-compete
make bench-payload
# system load (external broker):
go run ./tools/loadtest -impl gopherust -codec bytes -mode push -duration 30s
go run ./tools/loadtest -impl natsgo   -codec bytes -mode push -duration 30s

Methodology: docs/benchmarks.md · sample: benchcmp/results.sample.txt · baselines: ./scripts/bench-baseline.sh.

Extensions

Need Package / API
Consume dedup nats/idempotency
Dead letter nats/dlq (+ autopsy IncludeStack)
Shadow / canary nats/shadow · shadow.NewGraduate for ramp/abort
Sharding nats/shard
Worker pool nats/workerpool
Replay client.Replay()
Incident capsule client.Incidents() · CapsuleAuto
Adaptive backpressure Config.AdaptivePressure
Payload compress PublisherConfig / RequesterConfig / ResponderConfig .PayloadCompression
Supervised sub SuperviseQueueSubscribeBound, SupervisePullProcess

Footguns

Symptom Cause Fix
Publish “works”, nothing stored Subject not captured by any stream Fix stream Subjects / CLI
Wrong or missing deliveries Stream / durable / filter mismatch Use *SubscribeBound
Poison keeps disappearing Handler returns nil after failure Return error to Nak
Stream missing after deploy Relied on cfg.Stream in NewClient CLI or Streams().CreateOrUpdateStream
Request gets {"stream":...} JSON Subject captured by a JetStream stream Use core-only subjects for R/R
Consumer update rejected Deliver policy change on existing durable Recreate durable; see consumer docs

Development

# Start JetStream (nats-console: make nats-up — or nats-server -js)
make demo-nats    # ROLE=all|publisher|worker|puller
make test
make bench-compete
make loadtest

Docs

Guide When
Getting started Clone → Docker → demos
JetStream guide Publish / retain / consume
Recipes Production configs
Push vs pull Delivery model
Performance Codecs, AttrCache, throughput
Benchmarks Methodology + compete vs nats.go
Local Docker Compose labs (in nats-console)
API reference Binding, middleware, DefaultConfig

Contributing · Changelog

Compatibility and support

  • Supported Go/NATS versions are tracked in docs/compatibility.md.
  • main is active development; use tagged releases for production.
  • Stream topology is intentionally externalized (CLI/platform) for operational control.

License

Apache License 2.0 — see LICENSE.

Documentation

Index

Examples

Constants

View Source 
const (
	// MinPayloadCompressBytes is the exclusive lower bound for auto payload
	// compression (strictly greater than 32 KiB), matching nats-consol.
	MinPayloadCompressBytes = 32 << 10

	// MaxPayloadDecompressBytes caps expanded payload size (zip-bomb defense).
	MaxPayloadDecompressBytes = 64 << 20

	EncodingBrotli = "br"
	EncodingGzip   = "gzip"
)
View Source 
const (
	HeaderDLQOriginalSubject = dlq.HeaderOriginalSubject
	HeaderDLQStream          = dlq.HeaderStream
	HeaderDLQSequence        = dlq.HeaderSequence
	HeaderDLQNumDelivered    = dlq.HeaderNumDelivered
	HeaderDLQReason          = dlq.HeaderReason
	HeaderDLQConsumer        = dlq.HeaderConsumer

	HeaderAutopsyError = dlq.HeaderAutopsyError
	HeaderAutopsyHash  = dlq.HeaderAutopsyHash
	HeaderAutopsyStack = dlq.HeaderAutopsyStack
)

DLQ header keys (re-exported from nats/dlq).

View Source 
const (
	DefaultIncidentCapsuleBucket = "nats_incident_capsules"
	DefaultIncidentIndexBucket   = "nats_incident_index"
)
View Source 
const (
	MetaReplayUntilSeq = "replay_until_seq"
	MetaReplayLimit    = "replay_limit"
)

Consumer metadata keys for intended replay bounds (JetStream has no server-side end).

View Source 
const (
	// HeaderContentType is the historical NATS payload type header (json/protobuf/msgpack).
	HeaderContentType = "Nats-Content-Type"
	// HeaderNATSContentType is an alias of HeaderContentType.
	HeaderNATSContentType = HeaderContentType
	// HeaderNatsContentType is a legacy alias of HeaderContentType.
	HeaderNatsContentType = HeaderContentType
	HeaderHTTPContentType = "Content-Type"
	HeaderMsgID           = "Nats-Msg-Id"
	HeaderTraceID         = "Trace-Id"
	HeaderContentEncoding = "Content-Encoding"

	ContentTypeJSON    = "json"
	ContentTypeProto   = "protobuf"
	ContentTypeMsgPack = "msgpack"

	ApplicationJSON              = "application/json"
	ApplicationMsgPack           = "application/msgpack"
	ApplicationXMsgPack          = "application/x-msgpack"
	ApplicationProtobuf          = "application/protobuf"
	ApplicationXProtobuf         = "application/x-protobuf"
	ApplicationVNDGoogleProtobuf = "application/vnd.google.protobuf"
)
View Source 
const (
	DeliverAll             = natspkg.DeliverAllPolicy
	DeliverLast            = natspkg.DeliverLastPolicy
	DeliverNew             = natspkg.DeliverNewPolicy
	DeliverByStartSequence = natspkg.DeliverByStartSequencePolicy
	DeliverByStartTime     = natspkg.DeliverByStartTimePolicy
	DeliverLastPerSubject  = natspkg.DeliverLastPerSubjectPolicy

	ReplayInstant  = natspkg.ReplayInstantPolicy
	ReplayOriginal = natspkg.ReplayOriginalPolicy

	AckExplicit = natspkg.AckExplicitPolicy
	AckNone     = natspkg.AckNonePolicy
	AckAll      = natspkg.AckAllPolicy

	LimitsPolicy    = natspkg.LimitsPolicy
	InterestPolicy  = natspkg.InterestPolicy
	WorkQueuePolicy = natspkg.WorkQueuePolicy

	FileStorage   = natspkg.FileStorage
	MemoryStorage = natspkg.MemoryStorage

	DiscardOld = natspkg.DiscardOld
	DiscardNew = natspkg.DiscardNew
)
View Source 
const (
	HeaderSchema           = "X-NATS-Schema"
	HeaderSchemaVersion    = "X-NATS-Schema-Version"
	HeaderContractEncoding = "X-NATS-Contract-Encoding"
)

Schema contract headers (optional; not a full registry).

View Source 
const (
	SlowReasonPending    = "pending"
	SlowReasonLag        = "lag"
	SlowReasonAckPending = "ack_pending"
)
View Source 
const DefaultBehaviorFingerprintKVBucket = "nats_consol_fingerprints"

DefaultBehaviorFingerprintKVBucket is the JetStream KV bucket nats-consol reads for Consumer Detail behavior fingerprints.

View Source 
const DefaultMsgRangeMax = 1000

DefaultMsgRangeMax is the default cap for GetMsgRange / GetMsgRangeByTime.

Variables

View Source 
var (
	ErrEmptySubjectNotAllowed       = errors.New("empty subject not allowed")
	ErrEmptyConfigNotAllowed        = errors.New("empty config not allowed")
	ErrEmptyAddressNotAllowed       = errors.New("empty address not allowed")
	ErrInvalidSubscription          = errors.New("invalid subscription")
	ErrNatsConnectionNotEstablished = errors.New("NATS connection is not established")
	ErrInvalidStreamName            = errors.New("invalid stream name")
	ErrInvalidDurableName           = errors.New("invalid durable name")
	ErrInvalidQueueName             = errors.New("invalid queue name")
	ErrInvalidBucketName            = errors.New("invalid kv bucket name")
	ErrInvalidSubject               = errors.New("invalid subject")
	ErrInvalidKVKey                 = errors.New("invalid kv key")
	ErrBackpressureHandled          = errors.New("message handled by backpressure policy")
	ErrDrainTimeout                 = errors.New("connection drain timed out")
	ErrJetStreamV2Required          = errors.New("operation requires jetstream v2 API")
	ErrInvalidNKeySeed              = errors.New("invalid nkey seed")
	ErrConsumerRecreateRequired     = errors.New("consumer recreate required; delete and recreate explicitly to change immutable settings")
	ErrConflictingAuth              = errors.New(
		"conflicting auth: set only one of Address userinfo, Seed, User/Password, Secret, or CredentialsFile",
	)
	ErrSupervisorGiveUp   = errors.New("subscription supervisor gave up after max retries")
	ErrConsumerStall      = errors.New("consumer soft-liveness stall: pending rising without process activity")
	ErrInvalidReplayBound = errors.New("invalid replay bound")
)
View Source 
var (
	ErrInvalidMessageType         = errors.New("invalid message type")
	ErrInvalidTypeAssertion       = errors.New("invalid type assertion")
	ErrPoolFull                   = errors.New("worker pool queue full")
	ErrAsyncPublishPendingLimit   = errors.New("async publish pending limit exceeded")
	ErrUnsupportedContentEncoding = errors.New("unsupported content encoding")
	ErrPayloadTooLarge            = errors.New("decompressed payload exceeds size limit")
)
View Source 
var ErrDLQRouted = dlq.ErrDLQRouted

ErrDLQRouted is returned (and treated as success by processMessage) after a message has been published to the DLQ and Term'd.

View Source 
var ErrMonitoringBodyTooLarge = errors.New("monitoring response body too large")

ErrMonitoringBodyTooLarge is returned when a monitoring response exceeds the size cap.

View Source 
var ErrSchemaMismatch = errors.New("schema contract mismatch")

ErrSchemaMismatch is returned when message headers do not match SchemaExpectation.

View Source 
var ErrSendToDLQ = dlq.ErrSendToDLQ

ErrSendToDLQ may be returned from a handler to force DLQ routing + Term.

Functions

func ApplySchemaHeaders added in v0.7.0 

func ApplySchemaHeaders(msg *Message, name, version, encoding string)

ApplySchemaHeaders sets schema/contract headers on a Message before publish.

func BehaviorFingerprintKVKey added in v0.3.0 

func BehaviorFingerprintKVKey(stream, durable string) string

BehaviorFingerprintKVKey builds the KV key consol reads for a consumer.

func CheckSchema added in v0.7.0 

func CheckSchema(msg *natspkg.Msg, expect SchemaExpectation) error

CheckSchema validates msg headers against expect. When expect.Required is false and headers are absent, validation succeeds.

func ConsumerLagMessages added in v0.3.0 

func ConsumerLagMessages(streamLastSeq, deliveredStreamSeq uint64) uint64

ConsumerLagMessages returns max(0, streamLastSeq − deliveredStreamSeq).

func Decode 

func Decode(data []byte, typ MessageType, dst any) error

func DecodeMsg 

func DecodeMsg(msg *natspkg.Msg, typ MessageType, dst any) error

DecodeMsg decodes msg into dst. Content-Encoding is expanded only when decompress is desired by the caller via DecodeMsgWithDecompress, or when the consumer already expanded the payload (header stripped). Prefer processMessage PayloadDecompression for the hot path; this helper does not auto-decompress so it cannot bypass RuntimeConsumerConfig.PayloadDecompression.

func DecodeMsgWithDecompress added in v0.6.0 

func DecodeMsgWithDecompress(msg *natspkg.Msg, typ MessageType, dst any) error

DecodeMsgWithDecompress expands Content-Encoding (size-capped) then decodes.

func DecodeMsgWithSchema added in v0.7.0 

func DecodeMsgWithSchema(msg *natspkg.Msg, typ MessageType, dst any, expect SchemaExpectation) error

DecodeMsgWithSchema runs CheckSchema then DecodeMsg.

func DecodeProto 

func DecodeProto(data []byte, msg proto.Message) error

func DecodeTyped 

func DecodeTyped[T any](msg *natspkg.Msg, typ MessageType) (T, error)

DecodeTyped is a generic helper for typed message handlers.

func DecompressPayload added in v0.6.0 

func DecompressPayload(data []byte, encoding string) ([]byte, error)

DecompressPayload expands data according to Content-Encoding (br / gzip). Empty or "identity" encoding returns data unchanged.

func Encode 

func Encode(msg Message) ([]byte, error)

func EvaluateBehaviorFingerprint added in v0.3.0 

func EvaluateBehaviorFingerprint(current, baseline BehaviorSnapshot, cfg BehaviorFingerprintConfig) bool

EvaluateBehaviorFingerprint reports whether current looks anomalous vs baseline. Rate must stay within RateTolerance of baseline while latency exceeds LatencyFactor.

func EvaluateSlowConsumer added in v0.3.0 

func EvaluateSlowConsumer(pending, lag uint64, ackPending, maxAckPending int, cfg SlowConsumerConfig) (bool, []string)

EvaluateSlowConsumer returns whether thresholds are met and why. Thresholds of 0 mean "use defaults" via withDefaults on cfg first, or pass an already-defaulted config.

func InProgress 

func InProgress(msg *natspkg.Msg) error

InProgress tells JetStream the handler is still working (extends AckWait).

func ListStreams 

func ListStreams(ctx context.Context, s StreamManager) ([]*natspkg.StreamInfo, error)

ListStreams returns all stream infos (convenience; not on StreamManager interface).

func MarshalBehaviorFingerprintKV added in v0.3.0 

func MarshalBehaviorFingerprintKV(
	stream, durable string,
	anomaly bool,
	normal, current BehaviorSnapshot,
	sustained time.Duration,
	at time.Time,
) ([]byte, error)

MarshalBehaviorFingerprintKV encodes a consol-compatible fingerprint payload.

func MaybeCompressPayload added in v0.6.0 

func MaybeCompressPayload(plain []byte) ([]byte, string, bool)

MaybeCompressPayload compresses plain when larger than MinPayloadCompressBytes. Prefer brotli, then gzip (best-speed). Returns ok only when compressed size is strictly smaller than plain.

func NakWithDelay 

func NakWithDelay(msg *natspkg.Msg, delay time.Duration) error

NakWithDelay negatively acknowledges with a server-side redelivery delay.

func ReportBehaviorFingerprintKV added in v0.3.0 

func ReportBehaviorFingerprintKV(
	ctx context.Context,
	keys KeyValueKeys,
	bucket, stream, durable string,
	anomaly bool,
	normal, current BehaviorSnapshot,
	sustained time.Duration,
) error

ReportBehaviorFingerprintKV publishes a fingerprint snapshot for nats-consol.

func RespondBytes 

func RespondBytes(msg *natspkg.Msg, data []byte) error

RespondBytes replies with raw bytes (no compression).

func RespondJSON 

func RespondJSON(msg *natspkg.Msg, v any) error

RespondJSON encodes v as JSON and replies (sets content-type header; no compression).

func RespondMsgPack 

func RespondMsgPack(msg *natspkg.Msg, v any) error

RespondMsgPack encodes v as MessagePack and replies (no compression).

func RespondProto 

func RespondProto(msg *natspkg.Msg, v proto.Message) error

RespondProto encodes v as protobuf and replies (no compression).

func SchemaFromHeader added in v0.7.0 

func SchemaFromHeader(h natspkg.Header) (string, string, string)

SchemaFromHeader reads schema contract headers from a NATS message.

func ShardIndex 

func ShardIndex(key string, numShards int) int

ShardIndex returns a stable shard index for key in [0, numShards). Prefer importing github.com/gopherust-io/nats/shard directly.

func ShardSubject 

func ShardSubject(prefix, key string, numShards int, action string) string

ShardSubject builds a subject for keyed sharding. Prefer importing github.com/gopherust-io/nats/shard directly.

func StreamNames 

func StreamNames(ctx context.Context, s StreamManager) ([]string, error)

StreamNames returns sorted stream names via a full ListStreamsPage.

func SupervisePullProcess 

func SupervisePullProcess(
	ctx context.Context,
	cfg SupervisorConfig,
	metrics *clientMetrics,
	run func(context.Context) error,
) error

SupervisePullProcess runs Pull.Process in a loop, restarting after failures with backoff.

func TermWithReason 

func TermWithReason(msg *natspkg.Msg, reason string) error

TermWithReason terminates redelivery and attaches a reason (servers that support it). Empty reason falls back to Term().

func TraceIDFromHeader 

func TraceIDFromHeader(h natspkg.Header) string

TraceIDFromHeader returns the explicit Trace-Id header value, if present.

func ValidateBucketName 

func ValidateBucketName(name string) error

ValidateBucketName checks JetStream KV bucket naming rules (alphanumeric, underscore, hyphen only).

func ValidateDurableName 

func ValidateDurableName(name string) error

ValidateDurableName checks JetStream durable/consumer naming rules (same constraints as stream names).

func ValidateKVKey 

func ValidateKVKey(key string) error

ValidateKVKey checks JetStream KV key naming rules.

func ValidatePublishSubject 

func ValidatePublishSubject(subject string) error

ValidatePublishSubject checks that subject is a valid literal publish subject (no wildcards).

func ValidateQueueName 

func ValidateQueueName(name string) error

ValidateQueueName checks queue group naming rules (same constraints as durables).

func ValidateStreamName 

func ValidateStreamName(name string) error

ValidateStreamName checks JetStream stream naming rules. Names must be non-empty, filesystem-friendly, and must not contain whitespace, '.', '*', '>', '/', or '\'.

func ValidateSubject 

func ValidateSubject(subject string) error

ValidateSubject checks that subject is a valid NATS subject (wildcards allowed).

func ValidateSubjects 

func ValidateSubjects(subjects []string) error

ValidateSubjects validates a list of stream/filter subjects (wildcards allowed).

Types

type AckPolicy 

type AckPolicy = natspkg.AckPolicy

Deliver and replay policies mirror github.com/nats-io/nats.go constants.

type AdaptivePressure added in v0.6.0 

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

AdaptivePressure holds the latest decision for pull/push actuators.

goalign:ignore

func NewAdaptivePressure added in v0.6.0 

func NewAdaptivePressure(cfg AdaptivePressureConfig) *AdaptivePressure

func (*AdaptivePressure) Decision added in v0.6.0 

func (a *AdaptivePressure) Decision() AdaptivePressureDecision

func (*AdaptivePressure) FetchBatchOr added in v0.6.0 

func (a *AdaptivePressure) FetchBatchOr(defaultBatch int) int

func (*AdaptivePressure) Observe added in v0.6.0 

func (a *AdaptivePressure) Observe(in AdaptivePressureInput) AdaptivePressureDecision

type AdaptivePressureConfig added in v0.6.0 

type AdaptivePressureConfig struct {
	Enabled       bool
	MinFetchBatch int
	MaxFetchBatch int
	MinAckPending int
	NakDelayMin   time.Duration
	NakDelayMax   time.Duration
}

AdaptivePressureConfig enables closed-loop fetch/backpressure tuning.

goalign:ignore

type AdaptivePressureDecision added in v0.6.0 

type AdaptivePressureDecision struct {
	Level      AdaptivePressureLevel
	FetchBatch int
	AckPending int
	NakDelay   time.Duration
}

AdaptivePressureDecision is the actuator output.

goalign:ignore

func DecideAdaptivePressure added in v0.6.0 

func DecideAdaptivePressure(cfg AdaptivePressureConfig, in AdaptivePressureInput) AdaptivePressureDecision

DecideAdaptivePressure is a pure control step (unit-testable).

type AdaptivePressureInput added in v0.6.0 

type AdaptivePressureInput struct {
	PoolDepth       int
	PoolCapacity    int
	HandlerLatency  time.Duration
	BaselineLatency time.Duration
	Lag             uint64
	Pending         uint64
	CurrentBatch    int
	CurrentAckPend  int
	MaxAckPending   int
}

AdaptivePressureInput is one observation sample for the controller.

goalign:ignore

type AdaptivePressureLevel added in v0.6.0 

type AdaptivePressureLevel uint8

AdaptivePressureLevel is the control-loop severity (0=idle … 4=critical). 

const (
	PressureIdle AdaptivePressureLevel = iota
	PressureWarm
	PressureElevated
	PressureHard
	PressureCritical
)

type AutopsyConfig 

type AutopsyConfig = dlq.AutopsyConfig

AutopsyConfig enriches DLQ publishes with forensic headers.

type BackpressureConfig 

type BackpressureConfig struct {
	Mode                     BackpressureMode
	MaxAckPending            int
	PendingMsgLimit          int
	PendingMsgBuffer         int
	QueueDepthSampleInterval time.Duration
}

type BackpressureMode 

type BackpressureMode uint8
const (
	BackpressureBlock BackpressureMode = iota + 1
	BackpressureNak
	BackpressureTerm
	BackpressureDrop
)

type BehaviorAnomalyEvent added in v0.3.0 

type BehaviorAnomalyEvent struct {
	Stream       string
	Durable      string
	Normal       BehaviorSnapshot
	Current      BehaviorSnapshot
	SustainedFor time.Duration
}

BehaviorAnomalyEvent is emitted when throughput stays near baseline while handling latency regresses by LatencyFactor for SustainFor.

type BehaviorFingerprint added in v0.3.0 

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

BehaviorFingerprint learns normal msg/min and processing latency, then detects regressions. goalign:ignore

func WatchBehaviorFingerprint added in v0.3.0 

func WatchBehaviorFingerprint(
	ctx context.Context,
	sub Subscription,
	cfg BehaviorFingerprintConfig,
	metrics *clientMetrics,
) (*BehaviorFingerprint, error)

WatchBehaviorFingerprint starts learning baselines from Observe samples. Register Observe via consumer.OnMessageHandled, or call Observe manually.

func (*BehaviorFingerprint) Anomalous added in v0.3.0 

func (b *BehaviorFingerprint) Anomalous() bool

Anomalous reports whether an anomaly has been detected (sticky until Stop).

func (*BehaviorFingerprint) Events added in v0.3.0 

func (b *BehaviorFingerprint) Events() <-chan BehaviorAnomalyEvent

Events returns anomaly notifications (buffered; overflows are dropped).

func (*BehaviorFingerprint) Observe added in v0.3.0 

func (b *BehaviorFingerprint) Observe(elapsed time.Duration)

Observe records one completed message handling duration.

func (*BehaviorFingerprint) Snapshot added in v0.3.0 

func (b *BehaviorFingerprint) Snapshot() (BehaviorSnapshot, BehaviorSnapshot, bool)

Snapshot returns the learned baseline, latest window stats, and whether a baseline exists.

func (*BehaviorFingerprint) Stop added in v0.3.0 

func (b *BehaviorFingerprint) Stop()

Stop ends the watch loop.

type BehaviorFingerprintConfig added in v0.3.0 

type BehaviorFingerprintConfig struct {
	OnAnomaly func(BehaviorAnomalyEvent)
	// ReportKV, when set, publishes Normal/Current snapshots for nats-consol.
	ReportKV KeyValueKeys
	// ReportBucket defaults to DefaultBehaviorFingerprintKVBucket.
	ReportBucket string
	// PollInterval is how often the rolling window is evaluated (default 5s).
	PollInterval time.Duration
	// Window is the sample window used for current rate/latency (default 60s).
	Window time.Duration
	// Warmup is how long to learn baselines before detecting (default 5m).
	Warmup time.Duration
	// MinSamples required before detection (default 50).
	MinSamples int
	// LatencyFactor fires when current processing >= factor × baseline (default 3).
	LatencyFactor float64
	// RateTolerance is the ± fraction of baseline rate treated as "same throughput" (default 0.3).
	RateTolerance float64
	// SustainFor is how long the anomaly condition must hold before firing (default 30s).
	SustainFor time.Duration
	// CircuitStop stops the watcher after the first anomaly.
	CircuitStop bool
}

BehaviorFingerprintConfig controls WatchBehaviorFingerprint. goalign:ignore

type BehaviorFingerprintKVPayload added in v0.3.0 

type BehaviorFingerprintKVPayload struct {
	UpdatedAt      time.Time                     `json:"updatedAt"`
	Stream         string                        `json:"stream"`
	Durable        string                        `json:"durable"`
	Normal         BehaviorFingerprintKVSnapshot `json:"normal"`
	Current        BehaviorFingerprintKVSnapshot `json:"current"`
	SustainedForMs int64                         `json:"sustainedForMs,omitempty"`
	Anomaly        bool                          `json:"anomaly"`
}

BehaviorFingerprintKVPayload is stored at key stream/durable in the fingerprint bucket. goalign:ignore

type BehaviorFingerprintKVSnapshot added in v0.3.0 

type BehaviorFingerprintKVSnapshot struct {
	MsgPerMin    float64 `json:"msgPerMin"`
	ProcessingMs float64 `json:"processingMs"`
}

BehaviorFingerprintKVSnapshot is the JSON shape published for consol UI.

func SnapshotToKV added in v0.3.0 

func SnapshotToKV(s BehaviorSnapshot) BehaviorFingerprintKVSnapshot

SnapshotToKV converts library snapshots to the consol JSON shape.

type BehaviorSnapshot added in v0.3.0 

type BehaviorSnapshot struct {
	MsgPerMin  float64
	Processing time.Duration
}

BehaviorSnapshot is a rate + processing-time fingerprint for a consumer.

type Capsule added in v0.6.0 

type Capsule struct {
	CreatedAt      time.Time                     `json:"createdAt"`
	Fingerprint    *BehaviorFingerprintKVPayload `json:"fingerprint,omitempty"`
	ID             string                        `json:"id"`
	Stream         string                        `json:"stream"`
	Consumer       string                        `json:"consumer"`
	Trigger        IncidentTrigger               `json:"trigger"`
	Subject        string                        `json:"subject,omitempty"`
	Reason         string                        `json:"reason,omitempty"`
	Messages       []CapsuleMessage              `json:"messages"`
	FlightTimeline []IncidentEvent               `json:"flightTimeline,omitempty"`
	FailingSeq     uint64                        `json:"failingSeq,omitempty"`
	SchemaVersion  int                           `json:"schemaVersion"`
}

Capsule is a portable forensic pack for offline ReplayLocal.

goalign:ignore

type CapsuleAuto added in v0.6.0 

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

CapsuleAuto captures capsules from DLQ / shadow / anomaly hooks.

goalign:ignore

func NewCapsuleAuto added in v0.6.0 

func NewCapsuleAuto(client Client, stream, durable string, cfg CapsuleAutoConfig) *CapsuleAuto

func (*CapsuleAuto) DLQRecorder added in v0.6.0 

func (a *CapsuleAuto) DLQRecorder() dlq.Recorder

DLQRecorder implements dlq.Recorder and captures capsules.

func (*CapsuleAuto) OnAnomaly added in v0.6.0 

func (a *CapsuleAuto) OnAnomaly() func(BehaviorAnomalyEvent)

OnAnomaly returns a BehaviorFingerprint OnAnomaly callback.

func (*CapsuleAuto) ShadowRecorder added in v0.6.0 

func (a *CapsuleAuto) ShadowRecorder() shadow.Recorder

ShadowRecorder implements shadow.Recorder and captures on mismatch.

type CapsuleAutoConfig added in v0.6.0 

type CapsuleAutoConfig struct {
	StoreBucket    string
	IndexBucket    string
	Window         int
	FlightRecorder *FlightRecorder
	OnReady        func(*Capsule)
	Redact         func(*CapsuleMessage)
	Enabled        bool
}

CapsuleAutoConfig wires automatic Incident Capsule capture from ops hooks.

goalign:ignore

type CapsuleMessage added in v0.6.0 

type CapsuleMessage struct {
	Time     time.Time           `json:"time"`
	Header   map[string][]string `json:"header,omitempty"`
	Subject  string              `json:"subject"`
	Data     []byte              `json:"data"`
	Sequence uint64              `json:"sequence"`
}

CapsuleMessage is one stored message inside an incident capsule.

type CircuitBreaker added in v0.7.0 

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

CircuitBreaker pauses pull fetches when pressure is critical (metric: circuit_open).

func NewCircuitBreaker added in v0.7.0 

func NewCircuitBreaker(cfg CircuitConfig) *CircuitBreaker

NewCircuitBreaker returns a breaker or nil when disabled.

func (*CircuitBreaker) IsOpen added in v0.7.0 

func (b *CircuitBreaker) IsOpen() bool

IsOpen reports the last observed open state.

func (*CircuitBreaker) Opens added in v0.7.0 

func (b *CircuitBreaker) Opens() int64

Opens returns how many times the circuit transitioned to open.

func (*CircuitBreaker) PauseDuration added in v0.7.0 

func (b *CircuitBreaker) PauseDuration() time.Duration

PauseDuration is how long Process sleeps while the circuit is open.

func (*CircuitBreaker) ShouldPause added in v0.7.0 

func (b *CircuitBreaker) ShouldPause(d AdaptivePressureDecision) bool

ShouldPause reports whether pull Process should delay the next fetch.

type CircuitConfig added in v0.7.0 

type CircuitConfig struct {
	PauseDuration time.Duration
	OpenOnLevel   AdaptivePressureLevel // opens at this level (default PressureCritical)
	Enabled       bool
}

CircuitConfig enables degrade / pause modes when Adaptive Pressure hits critical.

goalign:ignore

type Client 

type Client interface {
	Consumer() Consumer
	Publisher() Publisher
	Requester() Requester
	Responder() Responder
	Connector() Connector
	Streams() StreamManager
	Consumers() ConsumerManager
	KV() KeyValueManager
	KVKeys() KeyValueKeys
	Objects() ObjectStoreManager
	Monitoring() Monitoring
	Replay() Replay
	Incidents() Incidents
	// PublishRaw publishes bytes with optional headers and returns the JetStream ack.
	PublishRaw(ctx context.Context, subject string, data []byte, headers map[string]string) (*PubAck, error)
	SetupWorker(ctx context.Context, setup WorkerSetup, handler MsgHandler) (Subscription, error)
	OrderedSubscribe(ctx context.Context, stream string, cfg OrderedConsumerConfig, handler MsgHandler) (OrderedSubscription, error)
	SuperviseQueueSubscribeBound(ctx context.Context, stream, durable, queue, subject string, handler MsgHandler, cfg SupervisorConfig) (SupervisedSubscription, error)
	SuperviseSubscribeBound(ctx context.Context, stream, durable, subject string, handler MsgHandler, cfg SupervisorConfig) (SupervisedSubscription, error)
	SupervisePullProcess(ctx context.Context, stream, durable string, handler MsgHandler, cfg SupervisorConfig, opts ...ProcessOpt) error
	WatchSoftLiveness(ctx context.Context, sub Subscription, cfg SoftLivenessConfig) (*SoftLiveness, error)
	WatchBehaviorFingerprint(ctx context.Context, sub Subscription, cfg BehaviorFingerprintConfig) (*BehaviorFingerprint, error)
	// WithShadow wraps handlers and records shadow_* metrics when metrics are enabled.
	WithShadow(cfg ShadowConfig, primary, shadowHandler MsgHandler) MsgHandler
}

func NewClient 

func NewClient(ctx context.Context, cfg *Config) (Client, error)

type Config 

type Config struct {
	PublisherConfig  PublisherConfig
	RequesterConfig  RequesterConfig
	ResponderConfig  ResponderConfig
	Metrics          MetricsConfig
	RuntimeConsumer  RuntimeConsumerConfig
	Stream           StreamConfig
	Conn             Connection
	Backpressure     BackpressureConfig
	AdaptivePressure AdaptivePressureConfig
	Circuit          CircuitConfig
}

func DefaultConfig 

func DefaultConfig() Config
Example

ExampleDefaultConfig shows starting from DefaultConfig and applying job-worker knobs. 

cfg := DefaultConfig()
cfg.RuntimeConsumer.WorkerPoolEnabled = true
cfg.RuntimeConsumer.WorkerPoolSize = 8
cfg.RuntimeConsumer.WorkerBufferSize = 256
_ = cfg.Conn.Address

type Connection 

type Connection struct {
	// CustomReconnectDelay overrides the built-in capped exponential delay.
	CustomReconnectDelay func(attempts int) time.Duration

	// Optional hooks run after the library's internal connection handlers.
	OnDisconnect func(*natspkg.Conn, error)
	OnReconnect  func(*natspkg.Conn)
	OnClosed     func(*natspkg.Conn)

	Address  string
	User     string
	Password string
	Seed     string
	Secret   string
	// CredentialsFile is a NATS .creds / chained credentials file (operator JWT model).
	// Mutually exclusive with Seed, User/Password, and Secret.
	CredentialsFile string
	MetricPrefix    string
	// Domain selects a JetStream domain ($JS.<domain>.API) for multi-domain / hub leaf setups.
	Domain string

	ClientName string
	// TLS configures secure connections (mTLS / CA verification).
	TLS ConnectionTLS

	ReconnectWait      time.Duration
	MaxReconnect       int
	ConnectTimeout     time.Duration
	DrainTimeout       time.Duration
	PingInterval       time.Duration
	MaxPingsOut        int
	FlusherTimeout     time.Duration
	ReconnectJitter    time.Duration
	ReconnectJitterTLS time.Duration

	// ReconnectBufSize is the outbound buffer while reconnecting.
	// 0 leaves the nats.go library default (8 MiB); -1 disables buffering.
	ReconnectBufSize int

	HealthCheckInterval  time.Duration
	InitialRetryAttempts int
	InitialRetryWait     time.Duration
	InitialRetryBackoff  float64
	RetryOnFailedConnect bool
	AllowReconnect       bool
	AllowMetrics         bool
	// DontRandomize keeps server order when Address is a comma-separated list.
	DontRandomize bool
}

Connection holds dial, auth, TLS, and reconnect settings.

goalign:ignore

type ConnectionStatus 

type ConnectionStatus struct {
	LastDisconnect time.Time
	LastError      error
	ServerURL      string
	ReconnectCount int64
	Connected      bool
	InLameDuck     bool
}

ConnectionStatus is a snapshot of connector health and reconnect state.

goalign:ignore

type ConnectionTLS 

type ConnectionTLS struct {
	// Config is used as-is when non-nil (takes precedence over PEM fields).
	Config *tls.Config
	// ServerName overrides the hostname used for certificate verification.
	ServerName string
	// CA is a PEM-encoded CA certificate pool.
	CA []byte
	// Cert and Key are a PEM-encoded client certificate pair (mTLS).
	Cert []byte
	Key  []byte
	// InsecureSkipVerify disables server certificate verification (dev only).
	InsecureSkipVerify bool
}

ConnectionTLS holds optional TLS material for NATS connections. Prefer PEM fields for config-file / env wiring; set Config for full control.

goalign:ignore

type Connector 

type Connector interface {
	IsConnected() bool
	Shutdown() error
	HealthCheck(ctx context.Context) error
	ConnectionStatus() ConnectionStatus
	WaitConnected(ctx context.Context) error
	Conn() *natspkg.Conn
	JetStream() natspkg.JetStreamContext
	AccountInfo(ctx context.Context) (*natspkg.AccountInfo, error)
}

type Consumer 

type Consumer interface {
	Subscribe(ctx context.Context, subject string, handler MsgHandler, opts ...natspkg.SubOpt) (Subscription, error)
	QueueSubscribe(ctx context.Context, queue, subject string, handler MsgHandler, opts ...natspkg.SubOpt) (Subscription, error)
	SubscribeBound(ctx context.Context, stream, durable, subject string, handler MsgHandler) (Subscription, error)
	QueueSubscribeBound(ctx context.Context, stream, durable, queue, subject string, handler MsgHandler) (Subscription, error)
	Pull(stream, durable string) (PullConsumer, error)
	OrderedSubscribe(ctx context.Context, stream string, cfg OrderedConsumerConfig, handler MsgHandler) (OrderedSubscription, error)
}

type ConsumerConfig 

type ConsumerConfig = RuntimeConsumerConfig

ConsumerConfig is an alias kept for backward compatibility.

type ConsumerManager 

type ConsumerManager interface {
	CreateOrUpdateConsumer(ctx context.Context, stream string, cfg DurableConsumerConfig) (*natspkg.ConsumerInfo, error)
	AddConsumer(ctx context.Context, stream string, cfg *natspkg.ConsumerConfig) (*natspkg.ConsumerInfo, error)
	UpdateConsumer(ctx context.Context, stream string, cfg *natspkg.ConsumerConfig) (*natspkg.ConsumerInfo, error)
	DeleteConsumer(ctx context.Context, stream, durable string) error
	ConsumerInfo(ctx context.Context, stream, durable string) (*natspkg.ConsumerInfo, error)
	ConsumerNames(ctx context.Context, stream string) ([]string, error)
	ListConsumers(ctx context.Context, stream string) ([]*natspkg.ConsumerInfo, error)
	ListConsumersPage(ctx context.Context, stream string, offset, limit int) ([]*natspkg.ConsumerInfo, int, error)
	PauseConsumer(ctx context.Context, stream, durable string, pauseUntil time.Time) error
	ResumeConsumer(ctx context.Context, stream, durable string) error
}

type DLQConfig 

type DLQConfig struct {
	Publisher Publisher
	// Recorder receives an IncidentDLQ event after a successful route (optional).
	Recorder *FlightRecorder
	// Subject is the dead-letter publish subject (e.g. "orders.dlq").
	Subject string
	// Reason is stored in HeaderDLQReason when auto-routing on MaxDeliver.
	Reason string
	// Autopsy adds forensic headers (error, payload hash, optional stack).
	Autopsy AutopsyConfig
	// MaxDeliver routes to DLQ when metadata.NumDelivered >= MaxDeliver (0 disables auto route).
	MaxDeliver uint64
}

DLQConfig configures WithDLQ.

type DeliverPolicy 

type DeliverPolicy = natspkg.DeliverPolicy

Deliver and replay policies mirror github.com/nats-io/nats.go constants.

type DiscardPolicy 

type DiscardPolicy = natspkg.DiscardPolicy

Deliver and replay policies mirror github.com/nats-io/nats.go constants.

type DurableConsumerConfig 

type DurableConsumerConfig struct {
	OptStartTime   *time.Time
	Metadata       map[string]string
	Durable        string
	FilterSubject  string
	FilterSubjects []string
	// DeliverSubject / DeliverGroup preserve push consumers across ResetConsumer.
	// Empty DeliverSubject means pull.
	DeliverSubject    string
	DeliverGroup      string
	DeliverPolicy     DeliverPolicy
	ReplayPolicy      ReplayPolicy
	AckPolicy         AckPolicy
	MaxDeliver        int
	AckWait           time.Duration
	MaxAckPending     int
	RateLimit         uint64
	Heartbeat         time.Duration
	InactiveThreshold time.Duration
	Replicas          int
	MaxWaiting        int
	OptStartSeq       uint64
	FlowControl       bool
	MemStorage        bool
	// HasAckPolicy marks AckPolicy as intentionally set so AckNone (zero) is preserved.
	HasAckPolicy bool
	// HasDeliverPolicy marks DeliverPolicy as intentionally set (including DeliverAll).
	HasDeliverPolicy bool
	// HasReplayPolicy marks ReplayPolicy as intentionally set (including ReplayInstant).
	HasReplayPolicy bool
}

DurableConsumerConfig holds durable consumer create/update options.

goalign:ignore

type FetchOpt 

type FetchOpt func(*fetchConfig)

func WithFetchHeartbeat 

func WithFetchHeartbeat(d time.Duration) FetchOpt

WithFetchHeartbeat sets a PullHeartbeat for the fetch request. Must be less than MaxWait; on miss, Fetch returns ErrNoHeartbeat.

func WithFetchMaxWait 

func WithFetchMaxWait(d time.Duration) FetchOpt

type FlightRecorder 

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

FlightRecorder is a ring buffer of incident events for ops storytelling. goalign:ignore

func NewFlightRecorder 

func NewFlightRecorder(capacity int) *FlightRecorder

NewFlightRecorder creates a recorder with the given capacity (default 128).

func (*FlightRecorder) AttachSoftLiveness 

func (r *FlightRecorder) AttachSoftLiveness(cfg *SoftLivenessConfig)

AttachSoftLiveness wraps cfg.OnStall to record stall incidents.

func (*FlightRecorder) AttachSupervisor 

func (r *FlightRecorder) AttachSupervisor(cfg *SupervisorConfig)

AttachSupervisor wraps cfg.OnEvent to record supervisor lifecycle events. On SupervisorGiveUp it auto-logs a compact dump.

func (*FlightRecorder) Events 

func (r *FlightRecorder) Events() <-chan IncidentEvent

Events is a live stream of newly recorded incidents (buffered; overflows dropped).

func (*FlightRecorder) LogSnapshot 

func (r *FlightRecorder) LogSnapshot(ctx context.Context, msg string)

LogSnapshot writes a compact slog dump of the current snapshot.

func (*FlightRecorder) Record 

func (r *FlightRecorder) Record(ev IncidentEvent)

Record appends an incident (drops oldest when full).

func (*FlightRecorder) RecordDLQ 

func (r *FlightRecorder) RecordDLQ(subject, stream, consumer, reason string, seq uint64)

RecordDLQ records a dead-letter route.

func (*FlightRecorder) RecordDLQAutopsy 

func (r *FlightRecorder) RecordDLQAutopsy(subject, stream, consumer, reason, errStr string, seq uint64)

RecordDLQAutopsy records a dead-letter route with forensic error detail.

func (*FlightRecorder) RecordReconnect 

func (r *FlightRecorder) RecordReconnect(detail string)

RecordReconnect records a reconnect spike for storytelling.

func (*FlightRecorder) Snapshot 

func (r *FlightRecorder) Snapshot() []IncidentEvent

Snapshot returns incidents in chronological order (oldest first).

func (*FlightRecorder) WriteJSON 

func (r *FlightRecorder) WriteJSON(w io.Writer) error

WriteJSON writes a Snapshot as JSON to w.

type IncidentCapture added in v0.6.0 

type IncidentCapture struct {
	Redact         func(*CapsuleMessage)
	FlightRecorder *FlightRecorder
	Fingerprint    *BehaviorFingerprintKVPayload
	Stream         string
	Consumer       string
	Subject        string
	Reason         string
	StoreBucket    string
	IndexBucket    string
	Trigger        IncidentTrigger
	FailingSeq     uint64
	Window         int
}

IncidentCapture configures Capsule assembly.

goalign:ignore

type IncidentEvent 

type IncidentEvent struct {
	Time       time.Time    `json:"time"`
	Detail     string       `json:"detail,omitempty"`
	Subject    string       `json:"subject,omitempty"`
	Stream     string       `json:"stream,omitempty"`
	Consumer   string       `json:"consumer,omitempty"`
	Reason     string       `json:"reason,omitempty"`
	Err        string       `json:"err,omitempty"`
	Sequence   uint64       `json:"sequence,omitempty"`
	NumPending uint64       `json:"num_pending,omitempty"`
	Attempt    int          `json:"attempt,omitempty"`
	Kind       IncidentKind `json:"kind"`
}

IncidentEvent is one entry in the flight-recorder timeline.

type IncidentKind 

type IncidentKind uint8

IncidentKind classifies flight-recorder events. 

const (
	IncidentSupervisor IncidentKind = iota + 1
	IncidentStall
	IncidentDLQ
	IncidentReconnect
	IncidentShadow
)

type IncidentTrigger added in v0.6.0 

type IncidentTrigger string

IncidentTrigger classifies why a capsule was captured. 

const (
	TriggerManual         IncidentTrigger = "manual"
	TriggerDLQ            IncidentTrigger = "dlq"
	TriggerAnomaly        IncidentTrigger = "anomaly"
	TriggerShadowMismatch IncidentTrigger = "shadow_mismatch"
	TriggerStall          IncidentTrigger = "stall"
)

func (IncidentTrigger) String added in v0.7.0 

func (it IncidentTrigger) String() string

type Incidents added in v0.6.0 

type Incidents interface {
	Capture(ctx context.Context, cfg IncidentCapture) (*Capsule, error)
	Load(ctx context.Context, bucket, id string) (*Capsule, error)
	// List returns capsule IDs for stream/consumer. Empty indexBucket uses DefaultIncidentIndexBucket.
	List(ctx context.Context, stream, consumer, indexBucket string) ([]string, error)
	ReplayLocal(ctx context.Context, capsule *Capsule, handler MsgHandler) error
	AttachToHandler(stream, durable string, cfg CapsuleAutoConfig, handler MsgHandler) MsgHandler
}

Incidents captures and replays incident capsules.

type KVBucketStatus 

type KVBucketStatus struct {
	Bucket  string
	Values  uint64
	History int64
}

KVBucketStatus summarizes a Key-Value bucket.

type KVEntry 

type KVEntry struct {
	Created  time.Time
	Bucket   string
	Key      string
	Value    []byte
	Revision uint64
}

KVEntry is a Key-Value entry with payload.

type KeyValueConfig 

type KeyValueConfig struct {
	Bucket      string
	Description string
	TTL         time.Duration
	MaxBytes    int64
	Storage     StorageType
	Replicas    int
	History     uint8 // defaults to 1 when zero
	Compression bool
}

KeyValueConfig configures a JetStream Key-Value bucket.

goalign:ignore

type KeyValueKeys 

type KeyValueKeys interface {
	ListKeys(ctx context.Context, bucket string, offset, limit int) ([]string, int, error)
	Get(ctx context.Context, bucket, key string) (*KVEntry, error)
	Put(ctx context.Context, bucket, key string, value []byte) (*KVEntry, error)
	DeleteKey(ctx context.Context, bucket, key string) error
	History(ctx context.Context, bucket, key string) ([]KVEntry, error)
}

KeyValueKeys provides key-level KV helpers (kept separate to avoid interfacebloat).

type KeyValueManager 

type KeyValueManager interface {
	CreateOrUpdate(ctx context.Context, cfg KeyValueConfig) (natspkg.KeyValue, error)
	CreateRaw(ctx context.Context, cfg *natspkg.KeyValueConfig) (KVBucketStatus, error)
	Open(ctx context.Context, bucket string) (natspkg.KeyValue, error)
	Delete(ctx context.Context, bucket string) error
	ListBuckets(ctx context.Context) ([]KVBucketStatus, error)
	BucketInfo(ctx context.Context, bucket string) (KVBucketStatus, error)
}

KeyValueManager manages JetStream Key-Value buckets.

type Message 

type Message struct {
	Data   any
	Header map[string][]string
	// Expect applies optimistic concurrency PubOpts (ExpectedStream / LastSeq / LastMsgId).
	Expect      *PublishExpectation
	MessageType MessageType
}

func (Message) WithExpectedLastMsgID 

func (m Message) WithExpectedLastMsgID(id string) Message

WithExpectedLastMsgID requires the stream's last message ID to equal id.

func (Message) WithExpectedLastSeq 

func (m Message) WithExpectedLastSeq(seq uint64) Message

WithExpectedLastSeq requires the stream's last sequence to equal seq before accept.

func (Message) WithExpectedLastSeqPerSubject 

func (m Message) WithExpectedLastSeqPerSubject(seq uint64) Message

WithExpectedLastSeqPerSubject requires the subject's last sequence to equal seq.

func (Message) WithExpectedStream 

func (m Message) WithExpectedStream(stream string) Message

WithExpectedStream requires the message to land in the named stream.

func (Message) WithMsgID 

func (m Message) WithMsgID(id string) Message

WithMsgID sets the JetStream deduplication header (requires stream DuplicateWindow).

Example

ExampleMessage_WithMsgID demonstrates publish deduplication header. 

_ = Message{Data: map[string]string{"id": "1"}, MessageType: JSON}.WithMsgID("pay-123")

type MessageType 

type MessageType int
const (
	JSON        MessageType = 1
	Proto       MessageType = 2
	MessagePack MessageType = 3
	// Raw publishes Data as []byte without encoding (used by DLQ passthrough).
	Raw MessageType = 4
)

func MessageTypeFromHeader 

func MessageTypeFromHeader(h natspkg.Header) MessageType

type MetricsConfig 

type MetricsConfig struct {
	Prefix           string
	TrackedStreams   []string
	TrackedConsumers []TrackedConsumer
	CollectInterval  time.Duration
	AllowMetrics     bool
	AllowTracing     bool
	// Lite registers only publish/consume/ack counters (skips JetStream gauges and RTT).
	Lite bool
	// FixedCardinality records hot-path metrics without per-subject attributes.
	FixedCardinality bool
}

MetricsConfig holds JetStream metrics collection options.

goalign:ignore

type Monitoring 

type Monitoring interface {
	// Fetch GETs baseURL+path. baseURL is the per-cluster monitoring root
	// (e.g. http://127.0.0.1:8222); path is typically "/jsz" or "/varz".
	Fetch(ctx context.Context, baseURL, path string) ([]byte, error)
}

Monitoring fetches NATS server monitoring HTTP endpoints (/varz, /jsz, …).

type MsgHandler 

type MsgHandler func(context.Context, *natspkg.Msg) error

func AttachIncidentsHandler added in v0.7.0 

func AttachIncidentsHandler(inc Incidents, stream, durable string, cfg CapsuleAutoConfig, handler MsgHandler) MsgHandler

AttachIncidentsHandler wraps handler using any Incidents implementation.

func HandlerTyped 

func HandlerTyped[T any](mt MessageType, fn func(ctx context.Context, subject string, payload T) error) MsgHandler

HandlerTyped wraps a typed handler with decode logic.

func WithDLQ 

func WithDLQ(cfg DLQConfig, handler MsgHandler) MsgHandler

WithDLQ wraps a handler so poison messages are published to a DLQ subject and Term'd. Prefer importing github.com/gopherust-io/nats/dlq for new code.

func WithShadow 

func WithShadow(cfg ShadowConfig, primary, shadowHandler MsgHandler) MsgHandler

WithShadow runs primary for delivery fate and optionally a shadow handler. Prefer importing github.com/gopherust-io/nats/shadow for new code.

type MsgRangeOpt added in v0.3.0 

type MsgRangeOpt func(*msgRangeConfig)

MsgRangeOpt configures GetMsgRange / GetMsgRangeByTime.

func WithMaxMessages added in v0.3.0 

func WithMaxMessages(n int) MsgRangeOpt

WithMaxMessages caps how many messages a range fetch may return (default DefaultMsgRangeMax).

type ObjectBucketStatus 

type ObjectBucketStatus struct {
	Bucket      string
	Description string
	Size        uint64
}

ObjectBucketStatus summarizes an Object Store bucket.

type ObjectEntry 

type ObjectEntry struct {
	Modified time.Time
	Bucket   string
	Name     string
	Data     []byte
	Size     uint64
}

ObjectEntry is an object payload plus metadata.

type ObjectStoreConfig 

type ObjectStoreConfig struct {
	Bucket      string
	Description string
	TTL         time.Duration
	MaxBytes    int64
	Storage     StorageType
	Replicas    int
}

ObjectStoreConfig configures a JetStream Object Store bucket.

type ObjectStoreManager 

type ObjectStoreManager interface {
	ListBuckets(ctx context.Context) ([]ObjectBucketStatus, error)
	Create(ctx context.Context, cfg ObjectStoreConfig) (ObjectBucketStatus, error)
	CreateRaw(ctx context.Context, cfg *natspkg.ObjectStoreConfig) (ObjectBucketStatus, error)
	BucketInfo(ctx context.Context, bucket string) (ObjectBucketStatus, error)
	Delete(ctx context.Context, bucket string) error
	ListObjects(ctx context.Context, bucket string, offset, limit int) ([]string, int, error)
	Get(ctx context.Context, bucket, name string) (*ObjectEntry, error)
	Put(ctx context.Context, bucket, name string, data []byte) (*ObjectEntry, error)
	DeleteObject(ctx context.Context, bucket, name string) error
}

ObjectStoreManager manages JetStream Object Store buckets and objects.

type OrderedConsumerConfig added in v0.7.0 

type OrderedConsumerConfig struct {
	FilterSubjects    []string
	FilterSubject     string
	OptStartTime      *time.Time
	NamePrefix        string
	InactiveThreshold time.Duration
	OptStartSeq       uint64
	DeliverPolicy     DeliverPolicy
	ReplayPolicy      ReplayPolicy
	HasDeliverPolicy  bool
	HasReplayPolicy   bool
	HeadersOnly       bool
}

OrderedConsumerConfig configures a client-managed ordered consumer.

goalign:ignore

type OrderedSubscription added in v0.7.0 

type OrderedSubscription interface {
	Stop()
	Stopped() <-chan struct{}
}

OrderedSubscription is a running ordered consume loop.

type PayloadCompressionMode added in v0.6.0 

type PayloadCompressionMode uint8

PayloadCompressionMode selects publisher-side payload compression. 

const (
	// PayloadCompressionOff never compresses (default).
	PayloadCompressionOff PayloadCompressionMode = iota
	// PayloadCompressionAuto compresses payloads strictly larger than
	// MinPayloadCompressBytes with br then gzip when the result shrinks.
	PayloadCompressionAuto
	// PayloadCompressionGzip forces gzip when the payload is above the threshold
	// and the compressed body is smaller.
	PayloadCompressionGzip
	// PayloadCompressionBrotli forces brotli (br) when the payload is above the
	// threshold and the compressed body is smaller.
	PayloadCompressionBrotli
)

type ProcessActivity 

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

ProcessActivity tracks the last successful message process (Ack or DLQ Term).

func NewProcessActivity 

func NewProcessActivity() *ProcessActivity

NewProcessActivity creates a tracker seeded to now.

func (*ProcessActivity) LastSuccess 

func (a *ProcessActivity) LastSuccess() time.Time

LastSuccess returns the time of the last successful process.

func (*ProcessActivity) Touch 

func (a *ProcessActivity) Touch()

Touch records a successful process at the current time.

type ProcessOpt 

type ProcessOpt func(*processConfig)

func WithFetchBatch 

func WithFetchBatch(batch int) ProcessOpt

func WithProcessConcurrency 

func WithProcessConcurrency(n int) ProcessOpt

WithProcessConcurrency sets parallel handler goroutines per fetched batch (default 1).

func WithProcessHeartbeat 

func WithProcessHeartbeat(d time.Duration) ProcessOpt

WithProcessHeartbeat sets PullHeartbeat used by Process fetch loops. When unset, Process uses RuntimeConsumerConfig.IdleHeartbeat if > 0.

func WithProcessMaxWait 

func WithProcessMaxWait(d time.Duration) ProcessOpt

type PubAck 

type PubAck = natspkg.PubAck

PubAck is a JetStream publish acknowledgement.

type PubAckFuture 

type PubAckFuture = natspkg.PubAckFuture

PubAckFuture is a future for an async JetStream publish ack.

type PublishExpectation 

type PublishExpectation struct {
	LastSeq           *uint64
	LastSeqPerSubject *uint64
	Stream            string
	LastMsgID         string
}

PublishExpectation is optimistic concurrency for JetStream publish. Zero LastSeq / LastSeqPerSubject are ignored unless the corresponding Set* helper is used (pointers distinguish "unset" from "expect zero").

type Publisher 

type Publisher interface {
	PublishMessage(ctx context.Context, subject string, msg Message) error
	PublishProto(ctx context.Context, subject string, payload proto.Message) error
	PublishJSON(ctx context.Context, subject string, data any) error
	PublishMsgPack(ctx context.Context, subject string, data any) error
	PublishBytes(ctx context.Context, subject string, data []byte) error
	PublishBytesWithMsgID(ctx context.Context, subject, id string, data []byte) error
	PublishWithMsgID(ctx context.Context, subject, id string, msg Message) error
	PublishAsync(ctx context.Context, subject string, msg Message) (PubAckFuture, error)
	PublishAsyncBytes(ctx context.Context, subject string, data []byte) (PubAckFuture, error)
	PublishAsyncComplete(ctx context.Context) error
}

type PublisherConfig 

type PublisherConfig struct {
	MetricPrefix string
	AllowMetrics bool
	AllowTracing bool
	// SkipSubjectValidation skips per-publish subject validation for trusted static subjects.
	SkipSubjectValidation bool
	// MaxAsyncPending caps in-flight PublishAsync requests (0 = defaultMaxAsyncPending, -1 = unlimited).
	MaxAsyncPending int
	// PayloadCompression opt-in auto compression (br→gzip) for payloads >32 KiB.
	PayloadCompression PayloadCompressionMode
}

PublisherConfig holds publish metrics and async pending limits.

goalign:ignore

type PullConsumer 

type PullConsumer interface {
	Fetch(ctx context.Context, batch int, opts ...FetchOpt) ([]*natspkg.Msg, error)
	FetchNoWait(batch int) ([]*natspkg.Msg, error)
	Process(ctx context.Context, handler MsgHandler, opts ...ProcessOpt) error
	Close() error
}

type PurgeOpt 

type PurgeOpt func(*natspkg.StreamPurgeRequest)

func PurgeSubject 

func PurgeSubject(subject string) PurgeOpt

type Replay 

type Replay interface {
	// ResetConsumer seeks an existing durable to a new deliver position while
	// preserving ack limits, filters, and other consumer settings when present.
	ResetConsumer(ctx context.Context, stream, durable string, opts ...ReplayOpt) (ReplayConsumerResult, error)
	// CreateReplayConsumer creates a side-car durable for backfill. The live
	// sourceDurable is left untouched. When WithReplayDurable is omitted, a
	// unique name is generated from sourceDurable.
	CreateReplayConsumer(ctx context.Context, stream, sourceDurable string, opts ...ReplayOpt) (ReplayConsumerResult, error)
	GetMsg(ctx context.Context, stream string, seq uint64) (*StoredMessage, error)
	GetLastMsgForSubject(ctx context.Context, stream, subject string) (*StoredMessage, error)
	GetNextMsgAfter(ctx context.Context, stream string, seq uint64) (*StoredMessage, error)
	GetMsgRange(ctx context.Context, stream string, startSeq, endSeq uint64, opts ...MsgRangeOpt) ([]*StoredMessage, bool, error)
	GetMsgRangeByTime(ctx context.Context, stream string, start, end time.Time, opts ...MsgRangeOpt) ([]*StoredMessage, bool, error)
	FindFirstSeqAtOrAfter(ctx context.Context, stream string, t time.Time) (uint64, error)
	FindLastSeqAtOrBefore(ctx context.Context, stream string, t time.Time) (uint64, error)
}

type ReplayConfig 

type ReplayConfig struct {
	OptStartTime   *time.Time
	UntilTime      *time.Time
	Durable        string // target durable for CreateReplayConsumer; ignored by ResetConsumer
	FilterSubject  string
	FilterSubjects []string
	DeliverPolicy  DeliverPolicy
	ReplayPolicy   ReplayPolicy
	OptStartSeq    uint64
	UntilSeq       uint64
	Limit          int
	// contains filtered or unexported fields
}

ReplayConfig holds replay seek settings. Prefer ReplayOpt helpers (FromSeq, FromTime, WithReplayPolicy, …) at call sites.

Zero-value JetStream enums (DeliverAll, ReplayInstant, AckNone) are valid; set flags track whether an option explicitly chose a policy or start position.

goalign:ignore

type ReplayConsumerResult added in v0.3.0 

type ReplayConsumerResult struct {
	StartTime *time.Time
	UntilTime *time.Time
	Durable   string
	StartSeq  uint64
	UntilSeq  uint64
	Limit     int
}

ReplayConsumerResult is returned after ResetConsumer / CreateReplayConsumer.

type ReplayOpt 

type ReplayOpt func(*ReplayConfig)

ReplayOpt configures ResetConsumer / CreateReplayConsumer.

func FromBeginning 

func FromBeginning() ReplayOpt

FromBeginning replays from the first available message (DeliverAll).

func FromNew 

func FromNew() ReplayOpt

FromNew only delivers messages published after the consumer is (re)created.

func FromSeq 

func FromSeq(seq uint64) ReplayOpt

FromSeq seeks by stream sequence (DeliverByStartSequence + OptStartSeq).

func FromTime 

func FromTime(t time.Time) ReplayOpt

FromTime seeks by start timestamp (DeliverByStartTime + OptStartTime).

func Limit added in v0.3.0 

func Limit(n int) ReplayOpt

Limit records a max message count bound for the intended replay window.

func OneMessage added in v0.3.0 

func OneMessage(seq uint64) ReplayOpt

OneMessage seeks to deliver exactly one stored message at seq.

func UntilSeq added in v0.3.0 

func UntilSeq(seq uint64) ReplayOpt

UntilSeq records an inclusive end sequence bound (client/metadata; JetStream has no server end).

func UntilTime added in v0.3.0 

func UntilTime(t time.Time) ReplayOpt

UntilTime records an inclusive end timestamp bound (resolved to seq when seeking when possible).

func WithDeliverPolicy 

func WithDeliverPolicy(policy DeliverPolicy) ReplayOpt

WithDeliverPolicy sets an explicit deliver policy (including DeliverAll).

func WithFilterSubject 

func WithFilterSubject(subject string) ReplayOpt

WithFilterSubject sets a single filter subject for the replay consumer.

func WithFilterSubjects 

func WithFilterSubjects(subjects ...string) ReplayOpt

WithFilterSubjects sets multi-filter subjects for the replay consumer.

func WithReplayDurable 

func WithReplayDurable(name string) ReplayOpt

WithReplayDurable sets the side-car durable name for CreateReplayConsumer. Ignored by ResetConsumer (which uses the durable argument).

func WithReplayPolicy 

func WithReplayPolicy(policy ReplayPolicy) ReplayOpt

WithReplayPolicy sets Instant vs Original timing replay (including ReplayInstant).

func WithStartSeq 

func WithStartSeq(seq uint64) ReplayOpt

WithStartSeq sets OptStartSeq (does not change DeliverPolicy by itself).

func WithStartTime 

func WithStartTime(t time.Time) ReplayOpt

WithStartTime sets OptStartTime (does not change DeliverPolicy by itself).

type ReplayPolicy 

type ReplayPolicy = natspkg.ReplayPolicy

Deliver and replay policies mirror github.com/nats-io/nats.go constants.

type Requester 

type Requester interface {
	RequestBytes(ctx context.Context, subject string, data []byte) (*natspkg.Msg, error)
	RequestMessage(ctx context.Context, subject string, msg Message) (*natspkg.Msg, error)

	RequestJSON(ctx context.Context, subject string, req any) (*natspkg.Msg, error)
	RequestJSONInto(ctx context.Context, subject string, req, resp any) error

	RequestMsgPack(ctx context.Context, subject string, req any) (*natspkg.Msg, error)
	RequestMsgPackInto(ctx context.Context, subject string, req, resp any) error

	RequestProto(ctx context.Context, subject string, req proto.Message) (*natspkg.Msg, error)
	RequestProtoInto(ctx context.Context, subject string, req, resp proto.Message) error
}

Requester issues core NATS request/reply calls (not JetStream).

type RequesterConfig 

type RequesterConfig struct {
	MetricPrefix string
	// Timeout is used when ctx has no deadline (0 = defaultRequestTimeout).
	Timeout      time.Duration
	AllowMetrics bool
	AllowTracing bool
	// SkipSubjectValidation skips per-request subject validation for trusted static subjects.
	SkipSubjectValidation bool
	// PayloadCompression opt-in compression for outbound requests (>32 KiB, shrink-only).
	PayloadCompression PayloadCompressionMode
	// PayloadDecompression expands Content-Encoding on replies before return / Into (default on).
	PayloadDecompression bool
}

RequesterConfig configures core NATS request/reply client calls.

goalign:ignore

type Responder 

type Responder interface {
	Subscribe(ctx context.Context, subject string, handler MsgHandler) (Subscription, error)
	QueueSubscribe(ctx context.Context, queue, subject string, handler MsgHandler) (Subscription, error)

	RespondBytes(msg *natspkg.Msg, data []byte) error
	RespondJSON(msg *natspkg.Msg, v any) error
	RespondMsgPack(msg *natspkg.Msg, v any) error
	RespondProto(msg *natspkg.Msg, v proto.Message) error
}

Responder registers core NATS reply handlers (not JetStream; no auto Ack/Nak). Respond* methods apply ResponderConfig.PayloadCompression; package-level Respond* helpers leave payloads uncompressed.

type ResponderConfig 

type ResponderConfig struct {
	MetricPrefix string
	AllowMetrics bool
	AllowTracing bool
	// PayloadCompression opt-in compression for Responder.Respond* replies (>32 KiB, shrink-only).
	PayloadCompression PayloadCompressionMode
	// PayloadDecompression expands Content-Encoding on inbound requests before the handler (default on).
	PayloadDecompression bool
}

ResponderConfig configures core NATS reply subscribers.

goalign:ignore

type RetentionPolicy 

type RetentionPolicy = natspkg.RetentionPolicy

Deliver and replay policies mirror github.com/nats-io/nats.go constants.

type RuntimeConsumerConfig 

type RuntimeConsumerConfig struct {
	MetricPrefix      string
	PendingMsgLimit   int
	PendingMsgBuffer  int
	WorkerPoolSize    int
	WorkerBufferSize  int
	AckWait           time.Duration
	IdleHeartbeat     time.Duration // 0 disables; unsupported on queue groups
	WorkerPoolEnabled bool
	FlowControl       bool
	AllowMetrics      bool
	AllowTracing      bool
	// PayloadDecompression expands Content-Encoding (br/gzip) before the handler
	// sees msg.Data. DefaultConfig enables this so compressed publishers interoperate.
	PayloadDecompression bool
}

RuntimeConsumerConfig holds push/pull worker and heartbeat settings.

goalign:ignore

type SchedulePublishConfig added in v0.7.0 

type SchedulePublishConfig struct {
	// SubjectTemplate uses %s for base subject and %d for delay milliseconds.
	// Default: "%s.delay.%d"
	SubjectTemplate string
}

SchedulePublishConfig delays delivery via a schedule subject convention. Default: publish to "{subject}.delay.{ms}" so an ops-owned stream/consumer can forward after the delay (or use a server-side scheduler when available).

goalign:ignore

type SchemaExpectation added in v0.7.0 

type SchemaExpectation struct {
	Name     string
	Version  string
	Encoding string // optional; when empty, Content-Type / Nats-Content-Type is used
	Required bool
}

SchemaExpectation validates optional contract headers on decode.

goalign:ignore

type ShadowConfig 

type ShadowConfig struct {
	// Compare returns true when primary and shadow outcomes match.
	// If nil, outcomes match when both succeed or both fail (nil-ness only).
	Compare func(primaryErr, shadowErr error) bool
	// Recorder receives IncidentShadow events on panic / mismatch (optional).
	Recorder *FlightRecorder

	// SampleRate is the fraction of messages sent to the shadow in (0,1].
	// Zero (default) means 0.1 (10% canary). Set explicitly to 1 for always-on dual-run.
	SampleRate float64
	// contains filtered or unexported fields
}

ShadowConfig configures WithShadow.

type SlowConsumer added in v0.3.0 

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

SlowConsumer watches a subscription for sustained JetStream backlog thresholds. goalign:ignore

func WatchSlowConsumer added in v0.3.0 

func WatchSlowConsumer(
	ctx context.Context,
	sub Subscription,
	streamLastSeq StreamLastSeqFunc,
	cfg SlowConsumerConfig,
	metrics *clientMetrics,
) (*SlowConsumer, error)

WatchSlowConsumer starts polling sub.ConsumerInfo for sustained threshold breaches. streamLastSeq may be nil; lag is then treated as 0 (pending / ack-pending still apply).

func (*SlowConsumer) Events added in v0.3.0 

func (s *SlowConsumer) Events() <-chan SlowConsumerEvent

Events returns slow-consumer notifications (buffered; overflows are dropped).

func (*SlowConsumer) Slow added in v0.3.0 

func (s *SlowConsumer) Slow() bool

Slow reports whether a slow condition has been detected (sticky until Stop).

func (*SlowConsumer) Stop added in v0.3.0 

func (s *SlowConsumer) Stop()

Stop ends the watch loop.

type SlowConsumerConfig added in v0.3.0 

type SlowConsumerConfig struct {
	OnSlow func(SlowConsumerEvent)
	// PollInterval is how often ConsumerInfo is polled (default 2s).
	PollInterval time.Duration
	// SustainFor is how long thresholds must hold before firing (default 30s).
	SustainFor time.Duration
	// PendingThreshold fires when NumPending >= this value (default 1000).
	PendingThreshold uint64
	// LagThreshold fires when stream tip − delivered stream seq >= this value (default 1000).
	LagThreshold uint64
	// AckPendingRatio fires when NumAckPending >= ratio * MaxAckPending (default 0.9).
	// Ignored when MaxAckPending <= 0.
	AckPendingRatio float64
	// CircuitStop stops the watcher after the first slow event.
	CircuitStop bool
}

SlowConsumerConfig controls WatchSlowConsumer threshold detection. goalign:ignore

type SlowConsumerEvent added in v0.3.0 

type SlowConsumerEvent struct {
	Stream        string
	Durable       string
	Reasons       []string
	Pending       uint64
	Lag           uint64
	AckPending    int
	MaxAckPending int
	SustainedFor  time.Duration
}

SlowConsumerEvent is emitted when backlog thresholds are sustained.

type SoftLiveness 

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

SoftLiveness watches a subscription for backlog growth without processing activity. goalign:ignore

func WatchSoftLiveness 

func WatchSoftLiveness(
	ctx context.Context,
	sub Subscription,
	activity *ProcessActivity,
	cfg SoftLivenessConfig,
	metrics *clientMetrics,
) (*SoftLiveness, error)

WatchSoftLiveness starts polling sub.ConsumerInfo for stall conditions. Register activity via SoftLiveness.Activity() and consumer.OnProcessSuccess, or call Touch manually.

func (*SoftLiveness) Activity 

func (s *SoftLiveness) Activity() *ProcessActivity

Activity returns the process-activity tracker used by this watcher.

func (*SoftLiveness) Events 

func (s *SoftLiveness) Events() <-chan SoftLivenessEvent

Events returns stall notifications (buffered; overflows are dropped).

func (*SoftLiveness) Stalled 

func (s *SoftLiveness) Stalled() bool

Stalled reports whether a stall has been detected (sticky until Stop).

func (*SoftLiveness) Stop 

func (s *SoftLiveness) Stop()

Stop ends the watch loop.

type SoftLivenessConfig 

type SoftLivenessConfig struct {
	OnStall func(SoftLivenessEvent)
	// PollInterval is how often ConsumerInfo is polled (default 2s).
	PollInterval time.Duration
	// StallAfter is the minimum time since last successful process before a stall can fire (default 10s).
	StallAfter time.Duration
	// RisingWindows is consecutive polls with rising NumPending required (default 3).
	RisingWindows int
	// CircuitStop stops the watcher after the first stall (apps may then restart the pod).
	CircuitStop bool
}

SoftLivenessConfig controls WatchSoftLiveness. goalign:ignore

type SoftLivenessEvent 

type SoftLivenessEvent struct {
	Err        error
	NumPending uint64
	StalledFor time.Duration
}

SoftLivenessEvent is emitted when backlog grows without successful processing.

type StorageType 

type StorageType = natspkg.StorageType

Deliver and replay policies mirror github.com/nats-io/nats.go constants.

type StoredMessage added in v0.3.0 

type StoredMessage struct {
	Time        time.Time
	Header      map[string][]string
	Subject     string
	Data        []byte
	Sequence    uint64
	MessageType MessageType
}

StoredMessage is a JetStream stream message loaded by sequence (peek / range export).

type StreamConfig 

type StreamConfig struct {
	Mirror          *StreamSource
	Name            string
	Description     string
	Subjects        []string
	Sources         []*StreamSource
	Storage         StorageType
	MaxBytes        int64
	MaxAge          time.Duration
	MaxMsgs         int64
	Discard         DiscardPolicy
	DuplicateWindow time.Duration
	MaxConsumers    int
	Retention       RetentionPolicy
	Replicas        int
	MaxMsgSize      int32
	NoAck           bool
}

StreamConfig holds JetStream stream topology fields for explicit provisioning.

goalign:ignore

type StreamLastSeqFunc added in v0.3.0 

type StreamLastSeqFunc func(ctx context.Context, stream string) (uint64, error)

StreamLastSeqFunc returns the stream tip sequence for lag calculation.

type StreamManager 

type StreamManager interface {
	CreateOrUpdateStream(ctx context.Context, cfg StreamConfig) (*natspkg.StreamInfo, error)
	AddStream(ctx context.Context, cfg *natspkg.StreamConfig) (*natspkg.StreamInfo, error)
	UpdateStream(ctx context.Context, cfg *natspkg.StreamConfig) (*natspkg.StreamInfo, error)
	DeleteStream(ctx context.Context, name string) error
	StreamInfo(ctx context.Context, name string) (*natspkg.StreamInfo, error)
	ListStreamsPage(ctx context.Context, offset, limit int) ([]*natspkg.StreamInfo, int, error)
	PurgeStream(ctx context.Context, name string, opts ...PurgeOpt) error
	GetMsg(ctx context.Context, stream string, seq uint64) (*natspkg.RawStreamMsg, error)
	GetLastMsg(ctx context.Context, stream, subject string) (*natspkg.RawStreamMsg, error)
	GetNextMsgAfter(ctx context.Context, stream string, seq uint64) (*natspkg.RawStreamMsg, error)
}

StreamManager manages JetStream streams (≤10 methods for interfacebloat). StreamNames is available via ListStreamsPage; ListStreams is a convenience wrapper implemented outside the interface.

type StreamSource 

type StreamSource = natspkg.StreamSource

StreamSource is a JetStream stream source or mirror origin.

type SubscribeFn 

type SubscribeFn func(ctx context.Context) (Subscription, error)

SubscribeFn creates a push subscription (e.g. QueueSubscribeBound).

type Subscription 

type Subscription interface {
	Unsubscribe() error
	Drain() error
	IsValid() bool
	Subject() string
	SetPendingLimits(msgLimit, bytesLimit int) error
	ConsumerInfo() (*natspkg.ConsumerInfo, error)
	Type() natspkg.SubscriptionType
}

type SupervisedSubscription 

type SupervisedSubscription interface {
	Subscription
	Events() <-chan SupervisorEvent
	Stop() error
}

SupervisedSubscription is a push subscription that auto-resubscribes when invalid.

func Supervise 

func Supervise(
	ctx context.Context,
	cfg SupervisorConfig,
	metrics *clientMetrics,
	subscribe SubscribeFn,
) (SupervisedSubscription, error)

Supervise runs subscribe and keeps the subscription alive by resubscribing when IsValid() becomes false (e.g. after idle-heartbeat miss / ErrConsumerNotActive).

type SupervisorConfig 

type SupervisorConfig struct {
	OnEvent        func(SupervisorEvent)
	InitialBackoff time.Duration
	MaxBackoff     time.Duration
	CheckInterval  time.Duration
	// HealthyBackoffInterval is the maximum poll interval after consecutive healthy checks (0 disables).
	HealthyBackoffInterval time.Duration
	// HealthyChecksBeforeBackoff is the number of healthy checks before increasing the poll interval.
	HealthyChecksBeforeBackoff int
	// MaxRetries is resubscribe attempts after interest is lost.
	// 0 means default (10). -1 means unlimited.
	MaxRetries int
}

type SupervisorEvent 

type SupervisorEvent struct {
	Err     error
	Kind    SupervisorEventKind
	Attempt int
}

SupervisorEvent is emitted when a supervised subscription is healed or abandoned. goalign:ignore

type SupervisorEventKind 

type SupervisorEventKind uint8

SupervisorEventKind classifies supervisor lifecycle events. 

const (
	SupervisorResubscribed SupervisorEventKind = iota + 1
	SupervisorGiveUp
	SupervisorInvalid
)

type TrackedConsumer 

type TrackedConsumer struct {
	Stream  string
	Durable string
}

type WorkerSetup 

type WorkerSetup struct {
	Queue    string
	Subject  string
	Stream   StreamConfig
	Consumer DurableConsumerConfig
}

WorkerSetup configures a one-shot stream + durable + queue subscribe.

Example

ExampleWorkerSetup documents one-shot worker setup fields. 

_ = WorkerSetup{
	Stream:   StreamConfig{Name: "ORDERS", Subjects: []string{"orders.>"}},
	Consumer: DurableConsumerConfig{Durable: "orders-processor", FilterSubject: "orders.>"},
	Queue:    "orders-workers",
	Subject:  "orders.>",
}

Source Files

View all Source files

Directories

Path Synopsis
examples
nats command
internal
bytesconv
Package bytesconv provides zero-allocation string↔[]byte conversions.
 Package bytesconv provides zero-allocation string↔[]byte conversions.
tools
loadtest command
Command loadtest publishes and consumes JetStream messages for CPU/memory profiling.
 Command loadtest publishes and consumes JetStream messages for CPU/memory profiling.

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.