store

package module
v1.4.1 Latest Latest
Warning

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

 Go to latest Published: Jul 6, 2026 License: Apache-2.0 Imports: 17 Imported by: 7

README

store

Transactional, capability-aware Data Store interface over several embedded key-value engines. One uniform API (Get/Set/Increment/CompareAndSet/Touch/Remove/Enumerate/Watch, context-propagated transactions, protobuf codec) lets you select and migrate between engines without rewriting application code.

This is a multi-module repository (one go.mod per provider/middleware), so an application only pulls the engine it actually imports. See go.work for the local development workspace.

Layout

store/                      module go.arpabet.com/store              (the interface)
  storetest/                module .../storetest                     (conformance + benchmark suite)
  benchmarks/               module .../benchmarks                    (cross-engine benchmark runner)
  providers/badger/         module .../providers/badger              (BadgerDB v4)
  providers/pebble/         module .../providers/pebble              (PebbleDB v2)
  providers/bbolt/          module .../providers/bbolt               (etcd bbolt)
  providers/bolt/           module .../providers/bolt                (boltdb, legacy)
  providers/mem/            module .../providers/mem                 (in-memory cache)
  providers/rosedb/         module .../providers/rosedb              (Bitcask, native TTL)
  providers/nutsdb/         module .../providers/nutsdb              (nutsdb, native txns + TTL)
  middleware/crypto/        module .../middleware/crypto             (AES-GCM at rest)
  middleware/otel/          module .../middleware/otel               (OpenTelemetry tracing)

Capabilities

Every store reports what it actually guarantees via Features() Capability, and the shared conformance suite (storetest) asserts a behavior for a provider only if it advertises the matching capability. Providers are interchangeable for a given feature when both report it.

Provider Backend Ordered TTL Atomic/CAS Watch Transactional BatchAtomic Encrypted Usage
providers/badger BadgerDB v4 Y Y Y Y Y N Y (native) User / PII data
providers/pebble PebbleDB v2 Y Y Y Y N Y via crypto App data
providers/bbolt etcd bbolt Y Y Y Y N Y via crypto Config
providers/bolt boltdb (legacy) Y Y Y Y N Y via crypto Config (legacy)
providers/mem ttlcache v3 Y Y Y Y N N via crypto Hot data
providers/rosedb rosedb (Bitcask) Y Y Y Y N Y via crypto App data
providers/nutsdb nutsdb Y Y Y Y Y Y via crypto App data

Batch writes (s.Batch(ctx).Put(...).Do() / SetBatchRaw) are available on every store; BatchAtomic marks the ones where a batch is all-or-nothing (single transaction). Badger uses WriteBatch (chunked, not atomic) and mem applies under a lock (not isolated from readers), so neither advertises BatchAtomic.

Range scans and pagination need ordering: Enumerate(ctx).Range(from, to) over the half-open [from, to) interval, and .After(token).Limit(n).DoPage(...) which returns an opaque continuation token (the last key) for the next page. These return ErrNotSupported on a store without Ordered; mem sorts on enumerate so it qualifies.

TTL, versioning and compare-and-set on the non-Badger engines are provided by a shared value envelope (version | expiresAt | value) with lazy expiry; watch on those engines is served by an in-process fan-out hub. Badger uses its native TTL, MVCC versions, transactions and Subscribe. rosedb uses native TTL (disk expiry — so it needs no sweeper) and native ordered iteration and atomic batches, with the envelope carrying the version and watch served by the hub.

nutsdb adds native transactions (exposed via BeginTransaction/EndTransaction, like Badger; a single writer at a time gives serialized read-modify-write) and native timer-wheel TTL (disk expiry — so it needs no sweeper). Its bucket model maps to the bolt-style bucket:key separator: a full key "bucket:key" is stored under nutsdb bucket bucket, with each bucket a BTree for ordered scans. The envelope carries the version; watch is served by the hub, and events for writes inside an explicit transaction are published on commit.

Watch semantics. Watch is in-process (mutations through this process's handle only; a different process writing the same file is not observed) and best-effort (bounded per-watcher buffers drop events under load — treat an event as "re-read this key"). TTL expiry surfaces as a WatchDelete only when the entry is reclaimed: automatically on mem, and via a running sweeper (store.StartSweeper) on the disk backends. Writes made directly on the engine returned by Instance() bypass the store layer and are not delivered to watchers.

Multi-store — tenants × regions (MultiDataStore)

store.MultiDataStore serves per-(majorKey, region) DataStore views over one shared backend, GemFire/GemStone style: the region is a logical table, the majorKey is the collocation unit — a tenant or a per-user profile id — so all of one owner's data lives under one scannable prefix (exportable / erasable as a unit; the GDPR unit).

type MultiDataStore interface {
    Region(majorKey, region string) DataStore // cheap borrowed view
    Features() Capability                     // views inherit these
    // + glue lifecycle; Destroy closes the backend (views' Destroy is a no-op)
}

Two implementations:

  • Native — providers/cdb: cdb.NewMulti(name, address, opts...) sends MajorKey/RegionName on every request over one cluster connection; a view is a struct, not a dial, so per-request views (e.g. per profile id) are free.
  • Prefix wrapper — any flat provider becomes a MultiDataStore:
base, _ := badgerstore.New("app", dataDir)
multi := store.NewMulti("app", base, store.LenPrefix())
users := multi.Region("acme", "USERS") // full DataStore, keys are plain minor keys

The layout is a pluggable KeyStrategy with two built-ins: store.Separator(":") = major:region:minor (readable, any separator; names must not contain it) and store.LenPrefix() = [2B len]major[2B len]region + minor — the consensusdb backing-store convention, binary-safe with the minor key as the tail, so lexical ordering and prefix scans of minor keys are preserved inside a view. Custom layouts implement the two-method KeyStrategy interface. strategy.TenantPrefix(major) is the whole-tenant prefix for export / migration / GDPR erasure scans on the wrapped store.

The factory pattern stops hard-coding a region: open one MultiDataStore bean and carve Region views where they're needed —

// store.provider=cdb → native; anything else → prefix wrapper over the engine
multi, _ := cdbstore.NewMulti("app", address, cdbstore.WithTLSConfig(cfg))
// vs: multi := store.NewMulti("app", badgerBase, store.LenPrefix())
users, jobs := multi.Region(tenant, "USERS"), multi.Region(tenant, "JOBS")
profile := multi.Region(profileId, "PROFILE") // per-user collocation

Typed access (generics)

A codec-driven typed layer sits on top of the raw byte API (pure sugar, no provider requirements). Built-in Codec[T]: store.JSON[T](), store.MsgPack[T](), and store.Proto[*pb.T]().

users := store.Of[*pb.User](ds, store.Proto[*pb.User]())
users.Put(ctx, []byte("u:1"), &pb.User{Name: "Ann"}, store.NoTTL)
u, found, err := users.Get(ctx, []byte("u:1"))

// or as free functions
err = store.Put(ctx, ds, []byte("c:1"), cfg, store.JSON[Config](), store.NoTTL)
cfg, found, err := store.Get(ctx, ds, []byte("c:1"), store.JSON[Config]())

Middleware

Middleware are decorators that wrap any store.ManagedDataStore and are themselves stores, so they compose and remain interchangeable:

  • crypto — encrypts values at rest with AES-GCM (keys stay plaintext for ordering/prefix scans). Adds the Encrypted capability to any backend, so "encrypted config in bbolt" or "encrypted hot data in mem" is a one-liner. Each value stores the id of the key that sealed it, so a Keyring enables online key rotation: old values keep decrypting under their old key while new writes use the active key. Key sources are separate modules implementing Keyring, so their SDKs aren't pulled in unless used: middleware/crypto/age (age-wrapped data keys), middleware/crypto/kms/aws (AWS KMS) and middleware/crypto/kms/gcp (Google Cloud KMS).
  • compress — transparently compresses values (zstd or snappy) with a per-value codec marker; incompressible/tiny values are stored uncompressed so a value is never inflated. Compose it outside crypto (compress(crypto(base))) so plaintext is compressed before it is encrypted.
  • otel — OpenTelemetry observability: a span per operation plus metrics (operation counter store.operations, error counter store.errors, latency histogram store.operation.duration). Bean/op are recorded; values never are. Prometheus users scrape via the OTel Prometheus bridge.
base, _ := bboltstore.New("config", "config.db", 0600)
enc, _  := cryptostore.New(base.Interface(), key) // values now encrypted at rest
s       := otelstore.New(enc)                      // + tracing

// rotation: add a new key, make it active; old values still readable
kr := cryptostore.NewStaticKeyring().Add(1, oldKey)
enc, _ = cryptostore.NewWithKeyring(base.Interface(), kr)
kr.Add(2, newKey); kr.SetActive(2)               // new writes use key 2

Testing & benchmarks

storetest is the shared, capability-gated suite every provider must pass — a behavior is asserted only if the provider advertises the matching capability:

  • storetest.RunConformance(t, factory) — correctness (get/set/remove, TTL and expiry, batch + atomic batch, CAS/version, ordered enumeration, range (forward and reverse), pagination, sweep, watch).
  • storetest.RunRotation(t, factory) — online key rotation for encrypted stores (pre-rotation values still decrypt; new writes use the active key).
  • storetest.RunBenchmarks(b, factory) — the shared benchmark suite (Set, Get, Batch, Increment, CompareAndSet, Enumerate, Range, Sweep), gated the same way.

The benchmarks module wires RunBenchmarks to every engine plus a crypto key rotation benchmark, so they can be compared head-to-head:

go test -run '^$' -bench . -benchmem ./benchmarks/...      # all engines
go test -run '^$' -bench BenchmarkNutsdb/Set ./benchmarks/...

CI runs the conformance suite per module on every push/PR (.github/workflows/build.yml); benchmarks are heavy and noisy, so they run on demand and weekly via a separate workflow (.github/workflows/benchmark.yml) that publishes results as an artifact.

Releasing

All modules are versioned together with release.sh. See RELEASING.md.

./release.sh --dry-run v1.3.0   # preview
./release.sh v1.3.0             # tag + push every module

Documentation

Index

Constants

View Source 
const EnvelopeHeaderLen = 1 + 8 + 8

EnvelopeHeaderLen is the fixed size of the envelope header in bytes.

View Source 
const NoExpiry = -1

NoExpiry, passed as a TTL, pins a record: it overrides any region default TTL so the record never expires. (Distinct from NoTTL/0, which means "unspecified" and therefore takes the region default when one is configured.)

View Source 
const NoTTL = 0

* Marker that TTL (time-to-live in seconds) is not defined, therefore not setup, meaning eternal record

View Source 
const SystemRegion = "_system"

SystemRegion holds a major key's own bookkeeping (usage counters, manifests). It is never metered (that would count the meter's own writes).

Variables

View Source 
var (
	ErrNotFound = os.ErrNotExist

	// ErrInvalidRequest is returned if the user request is invalid.
	ErrInvalidRequest = xerrors.New("invalid request")

	// ErrConcurrentTransaction is returned when a transaction conflicts with another transaction.
	ErrConcurrentTxn = xerrors.New("concurrent transaction, try again")

	// ErrReadOnlyTxn is returned if an update function is called on a read-only transaction.
	ErrReadOnlyTxn = xerrors.New("read-only transaction has update operation")

	// ErrDiscardedTxn is returned if a previously discarded transaction is re-used.
	ErrDiscardedTxn = xerrors.New("transaction has been discarded")

	// ErrCanceledTxn is returned if user canceled transaction.
	ErrCanceledTxn = xerrors.New("transaction has been canceled")

	// ErrTooBigTxn is returned if too many writes are fit into a single transaction.
	ErrTooBigTxn = xerrors.New("transaction is too big")

	// ErrEmptyKey is returned if an empty key is passed on an update function.
	ErrEmptyKey = xerrors.New("empty key")

	// ErrInvalidKey is returned if the key has wrong character(s)
	ErrInvalidKey = xerrors.New("key is invalid")

	// ErrAlreadyClosed is returned when store is already closed
	ErrAlreadyClosed = xerrors.New("already closed")

	// ErrInternal
	ErrInternal = xerrors.New("internal error")

	// ErrNotSupported is returned when an operation is not supported by the backend capabilities.
	ErrNotSupported = xerrors.New("operation not supported by this store")
)
View Source 
var DataStoreClass = reflect.TypeOf((*DataStore)(nil)).Elem()
View Source 
var DataStoreManagerClass = reflect.TypeOf((*DataStoreManager)(nil)).Elem()
View Source 
var DefaultBatchSize = 256

* Default batch size, could be overwritten

View Source 
var ManagedDataStoreClass = reflect.TypeOf((*ManagedDataStore)(nil)).Elem()
View Source 
var ManagedTransactionalDataStoreClass = reflect.TypeOf((*ManagedTransactionalDataStore)(nil)).Elem()
View Source 
var MultiDataStoreClass = reflect.TypeOf((*MultiDataStore)(nil)).Elem()
View Source 
var SweepableClass = reflect.TypeOf((*Sweepable)(nil)).Elem()
View Source 
var TransactionClass = reflect.TypeOf((*Transaction)(nil)).Elem()
View Source 
var TransactionalDataStoreClass = reflect.TypeOf((*TransactionalDataStore)(nil)).Elem()
View Source 
var TransactionalManagerClass = reflect.TypeOf((*TransactionalManager)(nil)).Elem()
View Source 
var WatchBufferSize = 256

WatchBufferSize is the per-subscriber channel buffer used by WatchHub.

Functions

func DecodeEnvelope 

func DecodeEnvelope(raw []byte) (version, expiresAtUnix int64, value []byte, ok bool)

DecodeEnvelope unwraps an envelope. If raw is not an envelope (legacy value), it returns ok=false with the original bytes as the value and zero metadata. The returned value aliases raw; copy it if it must outlive raw's storage.

func EncodeEnvelope 

func EncodeEnvelope(version, expiresAtUnix int64, value []byte) []byte

EncodeEnvelope wraps value with version and expiry metadata.

func ExpiryFromTtl 

func ExpiryFromTtl(ttlSeconds int) int64

ExpiryFromTtl converts a relative ttl (seconds) into an absolute unix timestamp. ttlSeconds <= 0 (NoTTL) yields 0, meaning no expiry.

func Get 

func Get[T any](ctx context.Context, ds DataStore, key []byte, codec Codec[T]) (value T, found bool, err error)

Get loads and decodes the value at key. found is false (with the zero T) when the key is absent.

func IsExpired 

func IsExpired(expiresAtUnix int64) bool

IsExpired reports whether an entry with the given absolute expiry is expired now.

func Put 

func Put[T any](ctx context.Context, ds DataStore, key []byte, value T, codec Codec[T], ttlSeconds int) error

Put encodes and stores value at key with the given ttl (NoTTL for none).

func RegionUsage added in v1.4.0 

func RegionUsage(m MultiDataStore, majorKey, region string) int64

RegionUsage returns a major key's live byte footprint for one region.

func StartSweeper 

func StartSweeper(ds DataStore, interval time.Duration, onError ...func(error)) (stop func(), err error)

func TtlFromExpiry 

func TtlFromExpiry(expiresAtUnix int64) int

TtlFromExpiry converts an absolute expiry timestamp into the remaining ttl in seconds. It returns NoTTL (0) when there is no expiry; a value with an expiry that has essentially no time left reports -1 to stay distinguishable from NoTTL.

func Usage added in v1.4.0 

func Usage(m MultiDataStore, majorKey string) int64

Usage returns a major key's total live byte footprint recorded by WithMetering (0 when unmetered or empty).

func WithTransaction 

func WithTransaction(ctx context.Context, beanName string, tx Transaction) context.Context

Types

type BatchOperation 

type BatchOperation struct {
	DataStore                 // should be initialized
	Context   context.Context // should be initialized
	// contains filtered or unexported fields
}

BatchOperation accumulates entries and writes them in one SetBatchRaw call. The batch is atomic only on stores reporting BatchAtomicCapability.

func (*BatchOperation) Add 

func (t *BatchOperation) Add(entry RawEntry) *BatchOperation

Add appends a pre-built entry (Key, Value, Ttl are used).

func (*BatchOperation) Do 

func (t *BatchOperation) Do() error

Do writes all accumulated entries via SetBatchRaw.

func (*BatchOperation) Len 

func (t *BatchOperation) Len() int

Len returns the number of accumulated entries.

func (*BatchOperation) Put 

func (t *BatchOperation) Put(key, value []byte, ttlSeconds int) *BatchOperation

Put appends a binary key/value with optional ttl (NoTTL for none).

func (*BatchOperation) PutCounter 

func (t *BatchOperation) PutCounter(value uint64, ttlSeconds int, formatKey string, args ...interface{}) *BatchOperation

PutCounter appends a uint64 counter value under a formatted key.

func (*BatchOperation) PutProto 

func (t *BatchOperation) PutProto(msg proto.Message, ttlSeconds int, formatKey string, args ...interface{}) error

PutProto appends a marshaled protobuf message under a formatted key.

func (*BatchOperation) PutString 

func (t *BatchOperation) PutString(value string, ttlSeconds int, formatKey string, args ...interface{}) *BatchOperation

PutString appends a string value under a formatted key.

type Capability 

type Capability uint64
const (
	// TTLCapability means SetRaw/TouchRaw honor ttlSeconds and entries expire.
	TTLCapability Capability = 1 << iota

	// AtomicCapability means CompareAndSetRaw and IncrementRaw are atomic and
	// version-aware (GetRaw returns a meaningful, changing Version).
	AtomicCapability

	// TransactionCapability means the store implements TransactionalManager.
	TransactionCapability

	// EncryptedCapability means values are encrypted at rest.
	EncryptedCapability

	// OrderedCapability means EnumerateRaw returns keys in sorted (lexical) order.
	OrderedCapability

	// WatchCapability means WatchRaw delivers change notifications.
	WatchCapability

	// BatchAtomicCapability means SetBatchRaw is all-or-nothing (a failed batch
	// leaves no partial writes). Without it, SetBatchRaw is still correct but a
	// failure mid-way may leave some entries written.
	BatchAtomicCapability
)

func (Capability) Has 

func (c Capability) Has(other Capability) bool

Has reports whether all the bits in other are set in c.

func (Capability) String 

func (c Capability) String() string

String renders the capability set as a pipe-separated list, e.g. "TTL|Atomic|Ordered".

type Codec 

type Codec[T any] interface {
	Encode(value T) ([]byte, error)
	Decode(data []byte) (T, error)
}

Codec encodes/decodes values of type T to and from the stored byte form.

func JSON 

func JSON[T any]() Codec[T]

JSON returns a Codec[T] using encoding/json.

func MsgPack 

func MsgPack[T any]() Codec[T]

MsgPack returns a Codec[T] using MessagePack (github.com/vmihailenco/msgpack). Good for compact, schema-less encoding of arbitrary Go values.

func Proto 

func Proto[T proto.Message]() Codec[T]

Proto returns a Codec[T] using protobuf. T must be a concrete pointer message type, e.g. store.Proto[*pb.User]().

type CompareAndSetOperation 

type CompareAndSetOperation struct {
	DataStore                 // should be initialized
	Context   context.Context // should be initialized
	// contains filtered or unexported fields
}

func (*CompareAndSetOperation) Binary 

func (t *CompareAndSetOperation) Binary(value []byte) (bool, error)

func (*CompareAndSetOperation) ByKey 

func (t *CompareAndSetOperation) ByKey(formatKey string, args ...interface{}) *CompareAndSetOperation

func (*CompareAndSetOperation) ByRawKey 

func (t *CompareAndSetOperation) ByRawKey(key []byte) *CompareAndSetOperation

func (*CompareAndSetOperation) Counter 

func (t *CompareAndSetOperation) Counter(value uint64) (bool, error)

func (*CompareAndSetOperation) Proto 

func (t *CompareAndSetOperation) Proto(msg proto.Message) (bool, error)

func (*CompareAndSetOperation) String 

func (t *CompareAndSetOperation) String(value string) (bool, error)

func (*CompareAndSetOperation) WithTtl 

func (t *CompareAndSetOperation) WithTtl(ttlSeconds int) *CompareAndSetOperation

func (*CompareAndSetOperation) WithVersion 

func (t *CompareAndSetOperation) WithVersion(version int64) *CompareAndSetOperation

type CounterEntry 

type CounterEntry struct {
	Key     []byte
	Value   uint64
	Ttl     int
	Version int64
}

type DataStore 

type DataStore interface {
	glue.DisposableBean
	glue.NamedBean

	Features() Capability

	Get(ctx context.Context) *GetOperation

	Set(ctx context.Context) *SetOperation

	Batch(ctx context.Context) *BatchOperation

	// equivalent of i++ operation, always returns previous value
	Increment(ctx context.Context) *IncrementOperation

	CompareAndSet(ctx context.Context) *CompareAndSetOperation

	Touch(ctx context.Context) *TouchOperation

	Remove(ctx context.Context) *RemoveOperation

	Enumerate(ctx context.Context) *EnumerateOperation

	Watch(ctx context.Context) *WatchOperation

	GetRaw(ctx context.Context, key []byte, ttlPtr *int, versionPtr *int64, required bool) ([]byte, error)

	SetRaw(ctx context.Context, key, value []byte, ttlSeconds int) error

	SetBatchRaw(ctx context.Context, entries []RawEntry) error

	CompareAndSetRaw(ctx context.Context, key, value []byte, ttlSeconds int, version int64) (bool, error)

	IncrementRaw(ctx context.Context, key []byte, initial, delta int64, ttlSeconds int) (int64, error)

	TouchRaw(ctx context.Context, key []byte, ttlSeconds int) error

	RemoveRaw(ctx context.Context, key []byte) error

	EnumerateRaw(ctx context.Context, prefix, seek []byte, batchSize int, onlyKeys bool, reverse bool, cb func(*RawEntry) bool) error

	WatchRaw(ctx context.Context, prefix []byte, cb func(*WatchEvent) bool) error
}

type DataStoreManager 

type DataStoreManager interface {
	Compact(discardRatio float64) error

	Backup(w io.Writer, since uint64) (uint64, error)

	Restore(r io.Reader) error

	DropAll() error

	DropWithPrefix(prefix []byte) error
}

type EnumerateOperation 

type EnumerateOperation struct {
	DataStore                 // should be initialized
	Context   context.Context // should be initialized
	// contains filtered or unexported fields
}

func (*EnumerateOperation) After 

func (t *EnumerateOperation) After(token []byte) *EnumerateOperation

After resumes enumeration strictly after the given continuation token (the last key from a previous page). A nil/empty token starts from the beginning.

func (*EnumerateOperation) ByPrefix 

func (t *EnumerateOperation) ByPrefix(formatPrefix string, args ...interface{}) *EnumerateOperation

func (*EnumerateOperation) ByRawPrefix 

func (t *EnumerateOperation) ByRawPrefix(prefix []byte) *EnumerateOperation

func (*EnumerateOperation) Do 

func (t *EnumerateOperation) Do(cb func(*RawEntry) bool) error

Do enumerates all matching entries (honoring Range/After/Limit if set).

func (*EnumerateOperation) DoCounters 

func (t *EnumerateOperation) DoCounters(cb func(*CounterEntry) bool) (err error)

func (*EnumerateOperation) DoPage 

func (t *EnumerateOperation) DoPage(cb func(*RawEntry) bool) (next []byte, err error)

DoPage is like Do but returns an opaque continuation token to pass to After for the next page. The token is nil when the range has been exhausted.

func (*EnumerateOperation) DoProto 

func (t *EnumerateOperation) DoProto(factory func() proto.Message, cb func(*ProtoEntry) bool) error

func (*EnumerateOperation) Limit 

func (t *EnumerateOperation) Limit(n int) *EnumerateOperation

Limit caps the number of entries returned by Do/DoPage (a page size).

func (*EnumerateOperation) OnlyKeys 

func (t *EnumerateOperation) OnlyKeys() *EnumerateOperation

func (*EnumerateOperation) Range 

func (t *EnumerateOperation) Range(from, to []byte) *EnumerateOperation

Range restricts enumeration to the half-open key interval [from, to): from is inclusive (the start position), to is exclusive (nil means unbounded). Requires an ordered store (OrderedCapability), otherwise Do/DoPage return ErrNotSupported.

func (*EnumerateOperation) Reverse 

func (t *EnumerateOperation) Reverse() *EnumerateOperation

func (*EnumerateOperation) Seek 

func (t *EnumerateOperation) Seek(formatSeek string, args ...interface{}) *EnumerateOperation

func (*EnumerateOperation) WithBatchSize 

func (t *EnumerateOperation) WithBatchSize(batchSize int) *EnumerateOperation

type GetOperation 

type GetOperation struct {
	DataStore                 // should be initialized
	Context   context.Context // should be initialized
	// contains filtered or unexported fields
}

func (*GetOperation) ByKey 

func (t *GetOperation) ByKey(formatKey string, args ...interface{}) *GetOperation

func (*GetOperation) ByRawKey 

func (t *GetOperation) ByRawKey(key []byte) *GetOperation

func (*GetOperation) Required 

func (t *GetOperation) Required() *GetOperation

func (*GetOperation) ToBinary 

func (t *GetOperation) ToBinary() ([]byte, error)

func (*GetOperation) ToCounter 

func (t *GetOperation) ToCounter() (uint64, error)

func (*GetOperation) ToCounterEntry 

func (t *GetOperation) ToCounterEntry() (entry CounterEntry, err error)

func (*GetOperation) ToEntry 

func (t *GetOperation) ToEntry() (entry RawEntry, err error)

func (*GetOperation) ToProto 

func (t *GetOperation) ToProto(container proto.Message) error

func (*GetOperation) ToProtoEntry 

func (t *GetOperation) ToProtoEntry(factory func() proto.Message) (entry ProtoEntry, err error)

func (*GetOperation) ToString 

func (t *GetOperation) ToString() (string, error)

func (*GetOperation) WithTtl 

func (t *GetOperation) WithTtl(ttl *int) *GetOperation

func (*GetOperation) WithVersion 

func (t *GetOperation) WithVersion(version *int64) *GetOperation

type IncrementOperation 

type IncrementOperation struct {
	DataStore                 // should be initialized
	Context   context.Context // should be initialized

	Initial int64
	Delta   int64 // should be initialized by 1
	// contains filtered or unexported fields
}

func (*IncrementOperation) ByKey 

func (t *IncrementOperation) ByKey(formatKey string, args ...interface{}) *IncrementOperation

func (*IncrementOperation) ByRawKey 

func (t *IncrementOperation) ByRawKey(key []byte) *IncrementOperation

func (*IncrementOperation) Do 

func (t *IncrementOperation) Do() (prev int64, err error)

func (*IncrementOperation) WithDelta 

func (t *IncrementOperation) WithDelta(delta int64) *IncrementOperation

func (*IncrementOperation) WithInitialValue 

func (t *IncrementOperation) WithInitialValue(initial int64) *IncrementOperation

func (*IncrementOperation) WithTtl 

func (t *IncrementOperation) WithTtl(ttlSeconds int) *IncrementOperation

type KeyStrategy added in v1.3.3 

type KeyStrategy interface {
	Prefix(majorKey, region string) []byte

	TenantPrefix(majorKey string) []byte
}

func LenPrefix added in v1.3.3 

func LenPrefix() KeyStrategy

LenPrefix returns the binary-safe KeyStrategy matching the consensusdb backing store's header convention: 

[2B big-endian len][majorKey][2B big-endian len][region][minorKey]

No separator and no escaping, so any bytes are allowed in majorKey/region, with the minor key as the raw tail.

func Separator added in v1.3.3 

func Separator(sep string) KeyStrategy

Separator returns the readable KeyStrategy "major<sep>region<sep>minor" — e.g. Separator(":") lays out "acme:USERS:alice". majorKey and region must not contain sep (the separator is not escaped); use LenPrefix for binary-safe names. An empty sep defaults to ":".

type ManagedDataStore 

type ManagedDataStore interface {
	DataStore
	DataStoreManager

	Instance() interface{}
}

type ManagedTransactionalDataStore 

type ManagedTransactionalDataStore interface {
	DataStore
	DataStoreManager
	TransactionalManager
}

type MultiDataStore added in v1.3.3 

type MultiDataStore interface {
	glue.DisposableBean
	glue.NamedBean

	Features() Capability

	Region(majorKey, region string) DataStore
}

func NewMulti added in v1.3.3 

func NewMulti(name string, inner DataStore, strategy KeyStrategy) MultiDataStore

func NewSharding added in v1.4.0 

func NewSharding(name string, shards ...MultiDataStore) MultiDataStore

NewSharding routes major keys across shards. With a single shard it is a thin passthrough (the "sharding-ready, not yet sharded" configuration).

Each shard's BeanName is its routing identity, so it must be non-empty, unique, and stable across restarts and config edits — renaming a shard reassigns its major keys. Misconfiguration panics here, at wiring time, not on first use.

func WithMetering added in v1.4.0 

func WithMetering(m MultiDataStore) MultiDataStore

WithMetering wraps m so every write outside SystemRegion adjusts the owner's live byte counters (total and per-region) in its SystemRegion.

func WithRegionTTL added in v1.4.0 

func WithRegionTTL(m MultiDataStore, regionTTL map[string]int) MultiDataStore

WithRegionTTL wraps m so writes to a region named in regionTTL get that default TTL (seconds) unless the caller passes an explicit TTL or NoExpiry. Regions absent from the map are returned unchanged.

type ProtoEntry 

type ProtoEntry struct {
	Key     []byte
	Value   proto.Message
	Ttl     int
	Version int64
}

type RawEntry 

type RawEntry struct {
	Key     []byte
	Value   []byte
	Ttl     int
	Version int64
}

type RemoveOperation 

type RemoveOperation struct {
	DataStore                 // should be initialized
	Context   context.Context // should be initialized
	// contains filtered or unexported fields
}

func (*RemoveOperation) ByKey 

func (t *RemoveOperation) ByKey(formatKey string, args ...interface{}) *RemoveOperation

func (*RemoveOperation) ByRawKey 

func (t *RemoveOperation) ByRawKey(key []byte) *RemoveOperation

func (*RemoveOperation) Do 

func (t *RemoveOperation) Do() error

type SetOperation 

type SetOperation struct {
	DataStore                 // should be initialized
	Context   context.Context // should be initialized
	// contains filtered or unexported fields
}

func (*SetOperation) Binary 

func (t *SetOperation) Binary(value []byte) error

func (*SetOperation) ByKey 

func (t *SetOperation) ByKey(formatKey string, args ...interface{}) *SetOperation

func (*SetOperation) ByRawKey 

func (t *SetOperation) ByRawKey(key []byte) *SetOperation

func (*SetOperation) Counter 

func (t *SetOperation) Counter(value uint64) error

func (*SetOperation) Proto 

func (t *SetOperation) Proto(msg proto.Message) error

func (*SetOperation) String 

func (t *SetOperation) String(value string) error

func (*SetOperation) WithTtl 

func (t *SetOperation) WithTtl(ttlSeconds int) *SetOperation

type StripedMutex 

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

func (*StripedMutex) Lock 

func (t *StripedMutex) Lock(key []byte)

Lock locks the stripe owning key.

func (*StripedMutex) LockAll 

func (t *StripedMutex) LockAll() (unlock func())

LockAll locks every stripe (ascending) and returns the unlock function. Used by whole-store operations such as DropAll.

func (*StripedMutex) LockMany 

func (t *StripedMutex) LockMany(keys ...[]byte) (unlock func())

LockMany locks the distinct stripes owning the given keys in ascending order and returns the unlock function (release in reverse order).

func (*StripedMutex) Unlock 

func (t *StripedMutex) Unlock(key []byte)

Unlock unlocks the stripe owning key.

type Sweepable 

type Sweepable interface {
	SweepExpired(ctx context.Context) (removed int, err error)
}

type TouchOperation 

type TouchOperation struct {
	DataStore                 // should be initialized
	Context   context.Context // should be initialized
	// contains filtered or unexported fields
}

func (*TouchOperation) ByKey 

func (t *TouchOperation) ByKey(formatKey string, args ...interface{}) *TouchOperation

func (*TouchOperation) ByRawKey 

func (t *TouchOperation) ByRawKey(key []byte) *TouchOperation

func (*TouchOperation) Do 

func (t *TouchOperation) Do() error

func (*TouchOperation) WithTtl 

func (t *TouchOperation) WithTtl(ttlSeconds int) *TouchOperation

type Transaction 

type Transaction interface {
	ReadOnly() bool

	Commit() error

	Rollback()

	Instance() interface{}
}

func GetTransaction 

func GetTransaction(ctx context.Context, beanName string) (Transaction, bool)

func NewInnerTransaction 

func NewInnerTransaction(parent Transaction) Transaction

type TransactionalDataStore 

type TransactionalDataStore interface {
	DataStore
	TransactionalManager
}

type TransactionalManager 

type TransactionalManager interface {
	BeginTransaction(ctx context.Context, readOnly bool) context.Context

	// commit if errOps is nil or rollback otherwise
	EndTransaction(ctx context.Context, errOps error) error
}

type Typed 

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

Typed binds a DataStore and a Codec[T] for ergonomic repeated typed access.

func Of 

func Of[T any](ds DataStore, codec Codec[T]) Typed[T]

Of creates a typed view over ds using codec.

func (Typed[T]) Enumerate 

func (t Typed[T]) Enumerate(ctx context.Context, prefix []byte, cb func(key []byte, value T) bool) error

Enumerate decodes every value under prefix, invoking cb until it returns false.

func (Typed[T]) Get 

func (t Typed[T]) Get(ctx context.Context, key []byte) (T, bool, error)

func (Typed[T]) Put 

func (t Typed[T]) Put(ctx context.Context, key []byte, value T, ttlSeconds int) error

func (Typed[T]) Remove 

func (t Typed[T]) Remove(ctx context.Context, key []byte) error

type WatchEvent 

type WatchEvent struct {
	Key     []byte
	Value   []byte // nil for WatchDelete
	Type    WatchEventType
	Version int64
}

WatchEvent is delivered by WatchRaw for each change matching the watched prefix.

type WatchEventType 

type WatchEventType int
const (
	// WatchSet indicates a key was created or updated.
	WatchSet WatchEventType = iota
	// WatchDelete indicates a key was removed (or expired).
	WatchDelete
)

func (WatchEventType) String 

func (t WatchEventType) String() string

type WatchHub 

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

func NewWatchHub 

func NewWatchHub() *WatchHub

func (*WatchHub) Notify 

func (t *WatchHub) Notify(event *WatchEvent)

Notify fans an event out to every subscriber whose prefix matches the key.

func (*WatchHub) Watch 

func (t *WatchHub) Watch(ctx context.Context, prefix []byte, cb func(*WatchEvent) bool) error

Watch blocks delivering matching events to cb until cb returns false or ctx is done. It is the WatchRaw implementation for hub-backed providers.

type WatchOperation 

type WatchOperation struct {
	DataStore                 // should be initialized
	Context   context.Context // should be initialized
	// contains filtered or unexported fields
}

func (*WatchOperation) ByPrefix 

func (t *WatchOperation) ByPrefix(formatPrefix string, args ...interface{}) *WatchOperation

func (*WatchOperation) ByRawPrefix 

func (t *WatchOperation) ByRawPrefix(prefix []byte) *WatchOperation

func (*WatchOperation) Do 

func (t *WatchOperation) Do(cb func(*WatchEvent) bool) error

Do blocks delivering events to cb until cb returns false or the context is cancelled. A nil/empty prefix watches every key.

Source Files

View all Source files

Directories

Path Synopsis
middleware
crypto module
providers
badger module
bbolt module
bolt module
cdb module
mem module
nutsdb module
pebble module
rosedb module
storetest module

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.