store

package
v0.2.9 Latest Latest
Warning

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

 Go to latest Published: Aug 13, 2026 License: Apache-2.0 Imports: 28 Imported by: 0

Documentation

Overview

Package store implements offshoot's object-storage layout: a minimal conditional-write Backend interface, the local-directory implementation, and the typed manifest/ref schema on top.

Index

Constants

View Source 
const (
	LayoutVersion = 2
	RefSchema     = 2
)

Variables

View Source 
var (
	ErrNotFound = errors.New("store: not found")
	ErrCAS      = errors.New("store: compare-and-swap conflict")
	// ErrCopyUnsupported is returned by CopyObject when a backend cannot
	// perform a server-side or filesystem-level copy of THIS particular
	// object — every backend offshoot ships supports CopyObject in general
	// (local: a filesystem clone or plain-copy fallback; S3: a real
	// server-side CopyObject call, single-request under 5GB and multipart
	// UploadPartCopy above it), but S3 still cannot copy a source over its
	// own 5TiB per-object ceiling under any strategy, so the sentinel fires
	// for that case (see store.S3.CopyObject's doc comment). Callers
	// (ops.Fork's fast path) must treat this as "fall back to the slow,
	// materialize-and-re-encode path" — it is a capability signal, not a
	// hard failure.
	ErrCopyUnsupported = errors.New("store: CopyObject not supported by this backend")
)
View Source 
var (
	// ErrLeaseHeld reports that another holder owns an unexpired lease.
	ErrLeaseHeld = errors.New("store: branch lease is held")
	// ErrLeaseLost reports that the caller no longer holds the lease it
	// claimed: someone reclaimed the branch, the caller's epoch is dead, and
	// anything it writes now lands in an unreferenced prefix.
	ErrLeaseLost = errors.New("store: branch lease lost")
	// ErrReaping reports that db@branch has an active reap claim
	// (Reaping=true). A lease acquired in the window between Reap's claim
	// and its Destroy call would have its branch deleted out from under it
	// (Destroy's own GetRef can still read the pre-claim ref, and DeleteRef
	// is unconditional), so AcquireLease refuses outright rather than race
	// it. The claim is transient: it clears when Reap's Destroy call
	// unwinds it (failure) or the branch is gone (success), or — for a
	// claim stranded by a crashed reaper — the next Reap cycle's self-heal
	// (see ops.reapOne). Retrying shortly is always the right move.
	ErrReaping = errors.New("store: branch is being reaped")
	// ErrDeleting reports that db@branch has an active Destroy claim
	// (Ref.Deleting=true) — the generalization of ErrReaping's TOCTOU fix
	// (Milestone 4 Task 6b) to every Destroy call, not just Reap's: a lease
	// acquired in the window between Destroy's GetRef and its delete would
	// otherwise have its branch deleted out from under it, so AcquireLease
	// refuses outright here too. Transient exactly like ErrReaping: it
	// clears when Destroy's own claim unwind runs (a failure after the
	// claim landed), the branch is gone (success — GetRef itself then
	// returns ErrNotFound instead of this), or — for a claim stranded by a
	// crashed Destroy — ops.ClearStaleDeleteClaims's age-based self-heal
	// (see Ref.Deleting's doc comment). Retrying shortly is always the
	// right move, same as ErrReaping.
	ErrDeleting = errors.New("store: branch is being deleted")
)
View Source 
var ErrNoCAS = errors.New("store: backend does not enforce conditional writes")

ErrNoCAS reports a backend that does not enforce conditional writes. offshoot's single-writer-per-ref guarantee rests entirely on these semantics, so a store that fails the probe is refused outright rather than used with silently weaker safety.

Functions

func BaseKey 

func BaseKey(lineage string) string

BaseKey locates a lineage's durable base pointer. It lives under LineagePrefix(lineage) alongside the lineage's snapshot/segment objects, so it is swept together with the lineage by GC. It is deliberately NEITHER snapshot- nor segment-prefixed, so ParseMemberKey returns ok=false for it and Chain never mistakes it for a materialization member.

func LeaseLive 

func LeaseLive(ref Ref, now time.Time) bool

LeaseLive reports whether ref carries a lease that is still live at now: a holder is recorded AND its expiry parses AND that expiry is still in the future. Exported so callers outside this package that need the exact same liveness verdict AcquireLease itself uses — ops.BranchStateAt's "active" branch state, in particular — never independently reimplement (and risk drifting from) this check. A LeaseExpiry that fails to parse is treated as not live here, the same fail-open-to-reclaimable stance AcquireLease takes for corrupt expiries, just without that call's own stderr warning (a read-only liveness check has no "reclaim" action to warn about).

func LineagePrefix 

func LineagePrefix(lineage string) string

func NewLineageID 

func NewLineageID() string

func ProbeCAS 

func ProbeCAS(b Backend) error

ProbeCAS verifies create-only and compare-and-swap enforcement. offshoot's single-writer-per-ref guarantee rests entirely on these semantics, so a store that fails the probe is refused outright rather than used with silently weaker safety.

func RefKey 

func RefKey(db, branch string) string

func SegmentKey 

func SegmentKey(lineage string, epoch, minTXID, maxTXID uint64) string

SegmentKey locates an incremental segment covering (minTXID, maxTXID]. Sorting by key sorts by maxTXID, so a lexical List is already in apply order.

func SnapshotKey 

func SnapshotKey(lineage string, epoch, txid uint64) string

func StoreIdentity 

func StoreIdentity(spec string) (string, error)

StoreIdentity returns a canonical identity string for spec: it captures the fully resolved backend configuration (including env-derived S3 settings), not just the literal spec text. Two spec strings that resolve to the same backend (e.g. "s3://b/p" and "s3://b/p/") yield the same identity; the same spec string resolving to different backends across sessions (e.g. OFFSHOOT_S3_ENDPOINT pointed at MinIO one session and real AWS the next) yields different identities.

This exists so callers that need a stable, collision-safe cache key for per-store local state (e.g. ops.checkoutRoot's checkout cache directory) can key off what the store actually IS rather than the raw spec string, which can be ambiguous.

func ValidateName 

func ValidateName(name string) error

Types

type Backend 

type Backend interface {
	Get(key string) (data []byte, etag string, err error)
	Put(key string, data []byte) error
	PutIf(key string, data []byte, ifMatch string) (string, error)
	List(prefix string) ([]string, error)
	Delete(key string) error
	// CopyObject makes dst a byte-identical copy of src within the same
	// backend. Returns ErrNotFound if src does not exist, or
	// ErrCopyUnsupported if this backend cannot perform the copy at all —
	// see ErrCopyUnsupported's doc comment for what callers must do with
	// that.
	//
	// CopyObject OVERWRITES an existing dst — like Put, not like the
	// create-only PutIf every snapshot/segment write otherwise uses. Today's
	// only caller (ops.Fork's fast path) always mints a fresh destination
	// key (a brand-new lineage's snapshot key) that nothing else can be
	// racing to write, so this is never observably different from
	// create-only in practice; it is specified as overwrite because that is
	// what a rename-into-place (the local backend) and a single-request
	// server-side copy (S3, Task 6b) both do natively, and requiring
	// create-only here would mean an extra existence check or conditional
	// request for no caller that needs it. A future caller that DOES need
	// create-only-or-fail must check for an existing dst itself (e.g. via
	// Get) before calling CopyObject.
	CopyObject(dst, src string) error
}

func OpenBackend 

func OpenBackend(ctx context.Context, spec string) (Backend, error)

OpenBackend resolves a store spec to a Backend and verifies that it enforces conditional writes.

Specs: 

/path or ./path     local directory
file:///abs/path    local directory
s3://bucket/prefix  S3-compatible bucket (AWS S3, R2, Tigris, MinIO)

S3 endpoint, region and addressing style come from OFFSHOOT_S3_ENDPOINT, OFFSHOOT_S3_REGION and OFFSHOOT_S3_PATH_STYLE; credentials come from the AWS SDK default chain.

OpenBackend runs unconditionally on every call, i.e. every CLI invocation: ProbeCAS's ~8 sequential round trips (see probe.go) are re-paid each time against a remote store, not cached across invocations. That is deliberate fail-closed behavior, not an oversight — a long-lived daemon (Plan 4) is the intended way to amortize the probe across many operations instead of weakening or skipping it here.

type BasePointer 

type BasePointer struct {
	Lineage string `json:"lineage"`
	TXID    uint64 `json:"txid"`
}

BasePointer names the fork point a shared child reads through for anything at or below Base.TXID: Lineage identifies the ancestor lineage, TXID is the fork point within it. Deliberately just these two fields — no epoch (see Ref.Base's doc comment for why).

type BatchDeleter added in v0.2.2 

type BatchDeleter interface {
	DeleteObjects(keys []string) (deleted []string, err error)
}

BatchDeleter, ReaderGetter, and ReaderPutter (below) are all discovered by TYPE ASSERTION against a caller's store.Backend value (b.(store.BatchDeleter), etc.), never through the Backend interface itself. That means a wrapper type that implements Backend by embedding another Backend (e.g. for instrumentation, caching, or retry logic) SILENTLY HIDES whichever of these three the embedded value actually implements: the embedding type's own method set does not include them unless it redeclares each one, so the type assertion fails and the caller falls back to the buffered/ per-key path. That fallback stays CORRECT — nothing breaks — but it silently loses the batching/streaming benefit the capability exists for (perf audits H2/H3). A wrapper that wants to preserve a capability must forward that method explicitly, not just embed and hope. The one in-repo wrapper today, ops.markCache (internal/ops/gc.go, used only in the GC mark walk), implements store.Backend directly rather than embedding one, and needs none of these three capabilities for what it does (Get/List only) — so nothing is affected today.

BatchDeleter is an optional Backend capability: delete many keys in as few round trips as the backend allows (S3: the DeleteObjects API, 1000 keys per request — perf audit H2; Local: a plain loop, no RPC to save but the same contract so callers stay uniform). It is deliberately NOT part of Backend itself: test wrappers and future backends keep working unchanged, and callers (ops' GC sweep) type-assert and fall back to per-key Delete.

DeleteObjects returns the keys it SUCCESSFULLY deleted — so a caller pruning per-key state (GC tombstones) prunes exactly those — plus an error describing any failures. A key that did not exist counts as successfully deleted (idempotent), matching Delete. Empty input is a no-op: (nil, nil).

type ChainMember 

type ChainMember struct {
	Key              string
	Snapshot         bool
	MinTXID, MaxTXID uint64
	Epoch            uint64
}

ChainMember identifies one object in a materialization chain.

func ParseMemberKey 

func ParseMemberKey(key string) (ChainMember, bool)

ParseMemberKey parses a snapshot or segment key back into a ChainMember.

type Checkpoint 

type Checkpoint struct {
	TXID  uint64 `json:"txid"`
	Epoch uint64 `json:"epoch"`
	// CreatedAt is when this checkpoint was created, RFC3339 UTC, stamped by
	// every ops call site that creates one (Create's "init", Checkpoint,
	// Fork's "fork", Promote's "promote", and a named session flush).
	// Omitempty so a checkpoint written before this field existed (or a v1
	// ref's upgraded bare-number checkpoints, which have no creation time to
	// recover) decodes with it empty rather than a fabricated value.
	CreatedAt string `json:"created_at,omitempty"`
	// Meta is a small user-supplied string->string map describing this
	// specific checkpoint (e.g. eval run id, git SHA, agent id), capped at
	// the ops layer (ops.ValidateMeta: at most 32 keys, keys <= 64 bytes,
	// values <= 512 bytes — see the design spec's metadata cap). Omitempty;
	// nil/absent means no metadata was given.
	Meta map[string]string `json:"meta,omitempty"`
}

Checkpoint locates a snapshot object: its transaction id and the epoch of the prefix it was written under. Epoch matters because acquiring or reclaiming a branch bumps the epoch, and objects stay where they were written.

type ConditionalDeleter 

type ConditionalDeleter interface {
	DeleteIf(key, ifMatch string) error
}

ConditionalDeleter is an optional Backend capability: DeleteIf removes key only if its current content still matches ifMatch's etag, refusing with ErrCAS if the key has since changed or is already gone. It is the delete- side counterpart to PutIf's CAS.

Local implements this via the exact same per-key lock PutIf itself uses (see local.go's DeleteIf) — a true conditional delete. S3's DeleteObject API has no compare-and-delete precondition to give it (If-Match/If-None- Match are PUT/GET-only headers there; DELETE ignores them entirely), so S3 deliberately does NOT implement this interface rather than pretending to honor a condition it cannot actually enforce — see s3.go's Delete doc comment and DeleteRefIf below for what a caller gets instead.

type Lease 

type Lease struct {
	DB, Branch string
	Holder     string
	Epoch      uint64
	Expiry     time.Time
}

Lease is a claim on a branch, valid until Expiry unless renewed.

type Local 

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

Local is a directory-backed Backend. CAS is implemented with a per-key O_CREAT|O_EXCL lock file: acquire lock -> read+verify etag -> write temp, fsync, rename -> release lock. A bare rename alone is atomic REPLACE, not compare-and-swap; the lock provides the compare step.

func NewLocal 

func NewLocal(root string) (*Local, error)

func (*Local) CopyObject 

func (l *Local) CopyObject(dst, src string) error

CopyObject makes dst a byte-identical copy of src, using reflink.CopyFile (a filesystem clone on APFS/btrfs/xfs-with-reflink, a plain byte copy otherwise) so the copy is near-constant-time when the filesystem supports it. Like write(), it copies to a uniquely-named temp file first and renames into place, so a reader of dst never observes a partial file and a failure partway through leaves dst untouched.

func (*Local) Delete 

func (l *Local) Delete(key string) error

func (*Local) DeleteIf 

func (l *Local) DeleteIf(key, ifMatch string) error

DeleteIf implements store.ConditionalDeleter: a true compare-and-delete, using the exact same per-key O_CREAT|O_EXCL lock file PutIf uses to implement its own compare-and-swap (see the package doc comment on Local). Absent (already deleted, or never existed) or content that no longer hashes to ifMatch both fail with ErrCAS — a caller (ops.Destroy) that raced this against a concurrent write to the same key sees exactly the same "your compare failed, retry" signal PutIf's own callers already know how to handle.

func (*Local) DeleteObjects added in v0.2.2 

func (l *Local) DeleteObjects(keys []string) (deleted []string, err error)

DeleteObjects implements store.BatchDeleter as a plain sequential loop over Delete: a local directory has no round trips to batch away, but implementing the capability keeps callers (ops' GC sweep) on one uniform code path across backends. Semantics follow the interface contract and Delete exactly — a missing file counts as deleted (os.Remove's IsNotExist is already success in Delete above); the first real error stops the loop and is returned alongside the keys deleted so far, which matches what a per-key fallback loop would have done.

func (*Local) Get 

func (l *Local) Get(key string) ([]byte, string, error)

func (*Local) GetReader added in v0.2.2 

func (l *Local) GetReader(key string) (io.ReadCloser, string, error)

GetReader implements store.ReaderGetter: it opens the file directly rather than os.ReadFile-ing its full contents into memory (unlike Get), so a caller applying a large object (e.g. a snapshot/segment during chain materialization) holds only one open file descriptor, not the whole object's bytes. A missing file maps to store.ErrNotFound, same as Get. The caller MUST Close the returned file.

etag is always "" here: computing the real content etag (etagOf, a sha256 over the full data) would require reading the whole file, which defeats the point of a streaming Get. A caller that needs the content etag should use Get instead.

func (*Local) List 

func (l *Local) List(prefix string) ([]string, error)

func (*Local) Put 

func (l *Local) Put(key string, data []byte) error

func (*Local) PutIf 

func (l *Local) PutIf(key string, data []byte, ifMatch string) (string, error)

func (*Local) PutReader added in v0.2.2 

func (l *Local) PutReader(key string, r io.Reader, size int64) error

PutReader implements store.ReaderPutter's unconditional overwrite: same contract as Put, streamed via writeReader instead of write.

func (*Local) PutReaderIf added in v0.2.2 

func (l *Local) PutReaderIf(key string, r io.Reader, size int64, ifMatch string) (string, error)

PutReaderIf implements store.ReaderPutter's CAS write: same contract and same per-key lock as PutIf, streamed via writeReader instead of write.

The ifMatch == "" (create-only) case deliberately checks existence with os.Stat rather than PutIf's os.ReadFile: PutIf already holds its new payload buffered in the caller's []byte, so reading the old content too (needed for the ifMatch != "" comparison below) costs nothing extra end-to-end. Here the whole point of the call is to avoid buffering a large object — reading a potentially-large EXISTING orphan into memory just to discover it exists (flush.go's create-only retry after a crashed prior attempt is exactly this case: an existing, possibly multi-GB, object at objKey) would defeat that for the one case this method's caller actually uses. The ifMatch != "" branch still needs the old content's hash to compare, same as PutIf, and reads it the same way; no caller in this codebase exercises that branch on a large object today.

type Manifest 

type Manifest struct {
	LayoutVersion int    `json:"layout_version"`
	CreatedAt     string `json:"created_at"`
}

type ReaderGetter added in v0.2.2 

type ReaderGetter interface {
	GetReader(key string) (r io.ReadCloser, etag string, err error)
}

ReaderGetter is an optional Backend capability: fetch an object as a stream instead of a fully-buffered []byte, so a caller applying a large object (e.g. ops' chain materialization — perf audit H3) need not hold it all in memory at once. Like BatchDeleter, it is deliberately NOT part of Backend itself: test wrappers and backends that only ever handle small objects keep working unchanged, and callers type-assert and fall back to the buffered Get.

The caller MUST Close the returned reader on every path, including a mid-read error — the underlying stream (an S3 response body, an open local file descriptor) is not otherwise released. etag follows Get's contract where the backend can supply it without reading the body (S3: the GetObject response's ETag header, free); a backend for which computing a real content etag would require reading the whole object (defeating the point of streaming) MAY return "" instead — a caller that needs a content etag should use Get, not GetReader.

type ReaderPutter added in v0.2.2 

type ReaderPutter interface {
	PutReaderIf(key string, r io.Reader, size int64, ifMatch string) (etag string, err error)
	PutReader(key string, r io.Reader, size int64) error
}

ReaderPutter is an optional Backend capability: upload an object by STREAMING from r (exactly size bytes) instead of a fully-buffered []byte, so a caller uploading a large object (e.g. a session flush's snapshot encode — perf audit H3, write side) need not hold it all in memory at once. Like BatchDeleter/ReaderGetter, it is deliberately NOT part of Backend itself: test wrappers and backends that only ever handle small objects keep working unchanged, and callers type-assert and fall back to the buffered Put/PutIf.

PutReaderIf mirrors PutIf's CAS contract exactly: ifMatch == "" means create-only (fails with store.ErrCAS if the key already exists), otherwise it is a compare-and-swap against the given etag, failing with store.ErrCAS on a mismatch or on a missing key. PutReader mirrors Put: an unconditional overwrite.

r must yield exactly size bytes. A conforming implementation MUST NOT buffer the whole object in memory to satisfy either method — S3's streams the SDK's PutObject body directly (ContentLength: size); Local's streams to a temp file and renames into place, matching Put/PutIf's existing write-then-rename discipline.

type Ref 

type Ref struct {
	Schema  int    `json:"schema"`
	Lineage string `json:"lineage"`
	// Epoch identifies a lineage's current writer generation. AcquireLease
	// bumps it on every fresh acquisition (and on reclaim of a dead lease —
	// see lease.go), fencing out whatever a superseded holder might still
	// write: chain resolution collapses members establishing state at the
	// same TXID down to the highest epoch (keepHighestEpoch), so a fenced
	// writer's stragglers lose deterministically.
	Epoch       uint64                `json:"epoch"`
	HeadTXID    uint64                `json:"head_txid"`
	HeadEpoch   uint64                `json:"head_epoch"`
	Checkpoints map[string]Checkpoint `json:"checkpoints"`
	Parent      string                `json:"parent,omitempty"`
	Protected   bool                  `json:"protected"`
	// Lease fields are empty when no writer holds the branch.
	LeaseHolder string `json:"lease_holder,omitempty"`
	LeaseExpiry string `json:"lease_expiry,omitempty"` // RFC3339Nano UTC
	// TTL fields govern branch reaping (Plan 8). TTL is a Go duration string
	// ("2h"); empty means the branch is never reaped. TouchedAt is the
	// activity clock: Reaping (ops.Reap, Task 2) measures TTL from the later
	// of TouchedAt and LeaseExpiry, so any durable write or lease activity
	// defers expiry. Reaping is set by Reap's CAS claim (Task 2) and, while
	// true, refuses ops.Touch. All three are omitempty so a Plan-7 ref
	// decodes with them empty/false — no TTL means never reaped.
	TTL       string `json:"ttl,omitempty"`
	TouchedAt string `json:"touched_at,omitempty"` // RFC3339Nano UTC
	Reaping   bool   `json:"reaping,omitempty"`
	// Deleting and DeletingAt are ops.Destroy's own CAS claim (Milestone 4
	// Task 6b), the same shape as Reaping/TouchedAt above but a DELIBERATELY
	// SEPARATE field, not a unification of the two: Destroy CAS-writes
	// Deleting=true (stamping DeletingAt) before it does anything
	// irreversible, so a lease acquired in the window between Destroy's
	// initial GetRef and its actual delete refuses outright (AcquireLease
	// checks this exactly like it already checks Reaping) instead of racing
	// a branch out from under its new holder — the Destroy TOCTOU the M2
	// review documented. This was scoped as a sibling to Reaping rather than
	// a generalized {none,reaping,deleting} claim enum per Milestone 4 Task
	// 6b's timebox (PM Amendment 9): Reaping's CAS mechanics are
	// torture/race-tested and deliberately left untouched here. DeletingAt
	// (RFC3339Nano UTC, stamped at claim time) is Deleting's own staleness
	// clock — see ops.ClearStaleDeleteClaims — since a Deleting claim, unlike
	// Reaping, has no TTL/activity deadline to recompute against; it just
	// self-heals by age. Both omitempty: a ref decodes with Deleting false
	// and DeletingAt empty exactly like every pre-Task-6b ref.
	Deleting   bool   `json:"deleting,omitempty"`
	DeletingAt string `json:"deleting_at,omitempty"`
	// Meta is a small user-supplied string->string map describing this
	// branch's lineage (e.g. eval run id, git SHA, agent id), set by Fork
	// and capped at the ops layer (ops.ValidateMeta). Branch-level lineage
	// is the grain — no row-level provenance. Omitempty, no schema bump: a
	// ref written before this field existed decodes with it nil, same as
	// the TTL fields above.
	Meta map[string]string `json:"meta,omitempty"`
	// Base is the copy-on-write shared-fork pointer: when set, this
	// lineage's objects alone do not cover reads at targetTXID <= Base.TXID
	// — resolution must fall through to Base.Lineage (transitively, if that
	// lineage has its own Base) for anything at or below the fork point.
	// This is DELIBERATELY DISTINCT from Parent (a human-readable breadcrumb
	// with no resolution or GC meaning): Base is live, load-bearing data
	// that both Chain resolution (chainFrom's fall-through) and GC
	// reachability (ops.reachableObjects' base-spine marking) consult. Nil
	// means this ref has no base — every pre-CoW ref, and every fork that
	// fully materializes (a plain Fork SHARES by default, setting this,
	// whenever the resolved chain is below the fork-time depth bound; it
	// materializes — nil Base — only at the floor or under the test hooks,
	// see ops.Fork). No epoch field: an epoch here would let a fenced writer's
	// stale base survive past the point it was superseded, re-creating the
	// fenced-orphan bug epoch fencing exists to prevent — see the design
	// spec's "The base pointer" section. Omitempty, no schema bump: a ref
	// written before this field existed decodes with it nil.
	Base *BasePointer `json:"base,omitempty"`
}

func (*Ref) SetCheckpoint 

func (r *Ref) SetCheckpoint(name string, cp Checkpoint)

SetCheckpoint records name -> cp, allocating the map if needed.

func (*Ref) Touch 

func (r *Ref) Touch(now time.Time)

Touch stamps the ref's activity clock. Reaping (ops.Reap) measures TTL from the later of this stamp and the lease expiry, so any durable write or lease activity defers expiry.

type S3 

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

S3 is a Backend over S3-compatible object storage using conditional writes for compare-and-swap. Etags are provider-issued opaque tokens: they are returned and replayed verbatim, never parsed or compared to a hash.

Timeout model — four layers, each documented in full at its own decl:

  1. Transport: s3ResponseHeaderTimeout bounds every call's wait for response headers to begin arriving; never bounds a body transfer.
  2. Single-shot RPCs: singleShotRPCTimeout is the total per-call bound, scaled by payload size when known (singleShotDeadline / singleShotFloorBytesPerSecond).
  3. Multipart uploads: multipartRPCTimeout bounds each part/metadata RPC; multipartAbortTimeout the deferred abort (s3_multipart.go).
  4. Streaming reads: readProgressTimeout is GetReader's per-Read progress watchdog (watchdogReader) — kills a stalled stream, never a slow-but-progressing one.

func NewS3 

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

func (*S3) CopyObject 

func (s *S3) CopyObject(dst, src string) error

CopyObject makes dst a server-side copy of src, without downloading or re-uploading the bytes through this process — for sources at or under copyObjectMaxBytes (S3's 5 GiB single-request CopyObject limit) via one CopyObject call below; for larger sources, up to s3MaxObjectBytes (S3's 5 TiB per-object ceiling), via copyObjectMultipart's multipart UploadPartCopy sequence instead. Only a source over s3MaxObjectBytes — a size no S3 mechanism can copy at all — returns ErrCopyUnsupported; see store.Backend's doc comment for the overwrite-on-existing-dst contract this honors either way (S3's Copy/CompleteMultipartUpload overwrite dst natively; there is nothing extra to do here for that).

The size check is a HEAD request before the copy, not a check against whatever CopyObject itself returns on failure: S3's actual behavior for an over-limit single-request copy is an EntityTooLarge error, which this backend could also detect and translate, but checking first means a caller never pays for (and never has to unwind after) an API call that was always going to fail — and it is the same HEAD this method needs anyway to translate a missing source into store.ErrNotFound, since a 404 from CopyObject itself is not reliably distinguishable between "source missing" and "destination bucket misconfigured" the way isNotFound's typed-error path is for Get.

func (*S3) Delete 

func (s *S3) Delete(key string) error

Delete removes key unconditionally. S3 deliberately does NOT implement store.ConditionalDeleter here: DeleteObject has no compare-and-delete precondition in the S3 API (If-Match/If-None-Match only apply to GetObject/PutObject; DeleteObject accepts no conditional headers at all), so there is no honest way to make this a true CAS delete the way Local's DeleteIf is. Callers that need delete-time safety on this backend (Task 6b's Destroy claim-guard, in particular) get it from the claim-marker pattern instead — a CAS-written Ref.Deleting claim that a concurrent AcquireLease is taught to refuse — not from this method; see store.DeleteRefIf's doc comment for the full reasoning.

func (*S3) DeleteObjects added in v0.2.2 

func (s *S3) DeleteObjects(keys []string) (deleted []string, err error)

DeleteObjects implements store.BatchDeleter: keys are chunked into batches of at most 1000 (the API's limit) and each batch is one DeleteObjects round trip — 80k objects cost ~80 RPCs instead of 80k serial DeleteObject calls (perf audit H2). Batches are issued sequentially; that already collapses the sweep's wall time by three orders of magnitude, and issuing them concurrently would need a rate/ error story S3's 503-slowdown behavior makes nontrivial, so concurrency is deliberately left for later.

Per the interface contract it returns the keys (in the caller's un-prefixed form) that were actually deleted, plus an error naming any keys the API reported per-key Errors for. A key that did not exist counts as deleted (S3's DeleteObjects reports absent keys under Deleted), same as Delete. A transport-level batch failure returns the keys deleted by the batches that DID complete plus the error.

func (*S3) Get 

func (s *S3) Get(key string) ([]byte, string, error)

func (*S3) GetReader added in v0.2.2 

func (s *S3) GetReader(key string) (io.ReadCloser, string, error)

GetReader implements store.ReaderGetter: it returns the GetObject response's Body directly, without buffering it into memory first (unlike Get, which io.ReadAlls it) — so a caller applying a large object (e.g. a snapshot/segment during chain materialization) holds only the current object's stream open, not its full bytes. Same key namespacing and ErrNotFound mapping as Get. The caller MUST Close the returned reader; leaving it open leaks the underlying HTTP connection.

Stall protection: the returned stream outlives this call, so it CANNOT run under a singleShotRPCTimeout-style total deadline the way Get does — that would kill a legitimate long read of a large object. Instead the request runs under a cancelable context and the Body is wrapped in a watchdogReader: a single Read that blocks for readProgressTimeout with no bytes at all cancels the request, failing the Read with a recognizable "read stalled" error, while any progress at all re-arms the window and a slow CONSUMER (long pauses BETWEEN Reads) is never affected — see watchdogReader's doc comment. The stream's only production consumer (ops' lazyReader, chain materialization) treats any Read error as fatal and closes everything, so the watchdog error needs no special handling.

func (*S3) List 

func (s *S3) List(prefix string) ([]string, error)

func (*S3) Put 

func (s *S3) Put(key string, data []byte) error

func (*S3) PutIf 

func (s *S3) PutIf(key string, data []byte, ifMatch string) (string, error)

func (*S3) PutReader added in v0.2.2 

func (s *S3) PutReader(key string, r io.Reader, size int64) error

PutReader implements store.ReaderPutter's unconditional overwrite. For size <= multipartThreshold it issues a single PutObject with Body: r and ContentLength: size, so the SDK never needs to buffer r's content to determine its length (unlike Put, which wraps an already-in-memory []byte in bytes.NewReader). The SDK's own payload-hash-for-signing step (SigV4) does not require buffering either: over HTTPS (the normal case) S3 uses UNSIGNED-PAYLOAD and skips hashing the body at all; over plain HTTP it streams the body through a SHA256 hasher and rewinds (r must support io.Seeker, true of the *os.File this backend's only caller — flush.go's snapshot upload — passes) rather than holding it in memory.

For size > multipartThreshold (lifting S3's 5 GiB single-PutObject ceiling) it instead uses a multipart upload — see putMultipart's doc comment for the mechanics, part sizing, and the abort-on-every-error-path guarantee that makes this safe to call. PutReader sets no conditions on the multipart Complete (unconditional, matching the single-PUT path).

func (*S3) PutReaderIf added in v0.2.2 

func (s *S3) PutReaderIf(key string, r io.Reader, size int64, ifMatch string) (string, error)

PutReaderIf implements store.ReaderPutter's CAS write. For size <= multipartThreshold it issues identical ifMatch-to-precondition-header translation as PutIf (create-only via IfNoneMatch: "*", CAS via IfMatch) on a single PutObject, with Body: r/ContentLength: size in place of a buffered []byte — see PutReader's doc comment for why that does not require buffering the object.

For size > multipartThreshold it uses a multipart upload instead — see putMultipart's doc comment. The condition (IfNoneMatch/IfMatch) is placed on CompleteMultipartUpload, not on CreateMultipartUpload or any UploadPart call (the SDK's CompleteMultipartUploadInput supports both fields), and a precondition rejection there maps to store.ErrCAS via the same isPreconditionFailed/isNotFound helpers and the same error wording as the single-PUT path below, so CAS semantics are indistinguishable to a caller regardless of which path an object's size took.

One observable difference: a multipart object's ETag is NOT its MD5 the way a single-PUT object's is — S3 returns "<md5-of-the-part-md5s>-<part count>" instead (e.g. "d41d8cd9...-3"), a valid opaque etag for future If-Match calls but not a content hash. offshoot never parses or hashes etags (see S3's type doc comment), and this backend's only production caller (flush.go's snapshot upload) discards PutReaderIf's returned etag entirely, so this is harmless in practice — noted here for any future caller that might assume otherwise.

type S3Config 

type S3Config struct {
	Bucket       string
	Prefix       string // optional key prefix, no leading slash; "" allowed
	Endpoint     string // optional custom endpoint (R2/Tigris/MinIO/fake)
	Region       string // defaults to "auto" when Endpoint is set, else SDK default chain
	UsePathStyle bool   // required for MinIO and the fake
}

S3Config describes an S3-compatible bucket. Credentials come from the AWS SDK default chain (env, shared config, IAM role).

type Store 

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

func (*Store) AcquireLease 

func (s *Store) AcquireLease(db, branch, holder string, ttl time.Duration, now time.Time) (Lease, error)

AcquireLease claims db@branch for holder until now+ttl.

A ref with an active reap claim (Reaping=true) or an active Destroy claim (Deleting=true, Milestone 4 Task 6b) refuses outright — see ErrReaping/ ErrDeleting — before any of the lease logic below even runs.

A fresh acquisition, or a reclaim of an expired (or corrupt, see below) lease, bumps the epoch so any previous holder's subsequent writes are fenced into a dead prefix. But if the caller already holds a live lease (same holder, not yet expired), AcquireLease is instead an idempotent renew: it extends the expiry and returns the SAME epoch, exactly like RenewLease. Bumping in that case would fence the holder's own in-flight writes, which is never what a re-acquiring holder wants; a caller that genuinely wants a fresh epoch must ReleaseLease then AcquireLease again.

A LeaseExpiry that is present but fails to parse is corruption, not an available lease — but it is still treated as fail-open (reclaimable) here rather than fail-closed, because fail-closed would brick the branch permanently with no recovery path. The reclaim is logged to stderr so the corruption doesn't pass silently.

If PutRef loses a concurrent-acquire race (ErrCAS), that is reported as an error wrapping BOTH ErrLeaseHeld and ErrCAS: a caller that only checks ErrLeaseHeld still sees "someone else holds it," while a caller that wants the low-level detail can still find ErrCAS via errors.Is.

func (*Store) BaseSpine 

func (s *Store) BaseSpine(lineage string) ([]string, error)

BaseSpine returns every ANCESTOR lineage in lineage's base spine, nearest first: the lineage's own base, that base's base, and so on until a lineage with no base object. The starting lineage itself is NOT included; a lineage with no base returns an empty spine. It is a thin walk over lineageBase — deliberately NOT a reimplementation of Chain's member resolution — for callers (GC's reachability mark) that need the set of lineages whose base.json objects resolution will read, independent of which of them contribute chain members: a pass-through lineage (forked but never diverged) contributes zero members yet its base.json IS read.

An INVARIANT of the walk: lineage L's base.json exists exactly when the walk out of L found a base — i.e. BaseKey(L) exists for the starting lineage iff the spine is non-empty, and for spine[i] iff i+1 < len(spine) (the walk ended at spine[len-1] precisely because it has no base object). GC leans on this to mark exactly the existing base.json objects without extra existence probes.

WriteLineageBase's create-only, must-point-at-an-existing-lineage rules mean a cycle cannot be constructed through this binary's writers, but the walk still guards against one defensively (a corrupt or hand-edited store): revisiting a lineage fails loudly rather than looping forever.

func (*Store) Chain 

func (s *Store) Chain(lineage string, target uint64) ([]ChainMember, error)

Chain returns the members needed to materialize lineage at target: the newest snapshot with MaxTXID <= target, followed by every segment after it up to target, in apply order. It returns an error when no snapshot covers the target or the segments do not form a contiguous run — a caller must never be handed a chain with a hole.

It lists LineagePrefix(lineage) once and works from the parsed keys, so it costs one List regardless of epoch count — segments from superseded epochs are simply members like any other, since an epoch bump does not move objects.

Copy-on-write base following (the one hard rule): if the lineage has a durable base pointer (see lineageBase), resolution follows it under a STRICT, non-negotiable invariant — resolution NEVER merges members across lineages. Each half of a shared chain is resolved wholly within one lineage's List, and keepHighestEpoch is only ever run on a single lineage's members. A cross-lineage union would let a higher-epoch parent object win the child's own txid range and silently serve the parent's timeline.

Like BaseSpine, the base recursion guards defensively against a cycle: WriteLineageBase's create-only, no-self-base, must-point-at-an-existing- lineage rules mean this binary's writers cannot construct one, but a corrupt or hand-edited store could — revisiting a lineage fails loudly with a cycle error rather than recursing until the stack overflows.

func (*Store) CheckManifest 

func (s *Store) CheckManifest() error

func (*Store) DeleteRefIf 

func (s *Store) DeleteRefIf(db, branch, etag string) error

DeleteRefIf deletes db@branch's ref, conditional on etag when the backend can actually honor that (ConditionalDeleter — today, Local) and unconditional otherwise (S3). Milestone 4 Task 6b's Destroy claim-guard (ops.Workspace.Destroy) is what actually closes the GetRef -> lease-check -> delete TOCTOU on EVERY backend, by CAS-writing a Deleting claim before calling this — see Ref.Deleting's doc comment. DeleteRefIf's own etag-conditioning is a belt-and-suspenders extra on a backend that can give it for free, never the primary safety mechanism, precisely because S3 cannot give it at all: pretending otherwise here would be the "pretend S3 DeleteObject has preconditions" mistake the design review flagged.

func (*Store) EnsureLayoutV2 

func (s *Store) EnsureLayoutV2() error

EnsureLayoutV2 CAS-bumps the store's manifest to LayoutVersion 2 if it is currently below that; it is a no-op if the manifest is already at 2 or newer. It is idempotent and CAS-safe under concurrent callers: if this call's CAS loses the race because another caller already bumped the manifest to >= 2, that is re-checked and treated as success, not failure. Called by ops.Fork's SHARE path before writing the first base ref, since a base pointer must never land in a store an old binary could still open.

The ">= 2" observation is memoized on the Store (layoutAtLeastV2, see its field comment for why that is safe): after the first success this costs zero RPCs, so a swarm of shared forks pays the manifest Get once, not once per fork.

func (*Store) GetRef 

func (s *Store) GetRef(db, branch string) (Ref, string, error)

func (*Store) InitManifest 

func (s *Store) InitManifest() error

func (*Store) ListRefs 

func (s *Store) ListRefs() (map[string][]string, error)

func (*Store) PutRef 

func (s *Store) PutRef(db, branch string, r Ref, ifMatch string) (string, error)

func (*Store) ReleaseLease 

func (s *Store) ReleaseLease(l Lease) error

ReleaseLease clears the caller's lease. The epoch is left alone: a clean release means the holder's own objects stay reachable.

func (*Store) RenewLease 

func (s *Store) RenewLease(l Lease, ttl time.Duration, now time.Time) (Lease, error)

RenewLease extends the caller's own lease without touching the epoch.

func (*Store) WriteLineageBase 

func (s *Store) WriteLineageBase(lineage string, b BasePointer) error

WriteLineageBase durably records a lineage's base pointer, create-only: a lineage's base is immutable once written, so a second write is refused with ErrCAS (surfaced as-is). The shared-fork path (ops.Fork) is the real caller; it writes the base BEFORE the child ref CAS so resolution's source of truth exists by the time any reader can see the child.

A base naming the lineage itself is refused outright: Chain resolves a based lineage by recursing into its base, so a self-base would recurse forever. (A longer cycle through several lineages would too, but bases are create-only and always point at an ALREADY-EXISTING lineage, so a cycle cannot be constructed through this writer; the self-base is the one cheap, local mistake worth guarding here.)

Source Files

View all Source files

Directories

Path Synopsis
Package storetest provides the shared Backend contract suite and an in-process S3 subset server, so every backend is verified against exactly the same expectations.
 Package storetest provides the shared Backend contract suite and an in-process S3 subset server, so every backend is verified against exactly the same expectations.

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.