objstore

package module
v0.11.0 Latest Latest
Warning

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

 Go to latest Published: Oct 11, 2026 License: MIT Imports: 14 Imported by: 0

README

objstore

go.dev reference CI

Object storage with compare-and-swap for Go. Supports S3, Google Cloud Storage, Azure Blob Storage, local files and memory, with packages for write-ahead logs, leases and coalesced ranged reads.

Quick start

Requires Go 1.26 or later. This example uses local files on Unix; no cloud account needed. Solaris and AIX are unsupported by the file provider.

  1. In a Go module, install:

    go get github.com/axiomhq/objstore

  2. Save as main.go:

    package main

import (
    "context"
    "fmt"
    "log"

    "github.com/axiomhq/objstore"
    "github.com/axiomhq/objstore/fs"
)

func main() {
    ctx := context.Background()
    s := fs.Open("./data", "example", objstore.Config{})
    defer s.Close()

    if err := s.EnsureBucket(ctx); err != nil {
        log.Fatal(err)
    }
    if err := s.CheckConditionalWrites(ctx); err != nil {
        log.Fatal(err)
    }
    created, err := s.PutIfAbsent(ctx, "greeting", []byte("hello"))
    if err != nil {
        log.Fatal(err)
    }
    fmt.Println("created:", created)
}

  3. Run go run .. It prints created: true on the first run and created: false on later runs. The object stays in ./data/example.

Documentation

License

MIT.

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 lease a single-holder lease on one key, package rangeread coalesced ranged reads. Caching reads is the caller's business.

Index

Constants

View Source 
const MaxListPage = 1000

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

Variables

View Source 
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.

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

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

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

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

View Source 
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 DeleteFailures added in v0.10.0 

func DeleteFailures(keys []string, err error, unattempted bool) error

DeleteFailures reports a common failure for all keys. It returns nil for empty input or a nil error; useful to backends and admission wrappers.

func IsUrgent added in v0.5.0 

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 added in v0.5.0 

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 added in v0.5.0 

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 added in v0.5.0 

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, azure/blob); storetest.Conformance runs against each. Streaming, stat and signing are separate optional capability interfaces.

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 added in v0.5.0 

func (c Call) String() string

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

type ConditionalUploadBackend added in v0.10.0 

type ConditionalUploadBackend interface {
	UploadIfAbsent(ctx context.Context, key string, body io.Reader, opts UploadOptions) (ObjectInfo, bool, error)
	UploadIfMatch(ctx context.Context, key string, body io.Reader, etag string, opts UploadOptions) (ObjectInfo, bool, error)
}

ConditionalUploadBackend is streaming CAS. A lost condition returns zero info, false, nil. An error is not proof of non-commit; see Backend.

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 DeleteError added in v0.10.0 

type DeleteError struct {
	Failures []DeleteFailure
}

DeleteError reports every failed or unattempted input, preserving input order (including duplicates). All underlying errors participate in Is/As.

func (*DeleteError) Error added in v0.10.0 

func (e *DeleteError) Error() string

func (*DeleteError) Unwrap added in v0.10.0 

func (e *DeleteError) Unwrap() []error

type DeleteFailure added in v0.10.0 

type DeleteFailure struct {
	Key         string
	Err         error
	Unattempted bool
}

DeleteFailure names one input key whose deletion was not confirmed. Unattempted means no request for that input was sent; otherwise its outcome may be ambiguous. Missing objects count as successfully deleted.

type KMSKeyFunc added in v0.5.0 

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 ObjectInfo added in v0.10.0 

type ObjectInfo struct {
	Size         int64
	ETag         string
	Metadata     map[string]string
	ContentType  string
	CacheControl string
	StorageClass string
	LastModified time.Time
}

ObjectInfo describes one revision. ETag is the backend's opaque CAS token, interchangeable with GetWithETag, not a portable checksum. Upload results always contain Size and ETag from the commit; other fields may be absent.

type ReaderBackend added in v0.10.0 

type ReaderBackend interface {
	NewReader(ctx context.Context, key string, offset, length int64) (io.ReadCloser, error)
}

ReaderBackend is the optional streaming read capability. The caller closes the reader. length 0 means to EOF; positive lengths are capped at EOF. At or beyond EOF an existing object returns an empty reader. Negative or overflowing ranges wrap ErrRange. This does not change exact GetRange.

type SignBackend added in v0.10.0 

type SignBackend interface {
	Sign(ctx context.Context, key string, opts SignOptions) (SignedRequest, error)
}

SignBackend is the optional signed request capability.

type SignOptions added in v0.10.0 

type SignOptions struct {
	Method  string
	Expires time.Duration
	Headers http.Header
}

SignOptions selects an HTTP method, positive lifetime and required headers. Signing support and allowed headers are provider-specific; unsupported methods or headers wrap errors.ErrUnsupported. It never uploads a body.

type SignedRequest added in v0.10.0 

type SignedRequest struct {
	URL     string
	Method  string
	Headers http.Header
}

SignedRequest must be used with its Method and all its Headers unchanged. URL contains credentials: do not log it. Signing bypasses the write gate because it performs no object mutation.

type StatBackend added in v0.10.0 

type StatBackend interface {
	Stat(ctx context.Context, key string) (ObjectInfo, error)
}

StatBackend is the optional object metadata capability.

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 added in v0.5.0 

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 added in v0.5.0 

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) Close added in v0.8.0 

func (s *Store) Close() error

Close releases the provider's client, if its backend has one to release (gcs.Backend does); otherwise it does nothing. Stores from WithBackend share the backend: close one of them, once, when all are done.

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) ID added in v0.6.0 

func (s *Store) ID() string

ID names the bucket: Stores over the same bucket, opened apart or seen through WithBackend, share it. A backend without an ID method gets a random one per Open.

func (*Store) KMS added in v0.5.0 

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) NewReader added in v0.10.0 

func (s *Store) NewReader(ctx context.Context, key string, offset, length int64) (io.ReadCloser, error)

NewReader streams an EOF-tolerant range; it never falls back to Get.

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) Sign added in v0.10.0 

func (s *Store) Sign(ctx context.Context, key string, opts SignOptions) (SignedRequest, error)

Sign produces a signed request, or errors.ErrUnsupported; it does not write.

func (*Store) Stat added in v0.10.0 

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

Stat returns information for the current revision without downloading it on cloud providers. The file backend hashes the file to obtain its CAS token.

func (*Store) Upload added in v0.10.0 

func (s *Store) Upload(ctx context.Context, key string, body io.Reader, opts UploadOptions) (ObjectInfo, error)

Upload writes a stream unconditionally. A provider may use bounded multipart or resumable uploads. The returned token belongs to this commit, not a later HEAD. The write slot is held until the complete upload returns.

func (*Store) UploadIfAbsent added in v0.10.0 

func (s *Store) UploadIfAbsent(ctx context.Context, key string, body io.Reader, opts UploadOptions) (ObjectInfo, bool, error)

UploadIfAbsent creates an object only when no live object exists.

func (*Store) UploadIfMatch added in v0.10.0 

func (s *Store) UploadIfMatch(ctx context.Context, key string, body io.Reader, etag string, opts UploadOptions) (ObjectInfo, bool, error)

UploadIfMatch replaces an object only if its opaque CAS token matches etag.

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 added in v0.5.0 

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 added in v0.5.0 

func TimingsOf(ctx context.Context) *Timings

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

func (*Timings) Attrs added in v0.5.0 

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 added in v0.5.0 

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

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

type UploadBackend added in v0.10.0 

type UploadBackend interface {
	Upload(ctx context.Context, key string, body io.Reader, opts UploadOptions) (ObjectInfo, error)
}

UploadBackend is the optional unconditional streaming upload capability.

type UploadOptions added in v0.10.0 

type UploadOptions struct {
	Size         *int64
	Metadata     map[string]string
	ContentType  string
	CacheControl string
	StorageClass string
}

UploadOptions controls a streaming write. Size nil means read to EOF; a non-nil Size means consume exactly that many bytes (zero is an empty object). Surplus bytes are left unread. A short body fails, but any error may have an ambiguous commit outcome. The caller owns body and must unblock a blocked Read on cancellation; Upload never closes it. Options unsupported by a backend are rejected before writing, never silently ignored.

Source Files

View all Source files

Directories

Path Synopsis
aws
s3
Package s3 is the S3 backend for objstore.
 Package s3 is the S3 backend for objstore.
azure
blob
Package blob implements objstore on Azure block blobs.
 Package blob implements objstore on Azure block blobs.
Package fs is the file backend for objstore: one bucket as a directory of write-once files.
 Package fs is the file backend for objstore: one bucket as a directory of write-once files.
gcp
gcs
Package gcs is the Google Cloud Storage backend for objstore.
 Package gcs is the Google Cloud Storage backend for objstore.
internal
stream
Package stream holds body validation shared by provider implementations.
 Package stream holds body validation shared by provider implementations.
Package lease is a single-holder lease on one object in a Store: leader election, or a single writer, on S3 or any store with conditional PUT.
 Package lease is a single-holder lease on one object in a Store: leader election, or a single writer, on S3 or any store with conditional PUT.
Package mem is a Backend in process memory: for tests, and for a process whose data need not outlive it.
 Package mem is a Backend in process memory: for tests, and for a process whose data need not outlive it.
Package rangeread reads byte ranges of immutable objects in few GETs: Plan unions and coalesces extents under gap, extra-byte and range-size limits, and Read runs the plan on bounded workers and returns each extent's bytes.
 Package rangeread reads byte ranges of immutable objects in few GETs: Plan unions and coalesces extents under gap, extra-byte and range-size limits, and Read runs the plan on bounded workers and returns each extent's bytes.
Package storetest provides provider-free test helpers: a fault-injecting, metering wrapper around a Store, and Conformance, the suite every backend passes.
 Package storetest provides provider-free test helpers: a fault-injecting, metering wrapper around a Store, and Conformance, the suite every backend passes.
bucket
Package bucket opens a fresh test bucket on the file backend or, with OBJSTORE_TEST_S3, on S3.
 Package bucket opens a fresh test bucket on the file backend or, with OBJSTORE_TEST_S3, on S3.
Package wal is a write-ahead log in object storage: ordered, immutable pages sequenced by conditional PUT.
 Package wal is a write-ahead log in object storage: ordered, immutable pages sequenced by conditional PUT.

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.