nuts

README

go-nuts

A small, dependency-light toolkit over nats.go for building NATS and JetStream services in Go. The root package handles connection lifecycle — connecting, draining, and shutting down cleanly; opt-in subpackages add a durable JetStream pull-consumer scaffold, a publish helper, trace-context propagation, and a redelivery policy.

Install

go get github.com/entireio/go-nuts

Connection lifecycle (root package)

• Connect — dial with a resilient default posture: reconnect-forever (so a long-lived service rides out a NATS outage instead of permanently closing after the client's default 60 attempts), a bounded drain timeout, optional rotation-aware mTLS, and lifecycle log handlers that don't mistake a clean shutdown for a fault. Async errors (slow-consumer drops, permissions violations) and lame-duck notices are logged through the configured logger instead of nats.go's stderr default.
• Drain / IsShutdownFetchErr — graceful shutdown. Drain starts nats.go's asynchronous drain and blocks until the connection flushes and closes (bounded by a timeout), returning an error and force-closing when it cannot complete; IsShutdownFetchErr lets a fetch loop treat a drain/close during shutdown as a clean exit rather than an error.
• ShutdownGroup — cancels tracked background loops, waits for them to return, then drains the registered connections — the ordering that keeps a Drain from racing an in-flight Fetch. A panic or unexpected return in a tracked loop cancels the group and is returned by Shutdown, so a dead consumer cannot leave its process looking healthy.

Usage

The example below drives a hand-rolled fetch loop on nats.go's legacy pull API; consumers on the modern jetstream API get the whole loop from jsconsumer instead.

ctx, stop := signal.NotifyContext(context.Background(), os.Interrupt, syscall.SIGTERM)
defer stop()

g := nuts.NewShutdownGroup(ctx)
nc, err := nuts.Connect(g.Context(), natsURL, nuts.WithName("worker"))
if err != nil {
	return err
}
g.AddConn("worker", nc)

g.Go(func(ctx context.Context) {
	for {
		// Stop fetching once the group is shutting down, so the loop returns
		// before its connection is drained (Shutdown cancels, joins, then
		// drains — the loop must observe the cancel to keep that ordering).
		if ctx.Err() != nil {
			return
		}
		msgs, err := sub.Fetch(1, nats.MaxWait(5*time.Second))
		switch {
		case errors.Is(err, nats.ErrTimeout):
			continue // idle poll
		case nuts.IsShutdownFetchErr(ctx, err):
			return // clean shutdown: connection drained/closed
		case err != nil:
			// Real error — log/metric, then back off so a fast-failing Fetch
			// (e.g. connection closed mid-run) does not busy-spin the CPU.
			slog.ErrorContext(ctx, "fetch failed", slog.Any("error", err))
			select {
			case <-ctx.Done():
			case <-time.After(time.Second):
			}
			continue
		}
		handle(msgs)
	}
})

<-g.Context().Done() // process signal, or a tracked loop failed
if err := g.Shutdown(); err != nil { // cancel loops → join → drain, in that order
	return err
}

By default Connect builds mTLS from the ENTIRE_INTERNAL_TLS_{CERT,KEY,CA}_FILE environment variables; use WithTLSConfig, WithoutTLS, or the other Options to override.

Subpackages

The root package is deliberately nats.go-only. The consumer-side layers live in subpackages so importing the root links none of their dependencies (natsmsg and jsconsumer add the OpenTelemetry API; all three use nats.go/jetstream):

• natsmsg — small JetStream message helpers: W3C trace-context propagation over message headers (Inject / ExtractHeader / StartConsumerSpan, so publish → consume stitches into one trace), Publisher — the publish core (producer span with a caller-selected Operation name + standard messaging.* attributes + trace inject + Nats-Msg-Id dedup + bounded pub-ack wait + PubAck telemetry, with StartProducerSpan for callers composing by hand); it owns just that prologue — subject construction, payload encoding, domain metrics/logging, and the response to a failed publish stay with the caller. LegacyPublisher is the same core over the legacy nats.JetStreamContext API, a transitional bridge while callers migrate to the modern jetstream.JetStream. Also KeepInProgress, an AckWait heartbeat for long handlers, capped so a wedged handler still redelivers, and DeadLetter / SubjectToken, the dead-letter capture that copies a poison message to a DLQ subject with Nats-Dlq-* provenance before a consumer gives up on it (the capture step backoff's Term-on-exhaustion below expects). natsmsg/natsmsgtest ships FakeMsg / FakeLegacyMsg, scriptable modern and legacy messages for asserting a consumer's ack/nak/term disposition without a broker.
• jsconsumer — the durable JetStream pull-consumer scaffold: Start (create-or-update durable → consume → stop on context cancel), Run (Start under supervision: recreate on a closed loop with exponential backoff, tolerate a not-yet-provisioned stream at boot), and Process (consumer span re-parented across the NATS hop → decode → Term-on-undecodable → dispatch to the handler, which owns the message's disposition). AckExplicit, bounded AckWait and MaxDeliver, optional InactiveThreshold / MaxAckPending, shutdown-aware consume-error logging, optional KeepInProgress heartbeat.
• backoff — the redelivery policy for transiently-failed deliveries: a NakWithDelay envelope — flat by default, optionally growing per delivery (Factor/MaxDelay) — bounded by MaxDeliver, with opt-in Term-on-final-delivery so a work-queue message is removed cleanly instead of orphaning un-acked. It owns disposition and delay calculation only, and drives both the modern jetstream.Msg API (NakOrTerm) and the legacy *nats.Msg API (NakOrTermLegacy). MaxDeliver mirrors jetstream.ConsumerConfig: a non-positive value (0 or UnlimitedMaxDeliver) means unlimited redeliveries, so bridge an unset jsconsumer consumer with its EffectiveMaxDeliver() rather than the raw field.

Development

Requires Go 1.26. Tasks via mise: mise run test, mise run lint, mise run fmt.

License

MIT — see LICENSE.

Documentation

Overview

Package nuts provides shared NATS connection-lifecycle helpers for Entire services: a rotation-aware mTLS Connect with the org-standard resiliency options, a Drain that waits for a connection to flush and close on graceful shutdown, an IsShutdownFetchErr classifier so pull-consumer loops can treat a drain/close during shutdown as a clean exit rather than an error, and a ShutdownGroup that cancels background loops, joins them, and only then drains the connections — the ordering that keeps a Drain from racing an in-flight Fetch (COR-923).

This root package is the connection-lifecycle core extracted from entire-core (entiredb), entire-api, and mirror-pipeline, which had each grown their own copy of this plumbing (COR-925). It depends only on nats.go.

The subpackages carry the message-layer pieces those services shared (COR-929, deepened in COR-984) — natsmsg (trace-context propagation over message headers, the bounded deduped Publisher — modern, with LegacyPublisher bridging the legacy nats.JetStreamContext API and a caller-selected Operation span name — the KeepInProgress AckWait heartbeat, the DeadLetter dead-letter capture, and the natsmsgtest.FakeMsg / FakeLegacyMsg test doubles), jsconsumer (the durable JetStream pull-consumer scaffold, one-shot via Start or supervised via Run), and backoff (the NakWithDelay redelivery policy — flat or growing — with Term-on-final-delivery, COR-762, driving both the modern jetstream.Msg and the legacy *nats.Msg APIs). Both seams keep the modern API as the destination and expose the legacy path as a transitional bridge that deletes cleanly once no legacy caller remains. natsmsg and jsconsumer additionally depend on the OpenTelemetry API, and all three on nats.go's jetstream package; importing only the root package links none of that.

Index

• Constants
• Variables
• func Connect(ctx context.Context, url string, opts ...Option) (*nats.Conn, error)
• func Drain(ctx context.Context, nc *nats.Conn, name string, logger *slog.Logger, ...) error
• func IsShutdownFetchErr(ctx context.Context, err error) bool
• func TLSConfigFromFiles(certFile, keyFile, caFile string) (*tls.Config, error)
• type GroupOption
  • func WithGroupDrainTimeout(d time.Duration) GroupOption
  • func WithGroupLogger(l *slog.Logger) GroupOption
  • func WithJoinTimeout(d time.Duration) GroupOption
• type Option
  • func WithDrainTimeout(d time.Duration) Option
  • func WithLogger(l *slog.Logger) Option
  • func WithNATSOptions(opts ...nats.Option) Option
  • func WithName(name string) Option
  • func WithReconnectWait(d time.Duration) Option
  • func WithRetryOnFailedConnect() Option
  • func WithTLSConfig(t *tls.Config) Option
  • func WithoutTLS() Option
• type ShutdownGroup
  • func NewShutdownGroup(parent context.Context, opts ...GroupOption) *ShutdownGroup
  • func (g *ShutdownGroup) AddConn(name string, nc *nats.Conn)
  • func (g *ShutdownGroup) Context() context.Context
  • func (g *ShutdownGroup) Err() error
  • func (g *ShutdownGroup) Go(fn func(ctx context.Context))
  • func (g *ShutdownGroup) Shutdown() error

Constants

const (
	DefaultDrainTimeout  = 5 * time.Second
	DefaultReconnectWait = 2 * time.Second
)

Default connection tunables. DefaultDrainTimeout matches entire-core's tuned value (COR-923); nats.go's own default is 30s. It caps only the subscription-drain phase — see Drain for why it is not the whole budget.

const DefaultJoinTimeout = 10 * time.Second

DefaultJoinTimeout bounds how long ShutdownGroup.Shutdown waits for background loops to return before draining the connections anyway.

Variables

var ErrDrainTimeout = errors.New("nuts: NATS drain timed out before close")

ErrDrainTimeout is returned when a connection does not reach CLOSED within the total drain backstop. The connection is force-closed before return, so callers never inherit an ambiguous asynchronous drain.

Functions

func Connect

func Connect(ctx context.Context, url string, opts ...Option) (*nats.Conn, error)

Connect opens a NATS connection with the org-standard resiliency posture: rotation-aware mTLS (from the ENTIRE_INTERNAL_TLS_* env vars unless overridden), reconnect-forever (long-lived services must outlive arbitrary NATS outages rather than permanently CLOSE after the client default of 60 attempts), a bounded drain timeout, and lifecycle handlers that log at the right level — notably routing the nil-error DisconnectErr that nats.go fires on an explicit Close to INFO, so a clean shutdown does not look like a fault. Async errors (slow-consumer drops, permissions violations) and lame-duck notices are logged through the configured logger instead of nats.go's stderr default; override either handler via WithNATSOptions.

The initial dial is fail-fast: if the first connect does not succeed Connect returns an error, so a misconfigured service crash-loops visibly rather than booting into a broken state. Reconnect-forever applies only once a connection has been established. Pass WithRetryOnFailedConnect to opt into background retry of the initial dial instead.

The returned connection should be torn down with Drain (or via a ShutdownGroup) on graceful shutdown, not nc.Close, so in-flight work finishes and pull-consumer fetch loops exit cleanly.

func Drain

func Drain(ctx context.Context, nc *nats.Conn, name string, logger *slog.Logger, timeout time.Duration) error

Drain drains nc and waits (bounded by timeout) for it to flush pending acks/publishes and reach CLOSED. nats.go's Conn.Drain is asynchronous — it starts the drain on a background goroutine and returns immediately — so a bare Drain in a shutdown cleanup would let the process exit before the flush completes. This subscribes to the connection's CLOSED status event, starts the drain, and blocks until the connection closes or timeout elapses. name labels the connection in the logs so a wedged drain can be attributed, and diagnostics are written to logger (defaulting to slog.Default when nil). ctx supplies log context only: shutdown callers commonly pass an already cancelled process context, so timeout remains the authoritative drain budget and preserves the opportunity to flush during teardown.

timeout is the TOTAL time-to-CLOSED backstop, not the subscription-drain budget. The connect-time nats.DrainTimeout option caps only the subscription-drain phase; nats.go then runs an unconditional 5s publish flush before closing, so the true time-to-CLOSED is up to that flush longer than nats.DrainTimeout. timeout must exceed the connection's nats.DrainTimeout by at least the publish-flush budget or it will fire mid-drain and drop pending acks — ShutdownGroup sizes it automatically. timeout bounds how long we block; it is not a guarantee the drain finished. Once the consumer loops have joined (see ShutdownGroup) there is nothing left to drain, so on a reachable server this returns in milliseconds; the backstop only bites when the server is unreachable.

Unlike a raw nc.Drain, this does not touch the connection's ClosedHandler, so a handler installed by Connect or by the caller (via WithNATSOptions) is preserved, and concurrent Drain calls on the same connection do not clobber one another.

Drain is a no-op on a nil or already-closed connection. It returns nil only after the connection reaches CLOSED; immediate drain failures and timeouts are returned after force-closing the connection.

func IsShutdownFetchErr

func IsShutdownFetchErr(ctx context.Context, err error) bool

IsShutdownFetchErr reports whether a pull-consumer Fetch error is the benign result of the connection being drained/closed during shutdown rather than a real fault. It is only "clean" when ctx is already done — a mid-run connection loss (ctx still live) is a genuine error worth logging.

Use it to gate the error log in a fetch loop:

msgs, err := sub.Fetch(1, nats.MaxWait(5*time.Second))
if err != nil {
	if errors.Is(err, nats.ErrTimeout) || nuts.IsShutdownFetchErr(ctx, err) {
		// benign: idle poll, or a drain/close during shutdown
	}
	...
}

func TLSConfigFromFiles

func TLSConfigFromFiles(certFile, keyFile, caFile string) (*tls.Config, error)

TLSConfigFromFiles builds a fully rotation-aware mTLS *tls.Config from cert, key, and CA files on disk. Both sides of the trust are re-read from disk on every handshake: GetClientCertificate reloads the client keypair, and VerifyConnection rebuilds the root pool from the CA file and verifies the server against it. A cert-manager rotation of either the client identity or the trusted CA is therefore picked up on the next reconnect without a process restart. The files are read and parsed once here so a misconfiguration fails fast at construction time rather than on the first handshake.

Because the roots must be resolved per handshake, verification is done in VerifyConnection (which also re-checks the server hostname) rather than via the static RootCAs field.

Types

type GroupOption

type GroupOption func(*ShutdownGroup)

GroupOption configures a ShutdownGroup.

func WithGroupDrainTimeout

func WithGroupDrainTimeout(d time.Duration) GroupOption

WithGroupDrainTimeout overrides the fallback subscription-drain budget (DefaultDrainTimeout) used when draining a registered connection that does not carry its own nats.DrainTimeout. A connection created with WithDrainTimeout is drained on its own budget regardless; the total time-to-CLOSED backstop adds publish-flush headroom on top (see Drain).

func WithGroupLogger

func WithGroupLogger(l *slog.Logger) GroupOption

WithGroupLogger sets the logger for shutdown diagnostics. Defaults to slog.Default.

func WithJoinTimeout

func WithJoinTimeout(d time.Duration) GroupOption

WithJoinTimeout overrides DefaultJoinTimeout — the bound on waiting for background loops to return.

type Option

type Option func(*config)

Option configures a Connect call.

func WithDrainTimeout

func WithDrainTimeout(d time.Duration) Option

WithDrainTimeout overrides DefaultDrainTimeout.

func WithLogger

func WithLogger(l *slog.Logger) Option

WithLogger sets the logger used for connection lifecycle events. Defaults to slog.Default. Context-aware log methods are used, so a handler that reads trace context from the context still works.

func WithNATSOptions

func WithNATSOptions(opts ...nats.Option) Option

WithNATSOptions appends raw nats.Option values, applied after nuts's defaults so a caller can override any of them. An escape hatch for options nuts does not model.

func WithName

func WithName(name string) Option

WithName sets the NATS connection name (nats.Name), which labels the client in server monitoring and in the logs.

func WithReconnectWait

func WithReconnectWait(d time.Duration) Option

WithReconnectWait overrides DefaultReconnectWait.

func WithRetryOnFailedConnect

func WithRetryOnFailedConnect() Option

WithRetryOnFailedConnect keeps Connect from returning an error when the initial dial fails: nats.go instead returns a connection in RECONNECTING state and retries in the background (buffering publishes). It is off by default so a misconfigured connection (bad URL, wrong TLS material) fails fast at startup — crash-looping the pod so a bad rollout is visible — rather than silently accepting work it cannot deliver. Enable it only for optional/secondary connections a service is designed to run without.

func WithTLSConfig

func WithTLSConfig(t *tls.Config) Option

WithTLSConfig supplies an explicit *tls.Config and disables the default env-var mTLS lookup. Pass this (or WithoutTLS) when the standard ENTIRE_INTERNAL_TLS_* identity does not apply.

func WithoutTLS

func WithoutTLS() Option

WithoutTLS disables TLS entirely (plaintext). Intended for local development and tests against an embedded server; production dials always use mTLS.

type ShutdownGroup

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

ShutdownGroup coordinates graceful shutdown of a set of background pull-consumer loops and the NATS connections they use. It encodes the ordering that prevents rollout noise (COR-923): on Shutdown it cancels the loops' context, waits for them to return, and only THEN drains the connections — so a Drain never races an in-flight Fetch.

Typical use in a service main:

g := nuts.NewShutdownGroup(ctx)
nc, _ := nuts.Connect(g.Context(), url, nuts.WithName("worker"))
g.AddConn("worker", nc)
g.Go(func(ctx context.Context) { consumer.Run(ctx) }) // returns on ctx.Done
<-g.Context().Done() // SIGTERM/SIGINT, panic, or premature loop return
if err := g.Shutdown(); err != nil { return err }

func NewShutdownGroup

func NewShutdownGroup(parent context.Context, opts ...GroupOption) *ShutdownGroup

NewShutdownGroup returns a group whose context derives from parent and is cancelled by ShutdownGroup.Shutdown.

func (*ShutdownGroup) AddConn

func (g *ShutdownGroup) AddConn(name string, nc *nats.Conn)

AddConn registers a connection to be drained after the loops have joined. Draining last — never before the fetch loops stop — is the ordering that keeps a Drain/Close from racing an in-flight Fetch (COR-923).

Like ShutdownGroup.Go, registering after Shutdown has begun is a no-op: the drain set is snapshotted when shutdown starts, so a connection added afterward (including during the join window) would never be drained. It is refused and logged rather than silently dropped — the caller then owns tearing it down. Uses the same lock as shutdown's snapshot, so a connection is either in the snapshot or explicitly refused, never lost to a race.

func (*ShutdownGroup) Context

func (g *ShutdownGroup) Context() context.Context

Context returns the group's context, cancelled when Shutdown is called or a tracked loop fails. Pass it to every background loop and to Connect so the loops exit before their connections are drained.

func (*ShutdownGroup) Err

func (g *ShutdownGroup) Err() error

Err returns the first unexpected loop or shutdown failure, or nil while the group is healthy and after a clean shutdown. It is safe to call from a readiness check while the group is running.

func (*ShutdownGroup) Go

func (g *ShutdownGroup) Go(fn func(ctx context.Context))

Go runs fn as a tracked background loop. fn MUST return when its context is cancelled; Shutdown blocks (bounded by the join timeout) until it does.

A panic in fn is recovered so Shutdown can still join the other loops and drain connections, but it remains fatal to the group: the panic is recorded, logged with a stack, and cancels the group context. A return while the group context is still live is handled the same way. The owning process can observe the failure through ShutdownGroup.Err or ShutdownGroup.Shutdown.

Registering a loop after Shutdown has begun is a no-op: the loop would not be joined before the connections drain, so it is refused and logged rather than started. Go is safe to call concurrently with Shutdown.

func (*ShutdownGroup) Shutdown

func (g *ShutdownGroup) Shutdown() error

Shutdown cancels the group context, waits (bounded by the join timeout) for every Go loop to return, then drains every registered connection. It returns the first unexpected loop, join, or drain failure. It is safe to call more than once; later calls return the same result.