objectstore

package module
v0.1.0 Latest Latest
Warning

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

 Go to latest Published: Jul 22, 2026 License: Apache-2.0 Imports: 29 Imported by: 0

README

Objectstore

Objectstore is a small Go abstraction over key-addressed object storage. It provides one Bucket interface with filesystem and S3-compatible implementations, plus wrappers for key prefixes, metering, resilient reads, and concurrent ranged downloads.

The interface deliberately follows object-store semantics rather than filesystem semantics. Writes support atomic create and compare-and-swap preconditions, reads return opaque entity tags, listings are ordered and paginated, and callers retain responsibility for closing response bodies.

bucket, err := objectstore.Open(ctx, "s3://example/data?region=us-west-2")
if err != nil {
	return err
}

etag, err := bucket.Put(
	ctx,
	"state/current",
	strings.NewReader("value"),
	int64(len("value")),
	objectstore.IfAbsent(),
)

See Storage semantics for the contract implemented by every backend.

Backends

  • OpenFS stores objects beneath a local filesystem directory. It is useful for development, tests, and single-host deployments.
  • OpenS3 supports Amazon S3 and services implementing the S3 API.
  • Open constructs either backend from a file:// or s3:// URL.

Objectstore is pre-1.0. Public APIs and operational behavior may change between minor releases.

Development

Run the standalone checks with:

go test -timeout=2s ./...
go vet ./...

Objectstore is licensed under the Apache License 2.0.

Documentation

Overview

Package objectstore provides a small object-storage abstraction with S3 and filesystem implementations. Bucket mirrors common object-store semantics so implementations remain thin and callers can use conditional writes, ranged reads, and ordered listings without depending on a provider SDK.

Keys are byte-equivalent across calls; List returns lex-sorted keys; range reads use byte offsets; conditional writes use opaque ETags. Anything S3-specific that doesn't generalize (SSE, storage classes, lifecycle) lives behind constructor options on the S3 impl, not on the interface.

Errors: every method returns ErrNotFound for a missing key and ErrPreconditionFailed for a conditional write whose precondition did not hold. Other errors are wrapped with operation context.

Index

Constants

This section is empty.

Variables

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

ErrNotFound is returned by Get, GetRange, and Delete when the key does not exist. Callers can use errors.Is.

View Source 
var ErrPreconditionFailed = errors.New("objectstore: precondition failed")

ErrPreconditionFailed is returned by Put / PutStream when the supplied ifMatch precondition did not hold. Callers can use errors.Is.

Functions

func IfAbsent 

func IfAbsent() *string

IfAbsent is the ifMatch sentinel meaning "succeed iff the key does not currently exist."

func IsConsistentRead 

func IsConsistentRead(ctx context.Context) bool

IsConsistentRead is the exported form of consistentReadFromContext, for backends and test doubles outside this package that need to honor (or assert) the WithConsistentRead hint.

func WithCacheControl 

func WithCacheControl(ctx context.Context, cacheControl string) context.Context

WithCacheControl returns a context that requests the given Cache-Control value on objects written via Put or PutStream while it is in scope. An empty value is a no-op. Cache-aware backends persist it with the object; backends without a response cache ignore it.

func WithConsistentRead 

func WithConsistentRead(ctx context.Context) context.Context

WithConsistentRead returns a context that requests a strongly-consistent read on objects fetched via Get while it is in scope. On Tigris Global/Dual-region buckets an ordinary Get is served from the REGIONAL replica, which can lag the global leader — a reader in a trailing region then observes a stale object. For coordination objects that feed a compare-and-swap decision this is not merely slow but wrong: the stale read drives a doomed CAS that fails against the leader's current state. X-Tigris-Consistent routes the read to the global leader, pairing the read with the consistent CAS write (see conditionalOpts) so the whole read-modify-write is linearizable. Backends without regional replication (FS) ignore it. Carried on the context so the hint adds no parameter to the Bucket interface and passes transparently through the Prefixed/Metered wrappers.

Types

type Bucket 

type Bucket interface {
	// Put writes body of length bytes to key. ifMatch sets the
	// precondition (see package doc). Returns the new ETag.
	Put(ctx context.Context, key string, body io.Reader, length int64, ifMatch *string) (etag string, err error)

	// PutStream is Put for unknown-length bodies. The body is fully
	// consumed before the precondition is evaluated.
	PutStream(ctx context.Context, key string, body io.Reader, ifMatch *string) (etag string, err error)

	// Get returns the object's body and current ETag. The caller must
	// Close the reader. Returns ErrNotFound if the key does not exist.
	Get(ctx context.Context, key string) (body io.ReadCloser, etag string, err error)

	// GetRange returns length bytes starting at offset off. length == 0
	// means "to end of object." A negative off addresses from the end
	// (e.g., off=-4096, length=0 fetches the last 4 KiB). Returns
	// ErrNotFound if the key does not exist.
	GetRange(ctx context.Context, key string, off, length int64) (body io.ReadCloser, err error)

	// Stat returns object metadata (size, etag, last-modified) without
	// fetching the body. S3-backed impls use HeadObject; FS uses
	// os.Stat + a deferred-content-hash etag. Returns ErrNotFound if
	// the key does not exist.
	Stat(ctx context.Context, key string) (ObjectInfo, error)

	// List returns up to a backend-defined maximum (typically 1000)
	// objects whose keys begin with prefix and are lexicographically
	// greater than startAfter, sorted ascending.
	List(ctx context.Context, prefix, startAfter string) ([]ObjectInfo, error)

	// Delete removes the object at key. Returns ErrNotFound if absent.
	Delete(ctx context.Context, key string) error
}

Bucket is the object-storage abstraction. Every implementation guarantees:

  • Put with ifMatch is atomic and linearizable. nil ifMatch overwrites; ifMatch=&"" requires the key not exist (returns ErrPreconditionFailed if it does); ifMatch=&etag requires the current ETag equals etag.
  • PutStream is identical to Put but accepts a body of unknown length; bodies are always fully consumed before preconditions are evaluated, so streaming producers (e.g. an io.Pipe-fed compressor) cannot deadlock on early-return.
  • WithCacheControl scopes the stored response-cache policy for Put and PutStream without changing this interface. Cache-aware backends persist it with the object; other backends ignore it.
  • Get and GetRange return io.ReadCloser bodies that the caller must Close. Closing before EOF is permitted.
  • List returns keys in lexicographic order. startAfter is exclusive: results have Key > startAfter. Pages are bounded; iterate with startAfter = lastKey to enumerate.
  • Delete on a missing key returns ErrNotFound; idempotent callers should ignore that.

func NewHealing 

func NewHealing(b Bucket, opts HealOpts) Bucket

NewHealing wraps b so Get and GetRange bodies transparently survive a path-pinned or broken connection: a body that stalls, drops below the throughput floor, or errors mid-stream is closed (killing its TCP connection — Go does not re-pool an undrained conn) and the remaining bytes are re-requested with a fresh Range read, which lands on a different striped transport (clientSet) and therefore a different ECMP hash. This converts a bad path hash from hours to seconds. The final allowed connection is kept while it makes forward progress, however slowly, so the throughput floor cannot make a read impossible.

Objects are assumed immutable for the duration of a read (objectstore keys are content-addressed in practice); as a guard, the object's ETag is captured on the first heal and any later change aborts the read. Callers' integrity checks (sha verify) remain the backstop.

All other Bucket methods delegate unchanged.

func Open 

func Open(ctx context.Context, raw string) (Bucket, error)

Open constructs a Bucket from a URL string. Recognized schemes:

  • file:///abs/path An FS rooted at the given absolute path. No query options.

  • s3://bucket/prefix?region=…&endpoint=…&path-style=true|false An S3 over the named bucket. The path component (after the bucket) is the Prefix. Query parameters override defaults: region — bucket region; falls back to AWS_REGION / AWS_DEFAULT_REGION env vars when unset. endpoint — base URL for S3-compatible services (R2, MinIO, Tigris, GCS S3 API). Implies path-style=true unless overridden. path-style — force path-style addressing (true|false). ip-version — 4 to dial the endpoint over IPv4 only (some anycast endpoints route IPv6 to a far PoP). clients — striped transport count (distinct TCP conns to spread ECMP path risk); 0/unset = default (4). heal — false disables self-healing reads (watchdog + resume-on-fresh-connection); default true. Credentials come from the SDK default chain. The URL must not carry credentials.

func Prefixed 

func Prefixed(inner Bucket, prefix string) Bucket

Prefixed returns a Bucket view that prepends prefix to every key. A trailing "/" on prefix is normalized; an empty prefix returns the inner bucket unchanged.

Useful for topic-scoping a shared bucket across multiple workloads: `objectstore.Prefixed(bucket, "images/")` and `objectstore.Prefixed(bucket, "documents/")` give two isolated views over one underlying connection.

type FS 

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

FS stores objects under a root directory. Keys map directly onto paths: "a/b/c.bin" lives at <root>/a/b/c.bin. Conditional writes use atomic Link or Rename through per-key temp files. ETags are sha256 of the body.

Intended for tests, single-host development, and as a minimum- viable backend without object storage. Cross-process correctness relies on filesystem rename-into-existing-target atomicity, which POSIX guarantees.

func OpenFS 

func OpenFS(root string) (*FS, error)

OpenFS opens (or creates) an FS rooted at root. The root is created with 0o755 if it does not exist.

func (*FS) Delete 

func (f *FS) Delete(ctx context.Context, key string) error

Delete removes key. Returns ErrNotFound if missing.

func (*FS) Get 

func (f *FS) Get(ctx context.Context, key string) (io.ReadCloser, string, error)

Get returns the object's body and ETag.

func (*FS) GetRange 

func (f *FS) GetRange(ctx context.Context, key string, off, length int64) (io.ReadCloser, error)

GetRange returns length bytes starting at off.

func (*FS) List 

func (f *FS) List(ctx context.Context, prefix, startAfter string) ([]ObjectInfo, error)

List returns objects with key beginning with prefix and key > startAfter.

func (*FS) Put 

func (f *FS) Put(ctx context.Context, key string, body io.Reader, length int64, ifMatch *string) (string, error)

Put writes body to key with precondition ifMatch. See Bucket.Put.

func (*FS) PutStream 

func (f *FS) PutStream(ctx context.Context, key string, body io.Reader, ifMatch *string) (string, error)

PutStream writes a body of unknown length. See Bucket.PutStream.

func (*FS) Stat 

func (f *FS) Stat(ctx context.Context, key string) (ObjectInfo, error)

Stat returns object metadata without opening the body. The etag is computed from the file's contents (sha256 of bytes), matching the etag returned by Get / Put.

type FetchOpts 

type FetchOpts struct {
	Threshold   int64 // objects smaller than this use one GET; default 16MB
	PartSize    int64 // ranged part size; default 16MB
	Concurrency int   // concurrent part fetches; default 6
}

FetchOpts configures FetchRangedAt / FetchReader. Zero values take defaults.

type HealOpts 

type HealOpts struct {
	FloorBps int64         // rotate a non-final body below this rate; default 100 KiB/s
	Window   time.Duration // how long below floor (or stalled on final) before killing; default 10s
	Attempts int           // max connections per read (first + heals); default 6
	Backoff  time.Duration // pause before each re-issue; default 100ms
}

HealOpts configures self-healing reads. Zero values take defaults.

type LabelStats 

type LabelStats struct {
	PutBytes        uint64 `json:"put_bytes"`
	PutCount        uint64 `json:"put_count"`
	GetBytes        uint64 `json:"get_bytes"`
	GetCount        uint64 `json:"get_count"`
	ListCount       uint64 `json:"list_count"`
	ListObjects     uint64 `json:"list_objects"`
	ListObjectBytes uint64 `json:"list_object_bytes"`
}

LabelStats holds per-label byte and request counters.

type Metered 

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

Metered wraps a Bucket and counts bytes/requests, bucketing them by the label returned from the Classify function. The label set is open: any string returned by Classify becomes a row in Stats.

func NewMetered 

func NewMetered(b Bucket, classify func(key string) string) *Metered

NewMetered wraps b with byte/request counters labelled by classify. Pass classify=nil for a single "all" label. The returned Bucket is safe for concurrent use.

func (*Metered) Delete 

func (m *Metered) Delete(ctx context.Context, key string) error

func (*Metered) Get 

func (m *Metered) Get(ctx context.Context, key string) (io.ReadCloser, string, error)

func (*Metered) GetRange 

func (m *Metered) GetRange(ctx context.Context, key string, off, length int64) (io.ReadCloser, error)

func (*Metered) List 

func (m *Metered) List(ctx context.Context, prefix, startAfter string) ([]ObjectInfo, error)

func (*Metered) Put 

func (m *Metered) Put(ctx context.Context, key string, body io.Reader, length int64, ifMatch *string) (string, error)

func (*Metered) PutStream 

func (m *Metered) PutStream(ctx context.Context, key string, body io.Reader, ifMatch *string) (string, error)

func (*Metered) Stat 

func (m *Metered) Stat(ctx context.Context, key string) (ObjectInfo, error)

func (*Metered) Stats 

func (m *Metered) Stats() Stats

Stats returns a snapshot of counters with totals summed.

type ObjectInfo 

type ObjectInfo struct {
	Key          string
	Size         int64
	LastModified time.Time
	ETag         string
}

ObjectInfo describes one object in a List response.

func FetchRangedAt 

func FetchRangedAt(ctx context.Context, b Bucket, key string, dst io.WriterAt, opts FetchOpts) (ObjectInfo, error)

FetchRangedAt downloads key into dst. Objects at or above the threshold are fetched as concurrent byte ranges written at their offsets (bounded memory: parts stream straight to dst), spreading the transfer across the S3 client's striped transports so a single path-pinned TCP connection cannot cap throughput; smaller objects use one GET. If the ranged path fails for any reason other than the object being missing or the context ending, it degrades to the single-stream path rather than failing the download.

Integrity checking stays with the caller (objectstore keys carry a sha the caller verifies before an atomic rename).

func FetchReader 

func FetchReader(ctx context.Context, b Bucket, key string, opts FetchOpts) (io.ReadCloser, ObjectInfo, error)

FetchReader is FetchRangedAt for streaming consumers: it returns a sequential reader over the object while parts are fetched concurrently ahead of the read position. Memory is bounded by (Concurrency+1) × PartSize of in-flight part buffers. Objects below the threshold return the plain (healed) Get body.

If the very first ranged part fails (e.g. a backend without Range support), the reader degrades to a single-stream Get. Later part failures surface as read errors, exactly like a broken plain body.

type S3 

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

S3 implements Bucket against an S3 or S3-compatible bucket via aws-sdk-go-v2/service/s3.

func OpenS3 

func OpenS3(ctx context.Context, cfg S3Config) (*S3, error)

OpenS3 constructs an S3 from cfg. It does not perform a network round-trip; the first call surfaces connectivity issues.

func (*S3) Delete 

func (a *S3) Delete(ctx context.Context, key string) error

func (*S3) Get 

func (a *S3) Get(ctx context.Context, key string) (io.ReadCloser, string, error)

func (*S3) GetRange 

func (a *S3) GetRange(ctx context.Context, key string, off, length int64) (io.ReadCloser, error)

func (*S3) List 

func (a *S3) List(ctx context.Context, prefix, startAfter string) ([]ObjectInfo, error)

func (*S3) Put 

func (a *S3) Put(ctx context.Context, key string, body io.Reader, length int64, ifMatch *string) (string, error)

func (*S3) PutStream 

func (a *S3) PutStream(ctx context.Context, key string, body io.Reader, ifMatch *string) (string, error)

func (*S3) Stat 

func (a *S3) Stat(ctx context.Context, key string) (ObjectInfo, error)

type S3Config 

type S3Config struct {
	Bucket string // required.

	// Prefix is prepended to every key. A trailing "/" is normalized.
	Prefix string

	// Region for the bucket. If empty, the SDK reads AWS_REGION /
	// AWS_DEFAULT_REGION from the environment / shared config.
	Region string

	// EndpointURL overrides the default S3 endpoint. Set this for
	// S3-compatible services (R2, B2, MinIO, GCS-with-S3-API). When
	// non-empty, UsePathStyle defaults to true unless explicitly false.
	EndpointURL string

	// UsePathStyle forces path-style addressing
	// (https://endpoint/bucket/key) instead of virtual-host style.
	// Required for MinIO and most non-AWS S3-compatible services.
	UsePathStyle bool

	// AccessKey/SecretKey/SessionToken override the SDK credential
	// chain. Leave empty to use the default chain (env, shared
	// config, IAM role, IMDSv2).
	AccessKey    string
	SecretKey    string
	SessionToken string

	// HTTPClient overrides the SDK's default HTTP client. Tests pass
	// custom clients; production almost never needs this.
	HTTPClient *http.Client

	// ForceIPv4 dials the endpoint over IPv4 only. Some S3-compatible
	// anycast endpoints (Tigris) route IPv6 to a far PoP while IPv4 lands
	// on a near one, so forcing IPv4 can roughly halve per-GET latency.
	// Ignored when HTTPClient is set.
	ForceIPv4 bool

	// Clients is the number of striped transports (distinct TCP
	// connections under HTTP/2) requests are spread across, with
	// rate-based eviction of path-pinned connections — see clientSet.
	// 0 = default (4). Ignored when HTTPClient is set.
	Clients int

	// MaxAttempts caps SDK-level retries. 0 = SDK default (3).
	MaxAttempts int
}

S3Config is the constructor input for OpenS3. Only Bucket is required; everything else has reasonable defaults.

type Stats 

type Stats struct {
	ByLabel   map[string]LabelStats `json:"by_label"`
	TotalGet  LabelStats            `json:"total_get"`
	TotalPut  LabelStats            `json:"total_put"`
	TotalList LabelStats            `json:"total_list"`
}

Stats is a snapshot of one Metered's counters, broken out by label plus separate Get, Put, and List totals across all labels.

Source Files

View all Source files

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.