nkv

package module
v0.6.0 Latest Latest
Warning

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

 Go to latest Published: Aug 28, 2026 License: MIT Imports: 14 Imported by: 0

README

nkv: A wire-compatible NATS JetStream KV client

CI Go Reference Go Version License

nkv is a redesigned Go client for NATS JetStream Key-Value buckets. It keeps the standard KV wire format so buckets written by nkv and nats.go are fully interoperable, while replacing several costly client-side paths: point reads and enumeration use Direct Get instead of creating ephemeral consumers, watches use client-driven pull consumers for native backpressure, atomic transactions for grouped writes, and an API with per-operation options and typed values.

It is a deliberate alternative to the nats.go jetstream.KeyValue client, not a drop-in replacement. The stream layout (KV_<bucket>), subjects ($KV.<bucket>.<key>), and KV-Operation headers remain compatible. Rollup purges written by other clients are recognized as OpPurge; the public method signatures do not.

nkv does not expose a purge write operation. Buckets enforce a history of one, so a subject rollup provides no storage benefit over Delete.

Installation

go get github.com/blizzard/nkv.go@latest

Examples

See the examples guide for focused, runnable programs covering core operations, typed JSON values, optimistic concurrency, enumeration, watches, per-key TTL, atomic transactions, and nats.go interoperability. Each example is a standalone Go module with instructions for starting NATS Server and running the code.

Status

Pre-release. The API is not finalized and will change without notice.

nkv is at v0.x. Until a v1.0.0 release is tagged:

  • Any release may introduce breaking changes to exported types, functions, and behavior — including within a single minor version.
  • No deprecation period is guaranteed. Symbols may be renamed or removed outright.
  • Pin an exact version and read the release notes before upgrading.

Wire compatibility with the standard NATS KV layout is the one thing treated as stable: buckets written by nkv remain readable by nats.go, and vice versa. Data written today will not be orphaned by a future API change.

Wire compatibility describes the stored subjects, headers, and values; it does not relax bucket configuration validation. Open requires every setting listed below and therefore rejects a bucket created with the default nats.go KV configuration, which does not enable per-message TTL or atomic publish. Update the backing stream with the required settings before opening it with nkv.

Once v1.0.0 is released, the module follows Semantic Versioning and the exported API becomes subject to the usual Go compatibility guarantees.

Requirements

Minimum Why
Go 1.26 Declared in go.mod.
nats-server 2.14.0 Supported and tested server baseline.
nats.go v1.52.0 Client support for the above.
Required server features

CreateBucket enforces and Open requires a stream configuration that enables every feature below. Although the newest required feature was introduced in 2.12.0, nkv supports NATS Server 2.14.0 and newer.

Feature Stream setting / header Used by Since
Direct Get AllowDirect Get, List, Keys 2.9.0
Multi-subject consumer filters FilterSubjects Watch with WithAdditionalKeys 2.10.0
Batched Direct Get batch, next_by_subj, up_to_seq List, Keys paging 2.11.0
Per-message TTL AllowMsgTTL, Nats-TTL WithTTL 2.11.0
Limit markers SubjectDeleteMarkerTTL, Nats-Marker-Reason TTL/max-age tombstone detection 2.11.0
Atomic batch publish AllowAtomicPublish, Nats-Batch-* Tx, GenericTx 2.12.0

List and Keys page through Direct Get rather than creating ephemeral consumers, and Watch uses a pull ordered consumer — neither leaves server-side state behind.

CreateBucket defaults SubjectDeleteMarkerTTL to one minute so TTL and max-age expirations remain observable long enough for watchers to receive the server-generated marker. Set a longer value through Config.StreamConfig when watchers may be disconnected for longer periods. Delete also uses this value as its tombstone TTL when WithTTL is absent. Expiring delete tombstones carry Nats-Marker-Reason while retaining KV-Operation, so the server removes them without generating a second marker. WithTTL(0) explicitly keeps a tombstone indefinitely and omits the marker reason. Transactions follow the same rule, with WithTxTTL taking precedence. Values below one second are rejected by both nkv and the server.

External stream creation requirements

[!WARNING] nkv expects the KV-specific settings shown in the command below. Omitting or changing those settings causes Open to reject the stream. Deployment-specific settings may be changed as needed, including --replicas, --storage, resource limits such as --max-bytes, and placement, mirror, or source configuration.

Instead of calling CreateBucket, you can create the underlying KV stream with the NATS CLI. This command was verified against nats CLI v0.4.0 (nats stream add --help):

BUCKET_NAME=MY_BUCKET

stream_args=(
  "KV_${BUCKET_NAME}"
  --defaults
  --subjects "\$KV.${BUCKET_NAME}.>"  # required
  --retention limits                  # required
  --discard new                       # required
  --max-msgs-per-subject 1            # required
  --allow-rollup                      # required
  --deny-delete                       # required
  --allow-direct                      # required
  --allow-msg-ttl                     # required
  --subject-del-markers-ttl 1m        # required; can increase
  --allow-batch                       # required
  --storage file                      # can change
  --replicas 1                        # can change
)
nats stream add "${stream_args[@]}"

BUCKET_NAME supplies both the bucket name and the required naming relationship: a bucket named MY_BUCKET uses stream KV_MY_BUCKET and subject $KV.MY_BUCKET.>. --replicas 1 is suitable for a single-node JetStream cluster but provides no replica redundancy.

Upgrading an existing KV bucket

CreateBucket automatically updates an existing bucket to the required configuration. To perform the same upgrade with the NATS CLI while retaining the bucket's messages, edit its backing stream:

BUCKET_NAME=MY_BUCKET

stream_args=(
  "KV_${BUCKET_NAME}"
  --force
  --subjects "\$KV.${BUCKET_NAME}.>"   # required
  --retention limits                   # required
  --discard new                        # required
  --max-msgs-per-subject 1             # required
  --allow-rollup                       # required
  --deny-delete                        # required
  --allow-direct                       # required
  --allow-msg-ttl                      # required
  --allow-batch                        # required
  --subject-del-markers-ttl 1m         # required; can increase
)
nats stream edit "${stream_args[@]}"

The command enforces the nkv-specific settings without changing deployment settings such as storage, replicas, resource limits, placement, mirrors, or sources. The delete marker TTL controls how long TTL and max-age expiration markers remain available to watchers; increase it when watchers may be disconnected for longer than one minute.

Contributing

See CONTRIBUTING.md for the module layout, how to run the test suite, and the dependency rules.

Security

To report a vulnerability, see SECURITY.md. Please do not open a public issue.

License

See LICENSE.

Documentation

Overview

Package nkv is a replacement for the nats.go jetstream.KeyValue client. It is wire compatible with the standard KV bucket layout (stream KV_<bucket>, subjects $KV.<bucket>.<key>, KV-Operation headers) but NOT API compatible with nats.go.

Design goals:

  • every operation takes variadic options so the surface can grow
  • per-key TTL on Put/Create/Delete (Nats-TTL)
  • List/Keys via JetStream Direct Get (no ephemeral consumers)
  • Watch via pull ordered consumers (no push, native backpressure)
  • typed values via a generic codec wrapper instead of a []byte-only API

Index

Constants

This section is empty.

Variables

View Source 
var (
	ErrKeyNotFound      = errors.New("kv: key not found")
	ErrKeyExists        = errors.New("kv: key exists")
	ErrRevisionMismatch = errors.New("kv: revision mismatch")
	ErrInvalidKey       = errors.New("kv: invalid key")
	ErrInvalidOption    = errors.New("kv: invalid option")
)

Errors returned by bucket operations.

View Source 
var (
	ErrDuplicateKey = errors.New("kv: duplicate key in tx")
	ErrEmptyTx      = errors.New("kv: empty tx")
	ErrTxClosed     = errors.New("kv: tx already committed or aborted")
	ErrTxConflict   = errors.New("kv: tx conflict")
)

Tx errors.

View Source 
var ErrWatcherStopped = errors.New("kv: watcher stopped")

ErrWatcherStopped is returned by Next after the watcher has been stopped.

Functions

func WithAdditionalKeys 

func WithAdditionalKeys(patterns ...string) extraFiltersOption

WithAdditionalKeys adds more key patterns to the watch — one consumer can cover several prefixes (FilterSubjects is plural on pull consumers).

func WithHeaders 

func WithHeaders(h nats.Header) headersOption

WithHeaders attaches additional NATS message headers to the write operation. Headers set here will not override KV-internal headers (KV-Operation, Nats-Rollup, etc.); those take precedence.

func WithInactiveThreshold 

func WithInactiveThreshold(d time.Duration) inactiveThresholdOption

WithInactiveThreshold tunes how long the server keeps the watch consumer alive across disconnects before a recreate + replay (default 30s). It must be positive.

func WithListBatch 

func WithListBatch(n int) listBatchOption

WithListBatch sets the direct-get page size (default 256). It must be positive.

func WithPullBatch 

func WithPullBatch(n int) pullBatchOption

WithPullBatch sets the pull batch size for the underlying ordered consumer (default 512). It must be positive.

func WithRevision 

func WithRevision(rev uint64) revisionOption

WithRevision targets a specific revision. On Get it fetches that exact revision; on Delete it makes the operation conditional on the key's latest revision matching; on Watch it resumes inclusively from that stream revision.

func WithTTL 

func WithTTL(d time.Duration) ttlOption

WithTTL sets a per-key TTL (Nats-TTL) on the written revision. Delete defaults to the bucket's SubjectDeleteMarkerTTL when this option is absent. Zero explicitly disables TTL; positive values must be at least one second. Requires AllowMsgTTL on the bucket (set by CreateBucket and required by Open).

Types

type Bucket 

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

Bucket is a handle to a KV bucket. It is wire compatible with buckets created by nats.go's jetstream.KeyValue implementation.

func CreateBucket 

func CreateBucket(ctx context.Context, nc *nats.Conn, cfg Config) (*Bucket, error)

CreateBucket creates a KV bucket or updates its existing backing stream and returns a handle to it. Updating applies the required nkv settings, allowing an existing KV stream to be upgraded to nkv compatibility without manually modifying its stream configuration.

KV-specific invariants are enforced on the StreamConfig: conflicting values are rejected with an error, unset fields get correct defaults.

func Open 

func Open(ctx context.Context, nc *nats.Conn, bucket string) (*Bucket, error)

Open returns a handle to an existing bucket. It verifies the backing stream exists and has every configuration setting required by the nkv API.

func (*Bucket) Create 

func (b *Bucket) Create(ctx context.Context, key string, value []byte, opts ...CreateOption) (uint64, error)

Create stores a value only if the key does not currently exist (or its latest revision is a tombstone). Returns ErrKeyExists otherwise.

func (*Bucket) Delete 

func (b *Bucket) Delete(ctx context.Context, key string, opts ...DeleteOption) error

Delete writes a delete tombstone for key; history is retained up to the bucket's History limit. WithRevision makes it a CAS delete. The tombstone defaults to the bucket's SubjectDeleteMarkerTTL; WithTTL overrides it.

func (*Bucket) Get 

func (b *Bucket) Get(ctx context.Context, key string, opts ...GetOption) (*Entry, error)

Get returns the latest revision of a key, or a specific revision with WithRevision. Tombstoned keys return ErrKeyNotFound. Consumer-free: a single DIRECT.GET request served by any stream replica.

func (*Bucket) IsClusterLocal 

func (b *Bucket) IsClusterLocal(ctx context.Context) (bool, error)

IsClusterLocal reports whether the bucket's stream is hosted by the same cluster the connection is attached to. If either side reports no cluster information (e.g. a single non-clustered server), the bucket is considered local.

func (*Bucket) Keys 

func (b *Bucket) Keys(ctx context.Context, pattern string, opts ...KeysOption) iter.Seq2[string, error]

Keys returns an iterator over all keys matching the pattern. Tombstones are skipped. A thin wrapper around List that discards values.

func (*Bucket) List 

func (b *Bucket) List(ctx context.Context, pattern string, opts ...ListOption) iter.Seq2[*Entry, error]

List returns an iterator over all entries matching the key pattern (e.g. "users.*" or ">" for all). Tombstones are skipped unless WithDeletes() is specified. Serves from any replica via Direct Get — zero consumers.

Uses paged next_by_subj direct get (batched stream walk). Pins to the stream's last_seq at call time for a consistent snapshot. No subject count limits, no consumer lifecycle.

func (*Bucket) Name 

func (b *Bucket) Name() string

Name returns the bucket name.

func (*Bucket) Put 

func (b *Bucket) Put(ctx context.Context, key string, value []byte, opts ...PutOption) (uint64, error)

Put stores a value under key and returns the new revision.

func (*Bucket) Status 

func (b *Bucket) Status(ctx context.Context) (*jetstream.StreamInfo, error)

Status fetches live (non-cached) stream info for the backing stream.

func (*Bucket) Stream 

func (b *Bucket) Stream() string

Stream returns the backing stream name (KV_<bucket>).

func (*Bucket) Tx 

func (b *Bucket) Tx(opts ...TxOption) *Tx

Tx creates a new atomic transaction. All staged operations are buffered on the server and committed atomically. If any operation fails (e.g. CAS conflict), the entire tx is discarded.

A Tx is not safe for concurrent use. Use defer tx.Abort() to ensure cleanup if Commit is not reached.

func (*Bucket) Update 

func (b *Bucket) Update(ctx context.Context, key string, value []byte, rev uint64, opts ...UpdateOption) (uint64, error)

Update stores a value only if the key's latest revision matches rev (compare-and-swap). Returns ErrRevisionMismatch on conflict.

func (*Bucket) Watch 

func (b *Bucket) Watch(ctx context.Context, pattern string, opts ...WatchOption) (*Watcher, error)

Watch starts a pull-ordered-consumer-backed watcher on the given key pattern (e.g. "users.>" or "config.*"). After the initial replay of current state completes, Next returns live updates. WithRevision resumes inclusively from the given stream revision instead of replaying current state.

The caller must call Stop when done. Unlike the nats.go KV Watch, the underlying consumer is pull-based: stalls do not slow-consumer the connection.

func (*Bucket) WatchAll 

func (b *Bucket) WatchAll(ctx context.Context, opts ...WatchOption) (*Watcher, error)

WatchAll is a convenience for watching all keys in the bucket.

type Codec 

type Codec struct {
	// ContentType is the MIME type written to the Content-Type header
	// (e.g. "application/json", "application/protobuf").
	ContentType string
	Marshal     func(any) ([]byte, error)
	Unmarshal   func([]byte, any) error
}

Codec defines how values are marshaled to/from the wire []byte representation stored in the KV bucket. Each codec has a MIME content type that is stored in the Content-Type message header.

func JSONCodec 

func JSONCodec() Codec

JSONCodec returns a Codec that uses encoding/json with content type "application/json".

type Config 

type Config struct {
	// StreamConfig carries all non-KV-specific stream settings. Fields
	// that conflict with KV semantics will be overwritten or rejected.
	jetstream.StreamConfig

	// Bucket is the KV bucket name. Required. Must match [a-zA-Z0-9_-]+.
	Bucket string
}

Config describes a bucket on creation. Accepts a jetstream.StreamConfig for full control over non-KV-specific stream settings (replicas, storage, max bytes, placement, mirror, sources, etc.).

KV-invariant fields are enforced/overwritten:

  • Name (set to KV_<Bucket>)
  • Subjects (set to $KV.<Bucket>.>)
  • AllowRollup (true)
  • DenyDelete (true)
  • AllowDirect (true — required for consumer-free list)
  • AllowMsgTTL (true — required for per-key TTL)
  • SubjectDeleteMarkerTTL (must be at least 1s — required for TTL expiry notifications)
  • AllowAtomicPublish (true — required for batch writes)
  • Discard (DiscardNew)

Fields given reasonable defaults if zero:

  • MaxMsgsPerSubject → 1 (history=1)
  • Replicas → 1
  • Storage → FileStorage
  • Duplicates → 2m
  • SubjectDeleteMarkerTTL → 1m

Callers who want the simplest config can pass just a Bucket name and leave StreamConfig at its zero value.

type CreateOption 

type CreateOption interface {
	// contains filtered or unexported methods
}

CreateOption configures Create.

type DeleteOption 

type DeleteOption interface {
	// contains filtered or unexported methods
}

DeleteOption configures Delete.

type Entry 

type Entry struct {
	Bucket      string
	Key         string
	Value       []byte
	Revision    uint64 // stream sequence of this revision
	Delta       uint64 // messages after this entry in the delivery sequence
	Created     time.Time
	Operation   Operation
	ContentType string // MIME type from Content-Type header, empty if absent
	Headers     nats.Header
}

Entry is a single revision of a key.

func (*Entry) IsTombstone 

func (e *Entry) IsTombstone() bool

IsTombstone reports whether the entry is a delete/purge marker.

type Generic 

type Generic[T any] struct {
	// contains filtered or unexported fields
}

Generic wraps a Bucket with a codec for type-safe data operations.

The write codec is used for all encode operations and its ContentType is stamped into the Content-Type header. On reads, the Content-Type header selects the decode codec; if absent, the fallback codec is used (defaults to the write codec).

func NewGeneric 

func NewGeneric[T any](kv *Bucket, opts ...GenericOption) *Generic[T]

NewGeneric returns a generic typed wrapper around a Bucket.

If no WithWriteCodec option is provided, JSONCodec() is used as the default. The write codec is registered for read dispatch and used as the fallback for headerless entries unless overridden with WithFallbackCodec. It panics if any configured codec is incomplete or if two codecs use the same content type.

func (*Generic[T]) Bucket 

func (t *Generic[T]) Bucket() *Bucket

Bucket returns the underlying Bucket.

func (*Generic[T]) Create 

func (t *Generic[T]) Create(ctx context.Context, key string, value T, opts ...CreateOption) (uint64, error)

Create encodes and stores a value only if the key does not exist.

func (*Generic[T]) Delete added in v0.3.0 

func (t *Generic[T]) Delete(ctx context.Context, key string, opts ...DeleteOption) error

Delete writes a delete tombstone for key.

func (*Generic[T]) Get 

func (t *Generic[T]) Get(ctx context.Context, key string, opts ...GetOption) (T, uint64, error)

Get retrieves and decodes the latest value for key. Returns the decoded value and its revision (needed for CAS Update calls).

func (*Generic[T]) Keys added in v0.4.0 

func (t *Generic[T]) Keys(ctx context.Context, pattern string, opts ...KeysOption) iter.Seq2[string, error]

Keys returns an iterator over keys matching the pattern.

func (*Generic[T]) List 

func (t *Generic[T]) List(ctx context.Context, pattern string, opts ...ListOption) iter.Seq2[T, error]

List returns a typed iterator over values matching the pattern.

func (*Generic[T]) Put 

func (t *Generic[T]) Put(ctx context.Context, key string, value T, opts ...PutOption) (uint64, error)

Put encodes and stores a value, returning the new revision.

func (*Generic[T]) Tx 

func (t *Generic[T]) Tx(opts ...TxOption) *GenericTx[T]

Tx creates a GenericTx that inherits this wrapper's codec and prefix.

func (*Generic[T]) Update 

func (t *Generic[T]) Update(ctx context.Context, key string, value T, rev uint64, opts ...UpdateOption) (uint64, error)

Update encodes and stores a value only if the key's revision matches rev.

func (*Generic[T]) Watch added in v0.3.0 

func (t *Generic[T]) Watch(ctx context.Context, pattern string, opts ...WatchOption) (*GenericWatcher[T], error)

Watch starts a typed watcher for entries matching pattern.

type GenericEntry added in v0.3.0 

type GenericEntry[T any] struct {
	Entry

	Value T
}

GenericEntry is a watched entry with its value decoded as T. Entry retains the operation metadata and raw encoded value. Tombstones and metadata-only entries have the zero value of T.

type GenericOption 

type GenericOption func(*genericConfig)

GenericOption configures a Generic bucket wrapper. Options are non-generic so callers never need type constraints on configuration.

func WithDefaultTTL 

func WithDefaultTTL(ttl time.Duration) GenericOption

WithDefaultTTL sets a default TTL applied to all writes (Put/Create/Update) unless the caller explicitly passes WithTTL on the individual operation.

func WithFallbackCodec 

func WithFallbackCodec(c Codec) GenericOption

WithFallbackCodec sets the codec used to decode entries that have no Content-Type header (legacy data) or an unknown content type. If not specified, the write codec is used as fallback.

func WithPrefix 

func WithPrefix(prefix string) GenericOption

WithPrefix sets a key prefix that is automatically prepended (with a dot separator) to all keys in Get/Put/Create/Update/List/Tx operations. Empty prefixes and prefixes containing only dots disable prefixing. Other trailing dots are removed. It panics if the normalized prefix is not a valid key.

func WithReadCodec 

func WithReadCodec(c Codec) GenericOption

WithReadCodec registers an additional codec for read dispatch. When an entry's Content-Type matches the codec's ContentType, it is used for decoding. May be specified multiple times for different content types.

func WithWriteCodec 

func WithWriteCodec(c Codec) GenericOption

WithWriteCodec sets the write codec used for encoding on Put/Create/Update. Its ContentType is stamped into the Content-Type header and registered for read dispatch. If not specified, JSONCodec() is used.

type GenericTx 

type GenericTx[T any] struct {
	// contains filtered or unexported fields
}

GenericTx is a typed wrapper around a Tx. It marshals values using the provided codec and stamps Content-Type on each staged operation so the read side can dispatch correctly.

Multiple GenericTx instances can share the same underlying Tx to stage operations for different types in a single atomic commit.

func NewGenericTx 

func NewGenericTx[T any](tx *Tx, opts ...GenericOption) *GenericTx[T]

NewGenericTx wraps an existing Tx with a codec. The codec's ContentType is stamped on every message staged through this wrapper. It panics when the write codec is incomplete.

func (*GenericTx[T]) Abort 

func (g *GenericTx[T]) Abort()

Abort marks the tx as closed without committing. Idempotent and safe to defer.

func (*GenericTx[T]) Commit 

func (g *GenericTx[T]) Commit(ctx context.Context) (*TxResult, error)

Commit atomically writes all staged operations. On success all messages appear in the stream; on failure none do.

func (*GenericTx[T]) Create 

func (g *GenericTx[T]) Create(key string, val T) error

Create stages a create (fails on commit if key exists).

func (*GenericTx[T]) Delete 

func (g *GenericTx[T]) Delete(key string, opts ...DeleteOption) error

Delete stages a delete tombstone. WithRevision makes it CAS.

func (*GenericTx[T]) Len 

func (g *GenericTx[T]) Len() int

Len returns the number of staged operations.

func (*GenericTx[T]) Put 

func (g *GenericTx[T]) Put(key string, val T) error

Put stages an unconditional put with the marshaled value.

func (*GenericTx[T]) Update 

func (g *GenericTx[T]) Update(key string, val T, revision uint64) error

Update stages a CAS update (fails on commit if revision doesn't match).

type GenericWatcher added in v0.3.0 

type GenericWatcher[T any] struct {
	// contains filtered or unexported fields
}

GenericWatcher decodes entries delivered by a Watcher as T.

func (*GenericWatcher[T]) InitialDone added in v0.3.0 

func (w *GenericWatcher[T]) InitialDone() bool

InitialDone reports whether the initial replay has completed.

func (*GenericWatcher[T]) Next added in v0.3.0 

func (w *GenericWatcher[T]) Next() (*GenericEntry[T], error)

Next blocks until the next typed entry is available. It returns nil, nil at the end of the initial replay, matching Watcher.Next.

func (*GenericWatcher[T]) Stop added in v0.3.0 

func (w *GenericWatcher[T]) Stop()

Stop tears down the watcher and its underlying consumer.

func (*GenericWatcher[T]) Updates added in v0.3.0 

func (w *GenericWatcher[T]) Updates() <-chan *GenericEntry[T]

Updates returns a channel adapter around the typed watcher. Repeated calls return the same channel. A nil entry signals the end of the initial replay.

type GetOption 

type GetOption interface {
	// contains filtered or unexported methods
}

GetOption configures Get.

type KeysOption 

type KeysOption interface {
	// contains filtered or unexported methods
}

KeysOption configures Keys.

type ListOption 

type ListOption interface {
	// contains filtered or unexported methods
}

ListOption configures List.

func WithDeletes 

func WithDeletes() ListOption

WithDeletes makes List yield tombstone entries instead of skipping them.

type Operation 

type Operation int

Operation describes what a stored revision represents. 

const (
	OpPut Operation = iota
	OpDelete
	OpPurge
)

func (Operation) String 

func (o Operation) String() string

type PutOption 

type PutOption interface {
	// contains filtered or unexported methods
}

PutOption configures Put.

type Tx 

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

func (*Tx) Abort 

func (tx *Tx) Abort()

Abort marks the tx as closed without committing. Buffered messages on the server are discarded automatically. Abort is idempotent and safe to defer.

func (*Tx) Commit 

func (tx *Tx) Commit(ctx context.Context) (*TxResult, error)

Commit atomically writes all staged operations to the bucket. On success, all messages appear in the stream; on failure, none do.

After Commit (success or failure) the tx cannot be reused.

func (*Tx) Create 

func (tx *Tx) Create(key string, value []byte) error

Create stages a create (fails on commit if key exists).

func (*Tx) Delete 

func (tx *Tx) Delete(key string, opts ...DeleteOption) error

Delete stages a delete tombstone. WithRevision makes it CAS.

func (*Tx) Len 

func (tx *Tx) Len() int

Len returns the number of staged operations.

func (*Tx) Put 

func (tx *Tx) Put(key string, value []byte) error

Put stages an unconditional put.

func (*Tx) Update 

func (tx *Tx) Update(key string, value []byte, revision uint64) error

Update stages a CAS update (fails on commit if revision doesn't match).

type TxOption 

type TxOption func(*Tx)

TxOption configures a Tx.

func WithTxTTL 

func WithTxTTL(ttl time.Duration) TxOption

WithTxTTL sets the TTL for all messages in the tx. Values below 1s are clamped to 1s (NATS server minimum). Zero or negative explicitly disables TTL, including the SubjectDeleteMarkerTTL fallback for tombstones.

type TxResult 

type TxResult struct {
	ID       string
	Size     int
	Sequence uint64
}

TxResult contains metadata from a committed tx.

type TxStageError 

type TxStageError struct {
	OpIndex int
	Key     string
	Detail  string
}

TxStageError is returned when an intermediate tx message is rejected by the server (e.g. CAS conflict). The server discards the entire pending tx on any intermediate failure.

func (*TxStageError) Error 

func (e *TxStageError) Error() string

type UpdateOption 

type UpdateOption interface {
	// contains filtered or unexported methods
}

UpdateOption configures Update.

type WatchOption 

type WatchOption interface {
	// contains filtered or unexported methods
}

WatchOption configures Watch.

func WithMetaOnly 

func WithMetaOnly() WatchOption

WithMetaOnly delivers entries without values (consumer headers_only).

func WithUpdatesOnly 

func WithUpdatesOnly() WatchOption

WithUpdatesOnly skips the initial replay and only delivers new revisions.

type Watcher 

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

Watcher delivers a stream of entries from a pull-based ordered consumer. Backpressure is native: the client pulls what it can process. No push delivery, no connection-level slow consumer risk.

func (*Watcher) InitialDone 

func (w *Watcher) InitialDone() bool

InitialDone reports whether the initial replay has completed. After this returns true, all entries from Next are live updates.

func (*Watcher) Next 

func (w *Watcher) Next() (*Entry, error)

Next blocks until the next entry is available. Concurrent calls are serialized; each entry is delivered to exactly one caller. Returns nil, nil once the initial replay is complete (sentinel) — subsequent calls deliver live updates. Returns an error on context cancellation or unrecoverable consumer failure.

func (*Watcher) Stop 

func (w *Watcher) Stop()

Stop tears down the watcher and its underlying consumer.

func (*Watcher) Updates 

func (w *Watcher) Updates() <-chan *Entry

Updates returns a channel adapter around the pull-based watcher for callers that prefer channel semantics. Repeated calls return the same channel. The channel closes when the watcher is stopped or the context is canceled. A nil *Entry signals end of initial replay.

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.