natsstore

package module
v0.4.0 Latest Latest
Warning

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

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

README

natsstore

natsstore implements storage's four storage primitives — Ledger, Leaser, KV, and Blobs — over NATS JetStream. It is the only module in the tree that depends on the NATS packages; consumers depend on the neutral storage contracts and wire natsstore in at their composition root.

A natsstore.Store runs over one of two backends, chosen at Open:

  • Embedded (Options.EmbeddedDir) — the Store owns an in-process JetStream server (a DontListen engine, no TCP socket) over a durable on-disk StoreDir at the caller-provided directory, so a single process gets a durable JetStream backend with no external broker to run. The directory is used directly as the StoreDir (created 0700); the caller owns the path, so no ~/.looprig default and no $XDG_DATA_HOME confinement are applied. No cross-process store lock is taken, so the caller must guarantee a single open per EmbeddedDir — two Stores or processes over the same dir would corrupt the JetStream file store.
  • Remote (Options.URL) — the Store dials an external NATS server (secure defaults; no InsecureSkipVerify) and owns only the connection.

Exactly one of URL / EmbeddedDir must be set (else an *OptionsError).

Usage

ctx, cancel := context.WithTimeout(context.Background(), 30*time.Second)
defer cancel()

st, err := natsstore.Open(ctx, natsstore.Options{EmbeddedDir: "/var/lib/myapp/store"})
if err != nil {
    return err
}
defer st.Close(ctx)

// Hand the four-primitive bundle to a facade such as sessionstore.Open.
sess, err := sessionstore.Open(ctx, st.Composite) // or st.Backend()

Store embeds *storage.Composite, so each primitive is reachable as a promoted field (st.Ledger, st.Leaser, st.KV, st.Blobs); the whole bundle is st.Composite (or st.Backend()). Close drains the connection and — in embedded mode — shuts the in-process server down afterwards; it is idempotent.

Buckets & durability

The ledger provisions its stream lazily on first append; the lease KV, application KV, and blob object-store buckets are provisioned idempotently at Open (CreateOrUpdate*), so reopening the same embedded StoreDir rebinds the existing buckets and sees the persisted data rather than failing.

Dependencies

The NATS dependencies are sanctioned only in this module (github.com/nats-io/nats.go for the JetStream client, github.com/nats-io/nats-server/v2 for the embedded in-process server). Everything else is stdlib plus the local storage contracts. See CLAUDE.md.

Documentation

Overview

Package natsstore implements storage's storage primitives over NATS JetStream and owns an embedded, in-process JetStream server (no TCP socket) over a persistent on-disk StoreDir, so a single process gets a durable JetStream backend with no external broker.

This file owns the embedded engine lifecycle: it starts the in-process server over a confined StoreDir, connects to it in-process, hands back a bound JetStreamContext the adapters write through, and shuts the server down cleanly. It is the only place that imports nats-server/v2/server — the embedded server is a composition concern.

Index

Constants

This section is empty.

Variables

This section is empty.

Functions

This section is empty.

Types

type BlobOpError 

type BlobOpError struct {
	Key   string
	Op    string
	Cause error
}

BlobOpError reports a definite failure of a blob operation blobStore performs (read / put / get / delete / list / getInfo) that is NOT one of the expected outcomes (absence, content conflict) — a source-reader fault or a backend fault. It fails closed and names the key (or, for list, the prefix), the operation, and unwraps to the underlying cause. (The storage taxonomy's only blob errors are BlobNotFoundError and BlobConflictError — neither fits a reader/backend fault — so this is a natsstore-specific typed error.)

func (*BlobOpError) Error 

func (e *BlobOpError) Error() string

func (*BlobOpError) Unwrap 

func (e *BlobOpError) Unwrap() error

type ConnectError 

type ConnectError struct {
	URL   string
	Cause error
}

ConnectError reports that Open could not dial a remote NATS URL. URL is REDACTED (any userinfo password is stripped — a NATS URL may embed credentials, which must never reach a log), and Cause unwraps to the underlying nats.go dial error.

func (*ConnectError) Error 

func (e *ConnectError) Error() string

func (*ConnectError) Unwrap 

func (e *ConnectError) Unwrap() error

type Engine 

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

Engine owns the embedded JetStream server, the in-process client connection, and the bound JetStreamContext. It is the composition-root handle a consumer builds once at startup and Closes once at shutdown. It is NOT safe for concurrent Close, but the consumer closes it exactly once on exit.

func OpenEngine 

func OpenEngine(opts EngineOptions) (*Engine, error)

OpenEngine starts an embedded JetStream server on a home/XDG-CONFINED StoreDir, connects to it in-process (no TCP), and returns a live Engine. It resolves the containment root ($XDG_DATA_HOME or home), verifies opts.DataDir stays within it (fail-secure against traversal), then hands off to openEngineAt for the fail-secure startup. It is the confined convenience entry point (the ~/.looprig / $XDG_DATA_HOME app-policy path); natsstore.Open's embedded mode instead drives openEngineAt directly on a caller-owned absolute dir, deliberately WITHOUT this home/XDG confinement (the caller explicitly owns the path).

func (*Engine) Close 

func (e *Engine) Close() error

Close drains the client connection and shuts the embedded server down cleanly, flushing JetStream state to the StoreDir. It is best-effort and safe to call once at exit; a drain error is returned but the server is always shut down.

func (*Engine) Conn 

func (e *Engine) Conn() *nats.Conn

Conn returns the in-process client connection. It is the handle a caller uses to build the context-aware jetstream.JetStream (jetstream.New) that the lease KV seam needs — the legacy JetStreamContext above cannot carry a per-call context. It is valid until Close.

func (*Engine) JetStream 

func (e *Engine) JetStream() nats.JetStreamContext

JetStream returns the bound JetStreamContext the adapters write through. It is valid until Close.

type EngineOptions 

type EngineOptions struct {
	DataDir      string
	SyncInterval time.Duration
	// MaxPayload is the connection-level maximum message size the DontListen server
	// accepts. A zero/negative value falls back to the package default (maxPayload,
	// 8 MiB). It must be >= the ledger stream's per-message ceiling (ledgerMaxMsgSize)
	// so a floor-sized append is not rejected at the connection; natsstore.Open's
	// embedded mode validates that floor before driving the engine, and openEngineAt
	// applies the default when this is unset.
	MaxPayload int32
}

EngineOptions configures the embedded engine. DataDir is the StoreDir (resolved + confined to the home root); SyncInterval is the explicit fsync cadence (the power-loss knob). A zero SyncInterval falls back to the conservative default.

func DefaultEngineOptions 

func DefaultEngineOptions() (EngineOptions, error)

DefaultEngineOptions returns convenience engine options: StoreDir at ~/.looprig/jetstream (overridable by $XDG_DATA_HOME → $XDG_DATA_HOME/looprig/jetstream) and the conservative explicit SyncInterval. It resolves the home/XDG root via os, failing closed (typed *StoreDirError) if neither is available.

type KVOpError 

type KVOpError struct {
	Key   string
	Op    string
	Cause error
}

KVOpError reports a definite failure of a KV operation kvStore performs (get / create / update / delete / keys) that is NOT one of the expected CAS outcomes — a backend fault or an ambiguous read. It fails closed and names the key (or, for keys, the prefix), the operation, and unwraps to the underlying cause. (The storage taxonomy's only KV errors are KeyNotFoundError and ConflictError — neither fits a backend fault — so this is a natsstore-specific typed error, analogous to StreamOpError / LeaseOpError.)

func (*KVOpError) Error 

func (e *KVOpError) Error() string

func (*KVOpError) Unwrap 

func (e *KVOpError) Unwrap() error

type LeaseEncodeError 

type LeaseEncodeError struct{ Cause error }

LeaseEncodeError wraps a failure to marshal a leaseRecord to JSON. A leaseRecord is a uint64 + string + time, so this is effectively unreachable, but the codec returns a typed error rather than dropping the json.Marshal error, to satisfy the errors-are-typed contract.

func (*LeaseEncodeError) Error 

func (e *LeaseEncodeError) Error() string

func (*LeaseEncodeError) Unwrap 

func (e *LeaseEncodeError) Unwrap() error

type LeaseOpError 

type LeaseOpError struct {
	Name  string
	Op    string
	Cause error
}

LeaseOpError reports a definite failure of a KV operation the leaser performs (get / create / update / decode) that is NOT one of the expected CAS outcomes — an ambiguous or malformed read, or a backend fault. It fails closed: an ambiguous read never silently grants ownership. It names the lease, the operation, and unwraps to the underlying cause. (The storage taxonomy has no read-error type — LeaseHeldError and LeaseLostError are the only lease errors it defines — so this is a natsstore-specific typed error, analogous to StreamOpError.)

func (*LeaseOpError) Error 

func (e *LeaseOpError) Error() string

func (*LeaseOpError) Unwrap 

func (e *LeaseOpError) Unwrap() error

type LockedEngine 

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

LockedEngine is one embedded engine bound to a single StoreDir and guarded by a process-exclusive lock. It is the per-directory unit a consumer opens and closes; the Engine itself stays directory-agnostic.

func OpenLockedEngine 

func OpenLockedEngine(dir string) (*LockedEngine, error)

OpenLockedEngine takes the exclusive lock on dir, then opens an embedded engine whose StoreDir lives beneath dir (<dir>/nats). A directory already open by a live engine returns *StoreLockedError before any server starts. dir must already exist (it holds the lock file); the engine's StoreDir subdirectory is created on demand.

func (*LockedEngine) Close 

func (e *LockedEngine) Close() error

Close shuts the embedded engine down and always releases the store lock afterwards, even when the engine drain fails. The drain error takes precedence in the return.

func (*LockedEngine) JetStream 

func (e *LockedEngine) JetStream() nats.JetStreamContext

JetStream returns the locked engine's bound JetStreamContext, valid until Close.

type NameEncodingError 

type NameEncodingError struct {
	Value  string
	Reason string
}

NameEncodingError reports an encoded token — a JetStream subject or stream name — that does not decode back to a valid storage name, i.e. one this package's encoders never emit. Value is the offending token.

func (*NameEncodingError) Error 

func (e *NameEncodingError) Error() string

type Options 

type Options struct {
	// URL is a remote NATS server URL (e.g. "nats://host:4222"). XOR EmbeddedDir. When
	// set, Open dials it with explicit, secure defaults (no InsecureSkipVerify, an
	// explicit connect timeout).
	URL string
	// EmbeddedDir is the StoreDir for an embedded, in-process JetStream server the Store
	// owns. XOR URL. It is used DIRECTLY as the engine StoreDir (no ~/.looprig default,
	// no home/XDG confinement — the caller explicitly owns the path); it is
	// filepath.Clean'd and must be absolute (a relative/empty dir is an *OptionsError).
	// It is created 0700 if absent.
	EmbeddedDir string
	// MaxPayload is the connection-level maximum message size for the EMBEDDED server.
	// In embedded mode a zero/negative value applies the 8 MiB default and a positive
	// value must be >= the ledger stream's per-message ceiling (4 MiB) or Open returns an
	// *OptionsError. In remote mode it is ignored entirely (neither applied nor
	// validated): the remote broker owns its own MaxPayload.
	MaxPayload int32
}

Options configures Open. Exactly one of URL or EmbeddedDir must be set (else an *OptionsError): URL selects a remote NATS backend, EmbeddedDir selects an embedded in-process JetStream server that the Store owns.

type OptionsError 

type OptionsError struct {
	Field  string
	Reason string
}

OptionsError reports an invalid or unusable Open option: neither or both of URL/EmbeddedDir set, a non-absolute EmbeddedDir, or a MaxPayload below the ledger stream's per-message ceiling. Field names the offending option and Reason explains the fault. It is a pure validation failure (no underlying cause), so it does not Unwrap.

func (*OptionsError) Error 

func (e *OptionsError) Error() string

type RecordReadError 

type RecordReadError struct {
	Name  string
	Seq   uint64
	Cause error
}

RecordReadError reports that a ledger cursor could not read the record at Seq from the backend. It fails closed — a caller must NOT treat it as end-of-ledger — and unwraps to the underlying backend error.

func (*RecordReadError) Error 

func (e *RecordReadError) Error() string

func (*RecordReadError) Unwrap 

func (e *RecordReadError) Unwrap() error

type ServerStartError 

type ServerStartError struct{ Cause error }

ServerStartError reports that the embedded JetStream server could not be created, became ready within the timeout, or could not be connected to in-process. It fails closed: without a live engine there is no durable backend.

func (*ServerStartError) Error 

func (e *ServerStartError) Error() string

func (*ServerStartError) Unwrap 

func (e *ServerStartError) Unwrap() error

type Store 

type Store struct {
	*storage.Composite
	// contains filtered or unexported fields
}

Store is an open natsstore over one NATS backend — an owned embedded engine, or a remote connection. It embeds *storage.Composite, so a caller reaches each primitive as a promoted field (store.Ledger, store.Leaser, store.KV, store.Blobs) and hands the whole bundle to a consumer as store.Composite or via Backend. The four primitives collide on method names, so no single type can implement all four (see the file-level comment); embedding the field-bundle is how Store sidesteps that.

func Open 

func Open(ctx context.Context, opts Options) (*Store, error)

Open assembles a JetStream-backed storage bundle over a single NATS backend chosen by opts: an embedded in-process server the Store owns (Options.EmbeddedDir) or a remote URL (Options.URL). It validates opts (*OptionsError on a bad combination), stands the backend up, provisions the ledger stream lazily plus the lease/kv/object buckets idempotently (so a reopen of the same embedded dir rebinds rather than fails), and wires the four primitives with storage.NewComposite.

ctx bounds the bucket-provisioning round-trips; pass a ctx with a deadline. On any wiring failure Open tears the backend down (drains the connection, shuts an embedded engine down) and returns the typed error — never a half-open Store.

Embedded mode takes NO cross-process store lock (it drives the engine directly rather than via LockedEngine), so the caller must guarantee a single open per EmbeddedDir — two Stores or processes over the same dir would corrupt the JetStream file store.

func (*Store) Backend 

func (s *Store) Backend() *storage.Composite

Backend returns the assembled four-primitive bundle to hand to a consumer such as sessionstore.Open. It is the embedded *storage.Composite; callers may read store.Composite directly instead.

func (*Store) Close 

func (s *Store) Close(ctx context.Context) error

Close tears the Store's owned backend down: it drains the connection (flushing an in-flight append) and, in embedded mode, shuts the in-process server down afterwards — the drain-before-shutdown ordering ported from the embedded engine's Close. It surfaces the first error and is idempotent: a second call is a no-op returning nil. After Close the Store must not be reused.

ctx is accepted for contract uniformity; the drain itself is bounded by the connection's drain timeout (remote: remoteDrainTimeout; embedded: the engine's default), as the underlying nats.go drain is not context-aware.

func (*Store) StoragePaths 

func (s *Store) StoragePaths() []string

StoragePaths returns the Store's frozen local persistence roots. It deliberately has a pointer receiver: Store contains live backend handles and a mutex and must not gain a value-copy-friendly interface method set.

type StoreDirError 

type StoreDirError struct {
	Path  string
	Cause error
}

StoreDirError reports that the embedded server's StoreDir could not be resolved or created: an empty/unresolvable home, an empty data dir, a path that escapes the home root (traversal), or a mkdir failure. It fails closed — the engine never starts on an unconfined or unwritable StoreDir. Cause chains the underlying os error when present.

func (*StoreDirError) Error 

func (e *StoreDirError) Error() string

func (*StoreDirError) Unwrap 

func (e *StoreDirError) Unwrap() error

type StoreLockError 

type StoreLockError struct {
	Path  string
	Cause error
}

StoreLockError reports that the store lock file could not be opened or that the flock syscall failed for a reason other than contention. It fails closed: without the lock the engine never opens.

func (*StoreLockError) Error 

func (e *StoreLockError) Error() string

func (*StoreLockError) Unwrap 

func (e *StoreLockError) Unwrap() error

type StoreLockedError 

type StoreLockedError struct {
	Path string
}

StoreLockedError reports that a StoreDir is already locked by a live engine (in this or another process). The directory is in use; the caller must not open a second engine over it.

func (*StoreLockedError) Error 

func (e *StoreLockedError) Error() string

type StreamOpError 

type StreamOpError struct {
	Stream string
	Op     string
	Cause  error
}

StreamOpError reports a definite failure of a JetStream stream-management operation the ledger seam performs (ensure / info / delete) — as opposed to a publish, whose outcome the ledger classifies itself. It names the stream and the operation and unwraps to the underlying NATS error.

func (*StreamOpError) Error 

func (e *StreamOpError) Error() string

func (*StreamOpError) Unwrap 

func (e *StreamOpError) Unwrap() error

type WiringError 

type WiringError struct {
	Component string
	Cause     error
}

WiringError reports that Open connected but could not assemble the four-primitive bundle: a JetStream context could not be bound, a KV/object bucket could not be provisioned, or the composite rejected a nil primitive. Component names the failed step; Cause unwraps to the underlying error. On a WiringError Open tears the connection/engine down and returns no Store.

func (*WiringError) Error 

func (e *WiringError) Error() string

func (*WiringError) Unwrap 

func (e *WiringError) Unwrap() error

Source Files

View all Source files

Jump to

Keyboard shortcuts

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