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)

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.

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).

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.