objstore

README

objstore

go get github.com/axiomhq/objstore

Object storage with compare-and-swap, for Go. One Store over an S3 bucket (AWS, MinIO, any S3-compatible service) or a durable local directory, with:

• conditional PUT and GET, and the ETag with the body
• paginated list, delimited list, batch delete
• a file:// store that fsyncs every object and writes large ones back in chunks, so a small urgent write never waits behind a big one
• envelope encryption per name, with local, AWS KMS and GCP KMS key providers (objstore/kms)
• fault injection and metering for tests (objstore/storetest)
• a write-ahead log: one entry per second, conditional PUT, nonce read-back (objstore/wal)
• a memory and disk object cache with singleflight and per-request stats (objstore/cache)
• a ranged reader that coalesces nearby reads into one GET (objstore/rangeread)
• a lease for leader election or a single writer: acquire, renew, fence (objstore/lease)

Use

1. Open a store: s, err := objstore.New(ctx, "https://s3.us-east-1.amazonaws.com", "my-bucket"). Use "file:///var/lib/data" for a local store. objstore.NewConfigured(ctx, objstore.Config{...}) adds SSE, rate limits, timeouts and a write bound.
2. Create the bucket if you need to: s.EnsureBucket(ctx).
3. Write once: created, err := s.PutIfAbsent(ctx, key, data). created is false when the key existed.
4. Swap: body, etag, err := s.GetWithETag(ctx, key), then ok, err := s.PutIfMatch(ctx, key, next, etag). ok false means someone else won: re-read and retry.
5. Poll cheaply: body, etag, changed, err := s.GetIfChanged(ctx, key, etag).

A missing key returns an error wrapping objstore.ErrNotFound. Mark a context with objstore.Urgent(ctx) to skip pacing and the write bound, for heartbeats and log commits.

Methods

method	does
Put(ctx, key, data)	write, overwriting
PutIfAbsent(ctx, key, data) (bool, error)	write only if the key is absent
PutIfMatch(ctx, key, data, etag) (bool, error)	write only if the ETag still matches
Get(ctx, key)	read the object
GetRange(ctx, key, off, n)	read n bytes from off
GetWithETag(ctx, key)	read the object and its ETag
GetIfChanged(ctx, key, etag)	read only if the ETag moved
List(ctx, prefix)	every key under a prefix
ListPage(ctx, prefix, after, limit)	one page of keys (limit ≤ MaxListPage, 1,000)
ListPrefixes(ctx, prefix)	child prefixes under a prefix (S3 delimiter /)
ListPrefixesPage(ctx, prefix, after, limit)	one page of child prefixes
Delete(ctx, key)	delete one key
DeleteMany(ctx, keys...)	batch delete
EnsureBucket(ctx), DropBucket(ctx)	create or remove the bucket

ctx = objstore.WithTimings(ctx, &t) has the store add the wall time of every call made under ctx to t, per call kind; t.LogAttrs() returns them as slog arguments.

Write-ahead log

1. Make your row a wal.Record: Size() int and AppendTo(b []byte) ([]byte, error). wal.Bytes is one already.
2. Start a writer: w := wal.NewWriter(s, "log/", 1, onCommit). onCommit(seq, at, records) runs for each durable entry, before callers are acked.
3. Write: err := w.Append(ctx, records). It returns once the entry holding them is durable.
4. Read back: wal.Walk(ctx, s, "log/", after, 0, wal.Decode, visit). visit gets one wal.Entry per entry; a batch cut into several pages arrives whole.
5. Stop: w.Close() drains, then returns.

error	means
ErrOverloaded	128 MiB is already waiting for the store; retry later
ErrLostRace	another writer took the sequence (split brain)
ErrUnresolved	the outcome is unknown; reopen from your checkpoint and the log
ErrCorrupt	a page failed its checksum or framing, or a page is missing inside a bounded walk

Concurrent Appends share one entry, at most one per WithCommitInterval (default 1 s). An idle writer commits at once. Entries over 32 MiB are cut into pages. Nothing is listed: the walk GETs after+1, after+2, ... and stops at the first missing key. SetFloor refuses claims at or below a truncated watermark.

Cache and ranged reads

1. Build the tiers: disk, _ := cache.NewDisk(dir, 10<<30), then c := cache.New(s, 1<<30, disk, cache.Keys{}).
2. Read through them: b, err := c.FetchWith(ctx, key, func(ctx context.Context) ([]byte, error) { return s.Get(ctx, key) }).
3. For byte ranges: cfg, _ := rangeread.Config{}.Normalized(), r := rangeread.New(s, c, cfg), then scope, err := r.FetchRanges(ctx, loads) and cache.Scoped(scope, key) per load.

piece	does
cache.ByteCache	striped LRU under one byte budget; decoded values ride on their bytes (PutDecoded, charged via cache.Sizer)
cache.Disk	disposable disk tier, 4 KiB block checksums, pins, inactivity expiry
cache.Keys	tells the cache which keys are log pages (own share), low priority, or ranged
cache.WithRequestStats	per-request lookups through every tier as memory hits, disk hits and loads (ClassCounts, HitRatio); Cache.ClassCounts is the process's
rangeread.Plan	unions and coalesces extents under gap, extra-byte and range limits

Cache.Put writes through to the memory and disk tiers, so ranged reads of an object this process wrote are cache hits. Objects are immutable: the cache never re-validates a key. A key under ns/<name>/ belongs to namespace <name> for InvalidateNamespace, pins and expiry.

Leader election / single writer on S3

1. Acquire: l, err := lease.Acquire(ctx, s, "jobs/leader", lease.OwnerID(), lease.DefaultTTL). ErrNotOwner means someone else holds it; retry later.
2. Say what losing it costs: l.Start(func() { cancelWork() }). The lease renews itself every TTL/4 from Acquire on.
3. Before every guarded action: if err := l.Valid(); err != nil { stop }.
4. Hand it over when done: l.Release(). The next process takes it at once instead of waiting out the TTL.

event	what happens
renewal CAS loses to another owner	ErrNotOwner, the lease fences: your Start callback runs once, Valid refuses from then on
renewal fails without proof (timeout, 5xx)	one immediate retry; still a holder until the local deadline, fenced once it passes
holder partitioned or crashed	nobody takes the key until its stored expiry plus TTL/2 (the clock-skew margin)
PUT landed but the answer was lost	the object is read back: our own nonce there means held, not lost
an earlier lost PUT lands late	its nonce was recorded as pending, so the renewal adopts it instead of fencing itself

Every write carries a fresh nonce, so the read-back tells "my write landed" from "someone else's did" without guessing. The holder's deadline runs on its local monotonic clock from before the PUT, so it stops acting no later than a taker may start. Same TTL across the fleet, clock error below TTL/2.

• lease.Shared shares one lease among several holders in a process by reference count; the last Release hands it back.
• l.CheckHeadOnRenewal(fn) runs fn after each renewal; its error counts as the renewal's.
• A lease check cannot fence a write already in flight. Guard durable writes with their own CAS (the ETag, or a sequence), not a post-check.

Encryption

1. Pick a provider: kms.LocalFile{Dir: dir}, kms.AWSKMS{Client: c}, kms.GCPKMS{Client: c}, or kms.Router{Local, AWS, GCP, Default} to route by key name.
2. Turn it on: s.ConfigureCMEK(provider). Every object under ns/<name>/ is now encrypted with that name's data key.
3. Before the first write for a name, wrap a fresh 32-byte key with provider.Wrap and install it: s.InstallNamespaceKey(ctx, name, objstore.Envelope{Mode: objstore.EncryptionCustomerManaged, KeyName, KeyVersion, DEKWrapped}).
4. Rotate with s.RotateNamespaceKey(ctx, name). It re-wraps the data key, so no object is rewritten.
5. Check access with s.CheckNamespaceKey(ctx, name). A revoked key returns kms.ErrKeyUnavailable.

Tests with faults

1. s := storetest.New(t) gives a fresh bucket per test, dropped at cleanup.
2. s, f := storetest.NewFaulty(t) wraps it in a *storetest.Fault.
3. Arm one crash point: f.Set(storetest.Plan{Op: storetest.OpPut, N: 2, Key: "manifest", Mode: storetest.Ambiguous}). Fail errors before the write, Ambiguous writes and then errors, Hang blocks until cancel, Pause blocks until f.Resume().
4. Read what happened: f.Fired(), f.Ops(), f.ReadKeys(), f.WriteKeys(), f.ReadBytes(), f.WriteBytes().

f.SetShape(storetest.Shape{Latency, BytesPerSecond, ErrorRate, Seed}) adds seeded latency, bandwidth and errors to every call. f.WatchRewrites() flags any key rewritten with different bytes.

Test

go test -race ./...
OBJSTORE_TEST_S3=http://localhost:9000 go test -race ./...   # the same suite against MinIO

Without OBJSTORE_TEST_S3 every suite runs on a file:// bucket in a temp directory; the S3 client's error and paging mapping runs against a fake server either way.

License

MIT, see LICENSE.

Documentation

Overview

Package objstore is one object-storage bucket with compare-and-swap: conditional PUT (PutIfAbsent, PutIfMatch) and GET (GetIfChanged), the ETag with the body, paginated listing and batch delete. New picks the backend from the endpoint: an S3 URL (AWS, MinIO, any S3-compatible service) or file:// for a durable local store with chunked write-back.

Objects are meant to be written once and never mutated except by deletion; the conditional writes cover the few that must change. ListPrefixes returns the child prefixes under a prefix (S3's delimiter), so discovering a thousand top-level names costs one request, not one per object.

ConfigureCMEK adds envelope encryption of every object under ns/<name>/ with a per-name data key wrapped by a customer key (see package kms for the providers).

Package wal is a write-ahead log on a Store, package cache a memory and disk cache in front of one, package rangeread a coalescing ranged reader over the cache, package lease a single-holder lease on one key.

Index

• Constants
• Variables
• func Urgent(ctx context.Context) context.Context
• func WithTimings(ctx context.Context, t *Timings) context.Context
• type Backend
• type Call
• type Config
• type Envelope
• type Store
  • func New(ctx context.Context, endpoint, bucket string) (*Store, error)
  • func NewConfigured(ctx context.Context, cfg Config) (*Store, error)
  • func (s *Store) CheckNamespaceKey(ctx context.Context, name string) error
  • func (s *Store) CheckNamespaceKeyCached(ctx context.Context, name string) error
  • func (s *Store) ConfigureCMEK(provider kms.KeyProvider)
  • func (s *Store) Delete(ctx context.Context, key string) error
  • func (s *Store) DeleteMany(ctx context.Context, keys ...string) error
  • func (s *Store) DropBucket(ctx context.Context) error
  • func (s *Store) EnsureBucket(ctx context.Context) error
  • func (s *Store) Get(ctx context.Context, key string) ([]byte, error)
  • func (s *Store) GetIfChanged(ctx context.Context, key, etag string) ([]byte, string, bool, error)
  • func (s *Store) GetRange(ctx context.Context, key string, offset, length int64) ([]byte, error)
  • func (s *Store) GetWithETag(ctx context.Context, key string) ([]byte, string, error)
  • func (s *Store) InstallNamespaceKey(ctx context.Context, name string, envelope Envelope) error
  • func (s *Store) List(ctx context.Context, prefix string) ([]string, error)
  • func (s *Store) ListPage(ctx context.Context, prefix, after string, limit int) ([]string, string, error)
  • func (s *Store) ListPrefixes(ctx context.Context, prefix string) ([]string, error)
  • func (s *Store) ListPrefixesPage(ctx context.Context, prefix, after string, limit int) ([]string, string, error)
  • func (s *Store) Put(ctx context.Context, key string, data []byte) error
  • func (s *Store) PutIfAbsent(ctx context.Context, key string, data []byte) (bool, error)
  • func (s *Store) PutIfMatch(ctx context.Context, key string, data []byte, etag string) (bool, error)
  • func (s *Store) RetireNamespaceKey(ctx context.Context, name, incarnation string) error
  • func (s *Store) RotateNamespaceKey(ctx context.Context, name string) (*Envelope, error)
  • func (s *Store) SSE() (mode, kmsKeyID string)
  • func (s *Store) SetCMEKRefreshInterval(interval time.Duration)
  • func (s *Store) WithBackend(wrap func(Backend) Backend) *Store
• type Timings
  • func (t *Timings) LogAttrs() []any

Constants

const EncryptionCustomerManaged = "customer-managed"

EncryptionCustomerManaged is the Envelope mode InstallNamespaceKey accepts: the data key is wrapped under a key the customer controls.

const MaxListPage = 1000

Variables

var (
	ErrEndpointDenied = errors.New("store: endpoint denied by allow-list")
)

var ErrNotFound = errors.New("object not found")

var ErrRange = errors.New("invalid object range")

Functions

func Urgent

func Urgent(ctx context.Context) context.Context

Urgent returns ctx marked to bypass request pacing.

func WithTimings

func WithTimings(ctx context.Context, t *Timings) context.Context

WithTimings has the store add the calls made under ctx to t.

Types

type Backend

type Backend = backend

Backend is the backend contract, exported for test wrappers (storetest.Fault).

type Call

type Call int

Call is a kind of work the store does for a caller: the file store's system calls, and the write gate every backend shares.

const (
	CallGate      Call = iota // waiting for a write slot (Config.MaxInflightWrites)
	CallCreate                // creating an object's temp file (and its directory)
	CallWrite                 // write(2) of an object's temp file
	CallWriteBack             // sync_file_range of a chunk (writeBackChunk)
	CallFsync                 // fsync of an object's temp file
	CallLock                  // the key's stripe lock
	CallLink                  // rename(2) or link(2) into place
	CallDirSync               // fsync of a directory after a publish
	CallRead                  // an object read, whole or ranged
	CallList                  // a listing's directory walk
	CallDelete                // unlink(2)

)

type Config

type Config struct {
	Endpoint         string
	Bucket           string
	AllowedEndpoints []string
	SSE              string
	KMSKeyID         string
	// RequestsPerSecond paces every request to the object store; 0 = no
	// pacing. A burst past what the provider serves does not fail, it
	// stalls: at ~900 requests/s against Hetzner Object Storage every
	// in-flight request hung for 13-15 s, the lease renewal among them, and
	// the namespace was fenced. Pacing below that keeps a prefetch burst
	// and the heartbeat both moving.
	RequestsPerSecond float64
	// RequestTimeout bounds one S3 request end to end; 0 = 60 s. A store
	// whose queue is deeper than that truncates large uploads mid-body
	// (MinIO answers 400 IncompleteBody) rather than finishing them.
	RequestTimeout time.Duration
	// MaxInflightWrites bounds the object writes and deletes in flight at
	// once, except those marked Urgent (lease heartbeats, log commits,
	// manifest heads); 0 = 16. A background job publishes thousands of
	// objects in a burst; without a bound they queue ahead of the write
	// path's own small requests on the store.
	MaxInflightWrites int
}

type Envelope

type Envelope struct {
	Mode       string `json:"mode"`
	KeyName    string `json:"key_name,omitempty"`
	KeyVersion string `json:"key_version,omitempty"`
	DEKWrapped []byte `json:"dek_wrapped,omitempty"`
}

Envelope is a namespace's key record: the wrapped data encryption key and the customer key that wraps it. It never holds plaintext key bytes.

type Store

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

Store is one bucket of write-once objects with the OCC primitives a log and a manifest build on (PutIfAbsent to append, PutIfMatch to swap).

sse/kmsKeyID are the Config values the backend was built with, kept here so callers can REPORT the bucket's encryption (namespace metadata) without reaching into a backend. They are descriptive: the backend applies them.

func New

func New(ctx context.Context, endpoint, bucket string) (*Store, error)

New connects to a bucket. endpoint selects the backend: "" = real AWS S3, an http(s) URL = S3-compatible (MinIO), "file:///path" = local filesystem (no dependencies — dev and hermetic CI; unix-only).

func NewConfigured

func NewConfigured(ctx context.Context, cfg Config) (*Store, error)

func (*Store) CheckNamespaceKey

func (s *Store) CheckNamespaceKey(ctx context.Context, name string) error

CheckNamespaceKey revalidates KMS access at the API boundary, including cache-only queries. AWS checks share the lease renewal cadence.

func (*Store) CheckNamespaceKeyCached

func (s *Store) CheckNamespaceKeyCached(ctx context.Context, name string) error

CheckNamespaceKeyCached enforces the key-cache lease for internal calls. Use CheckNamespaceKey for access checks at a request boundary.

func (*Store) ConfigureCMEK

func (s *Store) ConfigureCMEK(provider kms.KeyProvider)

ConfigureCMEK enables transparent encryption of ns/<name>/ objects. A small envelope record outside that prefix bootstraps the encrypted manifest and metadata. The record contains a wrapped DEK, never plaintext key bytes.

func (*Store) Delete

func (s *Store) Delete(ctx context.Context, key string) error

func (*Store) DeleteMany

func (s *Store) DeleteMany(ctx context.Context, keys ...string) error

DeleteMany removes keys. Missing keys are not an error. Empty input is a no-op.

func (*Store) DropBucket

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

DropBucket deletes every object and then the bucket itself (test teardown; the harness's fresh bucket per test must not outlive the test).

func (*Store) EnsureBucket

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

EnsureBucket creates the bucket if missing (dev/test convenience).

func (*Store) Get

func (s *Store) Get(ctx context.Context, key string) ([]byte, error)

func (*Store) GetIfChanged

func (s *Store) GetIfChanged(ctx context.Context, key, etag string) ([]byte, string, bool, error)

GetIfChanged conditionally reads one revision. Empty etag is unconditional. An unchanged object returns nil bytes, its ETag, true, nil. A changed object returns its complete bytes and ETag with false. Absence is ErrNotFound; failures never report unchanged. ETags are opaque backend values.

func (*Store) GetRange

func (s *Store) GetRange(ctx context.Context, key string, offset, length int64) ([]byte, error)

GetRange reads exactly length bytes starting at offset. Short/out-of-bounds ranges fail instead of returning a plausible partial index block.

func (*Store) GetWithETag

func (s *Store) GetWithETag(ctx context.Context, key string) ([]byte, string, error)

GetWithETag returns the object and its ETag for conditional replacement.

func (*Store) InstallNamespaceKey

func (s *Store) InstallNamespaceKey(ctx context.Context, name string, envelope Envelope) error

InstallNamespaceKey publishes the recovery envelope before the first encrypted object. The caller holds the namespace lifecycle lease.

func (*Store) List

func (s *Store) List(ctx context.Context, prefix string) ([]string, error)

List returns all keys under prefix, lexically sorted.

func (*Store) ListPage

func (s *Store) ListPage(ctx context.Context, prefix, after string, limit int) ([]string, string, error)

ListPage returns at most limit keys under prefix, lexically after after. next is the last returned key when another page remains, otherwise empty.

func (*Store) ListPrefixes

func (s *Store) ListPrefixes(ctx context.Context, prefix string) ([]string, error)

ListPrefixes returns the immediate child prefixes under prefix — S3's CommonPrefixes with Delimiter "/", each ending in "/", lexically sorted. Discovering top-level names costs one request per page of names instead of one key per object in the bucket. A prefix appears only if at least one object lives under it, on both backends.

func (*Store) ListPrefixesPage

func (s *Store) ListPrefixesPage(ctx context.Context, prefix, after string, limit int) ([]string, string, error)

func (*Store) Put

func (s *Store) Put(ctx context.Context, key string, data []byte) error

func (*Store) PutIfAbsent

func (s *Store) PutIfAbsent(ctx context.Context, key string, data []byte) (bool, error)

PutIfAbsent writes key only if it does not already exist. Returns false when the key existed. This is the OCC primitive a write-ahead log builds on.

func (*Store) PutIfMatch

func (s *Store) PutIfMatch(ctx context.Context, key string, data []byte, etag string) (bool, error)

PutIfMatch replaces key only if its current ETag equals etag — compare-and- swap, the manifest-swap primitive. (false, nil) = precondition failed: the object changed under us, or no longer exists.

func (*Store) RetireNamespaceKey

func (s *Store) RetireNamespaceKey(ctx context.Context, name, incarnation string) error

RetireNamespaceKey leaves the deleted manifest head as a plaintext name-reuse fence, then removes its wrapped DEK. The lifecycle caller holds the lease. Repeating this after a crash between the CAS and delete is safe.

func (*Store) RotateNamespaceKey

func (s *Store) RotateNamespaceKey(ctx context.Context, name string) (*Envelope, error)

RotateNamespaceKey asks the provider to wrap the existing DEK with its active key version. The encrypted objects and plaintext DEK stay unchanged. It returns the current envelope so an interrupted metadata publication can be repaired on the next write.

func (*Store) SSE

func (s *Store) SSE() (mode, kmsKeyID string)

SSE reports the bucket's configured server-side encryption: the S3 mode ("", "AES256" or "aws:kms") and the KMS key id when one is set. The caller decides what it means for its users; the store does not interpret it.

func (*Store) SetCMEKRefreshInterval

func (s *Store) SetCMEKRefreshInterval(interval time.Duration)

SetCMEKRefreshInterval ties AWS access and rotation probes to the lease heartbeat cadence. Set it once, before use.

func (*Store) WithBackend

func (s *Store) WithBackend(wrap func(Backend) Backend) *Store

WithBackend returns a Store over wrap(s's backend) with s's encryption and write bound: the same bucket seen through a wrapper.

type Timings

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

Timings sums the wall time the store spent in each Call for the calls made under one context (WithTimings), and how many there were: a background job can log its own. Concurrent calls add up, so the sum can pass the span's wall time. The file store fills every Call; S3 only the gate.

func (*Timings) LogAttrs

func (t *Timings) LogAttrs() []any

LogAttrs is t as slog key-value pairs: sys_<call>_ms and sys_<call>_n per Call.