objstore

README

objstore

go get github.com/axiomhq/objstore

Object storage with compare-and-swap, for Go. One Store over an S3 bucket (objstore/aws/s3), Google Cloud Storage (objstore/gcp/gcs), or a durable local directory (objstore/fs). Packages are laid out by provider: aws/ holds s3, gcp/ holds gcs. Cloudflare R2 (at https://<account-id>.r2.cloudflarestorage.com with AWS_REGION=auto), MinIO, Ceph and Hetzner use aws/s3 because they speak the S3 API. With:

• conditional PUT and GET, and the ETag with the body
• paginated list, delimited list, batch delete
• one package per provider, so a binary links only the SDKs it uses
• a file store (objstore/fs) that fsyncs every object and writes large ones back in chunks, so a small urgent write never waits behind a big one
• a KMS key per object, applied by the store (S3 SSE-KMS, GCS kmsKeyName), picked by a function of the key
• a startup probe that the store honours conditional writes
• a conformance suite, 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

Imports: github.com/axiomhq/objstore/aws/s3, github.com/axiomhq/objstore/gcp/gcs, github.com/axiomhq/objstore/fs.

1. Open a store: s, err := s3.Open(ctx, s3.Config{Endpoint: "https://s3.us-east-1.amazonaws.com", Bucket: "my-bucket"}, objstore.Config{}). GCS: s, err := gcs.Open(ctx, gcs.Config{Bucket: "my-bucket"}, objstore.Config{}). Local: s := fs.Open("/var/lib/data", "bucket", objstore.Config{}) (Unix only). Any other objstore.Backend: s := objstore.Open(b, objstore.Config{}). s3.Config adds SSE, an endpoint allow-list and a request timeout; objstore.Config adds pacing (RequestsPerSecond), and a write bound (MaxInflightWrites).
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. objstore.ErrConflict (S3's 409) means the store never decided: retry.
5. Poll: body, etag, unchanged, err := s.GetIfChanged(ctx, key, etag). One conditional request on S3 and GCS; the file store reads the whole file either way, so it is not cheaper than Get there.

A missing key returns an error wrapping objstore.ErrNotFound. Mark a context with objstore.Urgent(ctx) for heartbeats and log commits: its calls skip pacing and the write bound, and on the file store are written whole instead of in chunks.

Migrating from v0.4

v0.4	v0.5
S3 in package objstore	package s3 at objstore/aws/s3; GCS (new) is objstore/gcp/gcs
objstore.New(ctx, endpoint, bucket)	s3.Open(ctx, s3.Config{Endpoint: endpoint, Bucket: bucket}, objstore.Config{}); for file://root, fs.Open(root, bucket, objstore.Config{})
Config.Endpoint, Bucket, SSE, KMSKeyID, AllowedEndpoints, RequestTimeout	s3.Config
Store.SSE(), objstore.ErrEndpointDenied	(*s3.Backend).SSE() (from s3.New), s3.ErrEndpointDenied
s.ConfigureCMEK(p), InstallNamespaceKey, RotateNamespaceKey, CheckNamespaceKey, package kms	gone: s = s.WithKMSKeys(fn), and the store encrypts (see Encryption)
Timings.LogAttrs()	Timings.Attrs()
l.Release(), ref.Release()	l.Release(ctx), ref.Release(ctx)
l.Log = logger	l.SetLogger(logger)
l, err = l.Acquire(ctx)	err = l.Acquire(ctx)
ByteCache.Generation.Load(), Resident.Generation.Load()	GenerationOf(key) (per namespace)
r := rangeread.New(...); r.Cfg	r, err := rangeread.New(...); r.Config()
storetest.New(t), storetest.NewFaulty(t)	bucket.New(t), bucket.NewFaulty(t) (objstore/storetest/bucket)

Objects v0.4 sealed client-side stay ciphertext: v0.5 does not decrypt them. See CHANGELOG.md for the rest.

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 exactly 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 1 to 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; never needs the namespace key
DeleteMany(ctx, keys...)	batch delete
EnsureBucket(ctx), DropBucket(ctx)	create or remove the bucket
CheckConditionalWrites(ctx)	prove the store honours If-None-Match (3 requests under _probe/); call once before writing

ctx = objstore.WithTimings(ctx, &t) has the store add the wall time and count of every call made under ctx to t, per objstore.Call; logger.LogAttrs(ctx, slog.LevelInfo, "job", t.Attrs()...) logs them. The file store fills every Call; S3 and GCS fill only the write gate.

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. If ctx ends first the outcome is unknown; w.Enqueue returns a receipt that always gets the verdict.
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. wal.WalkParallel fetches on workers goroutines; wal.WalkWithGet and wal.WalkParallelWithGet read through your own fetch (a cache, say).
5. Stop: w.Close() drains, then returns.

error	means
ErrOverloaded	128 MiB is already waiting for the store; retry later
ErrRecordTooLarge	a record no page can hold, or an append bigger than the queue; refused at once
ErrInvalidRecord	negative Size, or AppendTo failed or disagreed with Size; only that Append fails
ErrLostRace	another writer took the sequence (split brain). Finishes the writer
ErrUnresolved	the outcome is unknown; the records may be durable. Reopen from your checkpoint and the log. From a commit (covered floor, corrupt page) it finishes the writer
ErrWriterFailed	the writer already finished (Stats().Terminal says why); nothing written; reopen from your checkpoint
ErrCorrupt	a page failed its checksum or framing, or a page is missing inside a bounded walk

A finished writer never writes again: w.Stats().Terminal says why, and every later Append gets ErrWriterFailed. Concurrent Appends share one entry, at most one per WithCommitInterval (default 1 s, clamped to 1 ms to 5 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. One live writer per prefix, enforced by the caller (a lease); 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: r, err := rangeread.New(s, c, rangeread.Config{}) (zero fields take defaults, invalid ones are an error; r.Config() shows the result), then scope, err := r.FetchRanges(ctx, loads) and cache.Scoped(scope, key) per load.

piece	does
cache.ByteCache	striped LRU under one byte budget, each entry charged len(key) + cache.EntryOverhead extra; decoded values ride on their bytes (PutDecoded, charged via cache.Sizer; Recharge re-sizes one you still hold)
ByteCache.GenerationOf(key)	the key's namespace generation: read it before loading and pass it to Put, so an invalidation in between drops the fill
cache.Disk	disposable disk tier, 4 KiB block checksums, pins, inactivity expiry; Disk.Logger (nil = slog.Default())
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; after InvalidateNamespace, Unpin or re-Pin that namespace's disk pins. A loader panic comes back as an error, its stack to Cache.Logger. NewDisk locks its directory and removes stale ones a crashed process left under dir.

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. A long-lived caller that retries keeps one handle (l := lease.New(...), then err := l.Acquire(ctx) in its loop): the handle remembers its unresolved writes and adopts one that landed late, where a fresh handle waits it out (1.5 TTL).
2. Say what losing it costs: l.Start(func() { cancelWork() }). The lease renews itself every TTL/4 from Acquire on. l.SetLogger(logger) logs renewal failures and fences.
3. Before every guarded action: if err := l.Valid(); err != nil { stop }.
4. Hand it over when done: l.Release(ctx). The next process takes it at once instead of waiting out the TTL. If ctx ends first, the lease stops locally and the handover is skipped.

event	what happens
renewal CAS loses to another owner	ErrNotOwner, the lease fences: your Start callback runs once on its own goroutine (Done() waits for it), 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.

• The fence callback may call Release or Retire; neither runs it or waits for it. It must not wait on Done().
• A fence before Start is owed to it: Start runs the callback on the caller's goroutine, and Done() stays open until it returns (or until Release/Retire gives it up). A fence after Release or Retire runs no callback.
• lease.Shared shares one lease among several holders in a process by reference count; the last Ref.Release(ctx) hands it back.
• l.CheckHeadOnRenewal(fn) runs fn after each renewal. It fences only when fn's error wraps ErrNotOwner; any other error is logged and retried, never a fence, since the renewal already extended the lease.
• 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. Say which key each object gets: s = s.WithKMSKeys(func(ctx context.Context, key string) (string, error) { ... }). Return "" for the bucket's own policy.
2. Write as usual. S3 stores each keyed object with x-amz-server-side-encryption: aws:kms and that key id; GCS with that kmsKeyName.
3. Read as usual. The store decrypts with the key the object was written under; nothing to pass.
4. For one write only: ctx = objstore.WithKMSKey(ctx, id). On a store with a WithKMSKeys function, the function decides; it reads objstore.KMSKey(ctx) to honour it.

case	what happens
key disabled, deleted or grant removed	reads and writes of its objects fail with an error wrapping objstore.ErrAccessDenied
key rotated in KMS	nothing to do; the store uses the key's current version
s.KMS()	true on S3 and GCS; false on fs, which ignores keys and encrypts nothing
tests	s, k := storetest.NewKMS(s) records each object's key (k.KeyOf) and revokes one (k.Revoke, k.Restore)

s3.Config.SSE and KMSKeyID still set one policy for the whole bucket.

Tests with faults

1. s := bucket.New(t) (objstore/storetest/bucket) gives a fresh bucket per test, dropped at cleanup.
2. s, f := bucket.NewFaulty(t) wraps it in a *storetest.Fault; storetest.Faulty(t, s) wraps a Store you opened yourself.
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 the call's context ends (use t.Context()), 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 -skip '/ETagCASMissing' ./...   # 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. storetest.Conformance(t, s) is the suite every backend passes; run it on your own objstore.Backend via objstore.Open. storetest imports no provider, so that links neither the AWS nor the GCS SDK; the per-test buckets live in storetest/bucket. MinIO creates the object on a PutIfMatch of a missing key, hence the skip.

CI runs gofmt, go mod tidy -diff, vet, staticcheck and the race tests on Linux; the same tests against MinIO; the tests on macOS; and go vet cross-compiles for windows, darwin, illumos, solaris and aix/ppc64.

Real providers, each skipped when its variable is unset:

• OBJSTORE_TEST_S3: S3 endpoint, e.g. MinIO at http://localhost:9000 (each of AWS_ACCESS_KEY_ID, AWS_SECRET_ACCESS_KEY, AWS_REGION left unset defaults to MinIO's minioadmin, minioadmin, us-east-1).
• OBJSTORE_TEST_R2_ENDPOINT: https://<account-id>.r2.cloudflarestorage.com, with an R2 API token in AWS_ACCESS_KEY_ID / AWS_SECRET_ACCESS_KEY and AWS_REGION=auto. Runs TestR2 in ./aws/s3.
• OBJSTORE_TEST_GCS_PROJECT: a GCP project where Application Default Credentials can create buckets. Runs TestConformanceReal in ./gcp/gcs against a fresh objstore-test-<nanos> bucket, dropped afterwards. TestConformance runs on an in-process fake-gcs-server with no variable.

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. Each provider is its own package, so a binary links only the SDKs it uses: package aws/s3 (AWS S3, and R2, MinIO, Ceph and Hetzner, which speak the S3 API), package gcp/gcs (Google Cloud Storage) and package fs, a durable local store with chunked write-back. Each builds a Backend and hands it to Open.

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.

Store.WithKMSKeys names a KMS key per object; S3 (SSE-KMS) and GCS (kmsKeyName) encrypt it under that key.

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 IsUrgent(ctx context.Context) bool
• func KMSKey(ctx context.Context) string
• func OpErr(op, key string, err error) error
• func Urgent(ctx context.Context) context.Context
• func WithKMSKey(ctx context.Context, id string) context.Context
• func WithTimings(ctx context.Context, t *Timings) context.Context
• type Backend
• type Call
  • func (c Call) String() string
• type Config
• type KMSKeyFunc
• type Store
  • func Open(b Backend, cfg Config) *Store
  • func (s *Store) CheckConditionalWrites(ctx context.Context) error
  • 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) KMS() bool
  • 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) WithBackend(wrap func(Backend) Backend) *Store
  • func (s *Store) WithKMSKeys(fn KMSKeyFunc) *Store
• type Timings
  • func TimingsOf(ctx context.Context) *Timings
  • func (t *Timings) Attrs() []slog.Attr
  • func (t *Timings) Since(op Call, start time.Time)

Constants

const MaxListPage = 1000

MaxListPage is the largest limit ListPage and ListPrefixesPage accept, S3's own page size.

Variables

var ErrAccessDenied = errors.New("store: access denied")

ErrAccessDenied is wrapped by every error for a request the store refused (403), or for an object whose KMS key is disabled, deleted or out of reach.

var ErrConflict = errors.New("store: conditional write conflict, retry")

ErrConflict is a conditional write whose outcome the backend never decided: another conditional write on the same key was in flight (S3's 409 ConditionalRequestConflict). The object may or may not have been written; the caller retries. It is never a lost race, which is (false, nil).

var ErrInvalidKey = errors.New("store: invalid key")

ErrInvalidKey is wrapped by every error for a key a backend refuses to store (the file backend rejects empty elements, "." and "..", and its own reserved file names).

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

ErrNotFound is wrapped by every error for a missing object or bucket.

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

ErrRange is wrapped by every error for a range outside the object, or an offset or length that is not a valid range at all.

Functions

func IsUrgent

func IsUrgent(ctx context.Context) bool

IsUrgent reports whether ctx was marked Urgent. Backends consult it for their own fast paths (the file store writes an urgent object whole).

func KMSKey

func KMSKey(ctx context.Context) string

KMSKey is the KMS key id a write under ctx carries, "" for none. Backends read it; so do backend wrappers (storetest.KMS).

func OpErr

func OpErr(op, key string, err error) error

OpErr attaches the operation and the key to every error leaving a backend. A bare SDK error names neither, so a failure surfaced three layers up — mid-compaction, mid-replay, inside a query's fan-out — says only that storage was unhappy about something. %w keeps ErrNotFound unwrappable by errors.Is. nil stays nil.

func Urgent

func Urgent(ctx context.Context) context.Context

Urgent returns ctx marked as latency-critical (a lease heartbeat, a log commit, a manifest swap). Requests under it:

• skip request pacing (Config.RequestsPerSecond);
• skip the write bound (Config.MaxInflightWrites);
• on the file backend, are written whole instead of written back in chunks, and so never wait for their own write-back.

func WithKMSKey

func WithKMSKey(ctx context.Context, id string) context.Context

WithKMSKey returns ctx whose writes are encrypted server-side with KMS key id: S3 SSE-KMS with that key id, GCS kmsKeyName. On a Store with a KMSKeyFunc (WithKMSKeys) the function decides; it can read id with KMSKey.

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 interface {
	Put(ctx context.Context, key string, data []byte) error
	PutIfAbsent(ctx context.Context, key string, data []byte) (bool, error)
	Get(ctx context.Context, key string) ([]byte, error)
	GetRange(ctx context.Context, key string, offset, length int64) ([]byte, error)
	GetWithETag(ctx context.Context, key string) ([]byte, string, error)
	GetIfChanged(ctx context.Context, key, etag string) ([]byte, string, bool, error)
	PutIfMatch(ctx context.Context, key string, data []byte, etag string) (bool, error)
	ListPage(ctx context.Context, prefix, after string, limit int) ([]string, string, error)
	ListPrefixesPage(ctx context.Context, prefix, after string, limit int) ([]string, string, error)
	Delete(ctx context.Context, key string) error
	DeleteMany(ctx context.Context, keys ...string) error
	EnsureBucket(ctx context.Context) error
	DropBucket(ctx context.Context) error
}

Backend is the object-store contract every provider package satisfies (fs, s3, gcs); storetest.Conformance runs against each. It may grow.

Every error a backend returns is wrapped by OpErr. A failed condition is (false, nil), a missing object wraps ErrNotFound, a bad range wraps ErrRange. The ok of a conditional write is meaningful only when err == nil, except that a backend may report (true, err) when the object is visible but not proven durable (the file backend, when the directory sync after the publish fails); a retry of the same write settles it. ETags are opaque. Listing is lexical, after is exclusive, next is the last key returned when more remain and "" otherwise. The file backend reserves terminal .lock and .tmp-* files; dot-prefixed directory elements remain valid so documented namespace names work.

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)

)

func (Call) String

func (c Call) String() string

String is the Call's short name, as used in Attrs ("gate", "fsync", ...).

type Config

type Config struct {
	// 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
	// MaxInflightWrites bounds the object writes and deletes in flight at
	// once, except those marked Urgent (lease heartbeats, log commits,
	// manifest heads); 0 = 16, negative = unbounded. 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
}

Config is what Open adds around any backend. Provider-specific settings (endpoint, credentials, bucket-wide server-side encryption) live in each provider package's own Config.

type KMSKeyFunc

type KMSKeyFunc func(ctx context.Context, key string) (string, error)

KMSKeyFunc names the KMS key an object is written with; "" writes it under the bucket's own encryption policy. It runs on every Put, PutIfAbsent and PutIfMatch, before the write takes a slot.

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

func Open

func Open(b Backend, cfg Config) *Store

Open returns a Store over b with cfg's pacing, write bound and KMS keys. The provider packages (fs, s3, gcs) each have an Open that builds the backend and calls this.

func (*Store) CheckConditionalWrites

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

CheckConditionalWrites proves the backend honours If-None-Match: a first PutIfAbsent of a fresh _probe/ifnonematch/ key must create it and a second must not. A store that ignores the header would let two log writers both win one sequence. Three requests; call it once before writing.

func (*Store) Delete

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

Delete removes key. A missing key is not an 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). A bucket that is already gone is not an error.

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)

Get returns the whole object at key; a missing object wraps ErrNotFound.

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

func (s *Store) KMS() bool

KMS reports whether the backend applies per-object KMS keys (WithKMSKeys). The file backend does not: it ignores them and encrypts nothing.

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 (1 to MaxListPage) 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 every backend.

func (*Store) ListPrefixesPage

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

ListPrefixesPage is one page of ListPrefixes: at most limit (1 to MaxListPage) prefixes lexically after after, and the cursor as in ListPage.

func (*Store) Put

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

Put writes key unconditionally, replacing any object there.

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. See Backend for the one case where ok is true alongside an error.

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. See Backend for the one case where ok is true alongside an error.

func (*Store) WithBackend

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

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

func (*Store) WithKMSKeys

func (s *Store) WithKMSKeys(fn KMSKeyFunc) *Store

WithKMSKeys returns s with every write asking fn for its object's KMS key. fn decides: its answer replaces any WithKMSKey on the write's context, which fn can read with KMSKey. The store encrypts and decrypts: reads need no key, and an object whose key is disabled or revoked fails with ErrAccessDenied. The file backend ignores keys (KMS reports false).

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 TimingsOf

func TimingsOf(ctx context.Context) *Timings

TimingsOf returns the Timings ctx carries, or nil. For backends.

func (*Timings) Attrs

func (t *Timings) Attrs() []slog.Attr

Attrs is t as slog attributes: sys_<call>_ms and sys_<call>_n per Call. A nil t has none.

func (*Timings) Since

func (t *Timings) Since(op Call, start time.Time)

Since adds the time from start to op; a nil t records nothing. For backends.