clusterstore

package
v0.12.0 Latest Latest
Warning

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

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

Documentation

Overview

Package clusterstore is the replicating storage layer for go-faster/fs cluster mode (DESIGN.md §4, ROADMAP.md M3 Phase 7). It composes the pure cluster packages — placement (failure-domain-aware HRW), scheme/fragment (replication + Reed-Solomon codecs) and transport (authenticated peer API) — into the object data plane:

  • Writes ack only after the synchronous write quorum is durable (W=2 full replicas for the replica schemes, all k+m shards for EC) and the remainder — the RF=2.5 parity or the RF=3 trailing replica — is produced behind a bounded async queue. Sub-quorum writes are refused, never silently under-replicated.
  • Reads consult the object's sidecar (the per-object commit record) and stream the first available replica with open-time failover, or gather any k EC shards, reconstructing missing ones on the fly.

Each write gets a fresh generation stamp: fragments are named by generation and the sidecar — replaced atomically per store — is the commit point that makes a generation visible, so an overwrite never tears an existing object and stale generations are garbage-collected behind the async queue. Divergence between targets (e.g. two concurrent overwrites racing on different nodes) is reconciled by the repair worker, not the write path.

The fs.Storage implementation on top of this coordinator (bucket metadata, listing merge, multipart) is the next Phase 7 slice; see ROADMAP.md.

Index

Constants

View Source 
const DefaultSweepGrace = 10 * time.Minute

DefaultSweepGrace is the default protection window for unattributed generations — far beyond any write's commit latency.

Variables

View Source 
var (
	// ErrNotFound reports an object with no committed sidecar on any of its
	// placement targets.
	ErrNotFound = errors.New("object not found")
	// ErrInsufficientTargets mirrors fragment.ErrInsufficientTargets: the
	// topology cannot host the scheme's distinct targets, so the write is
	// refused rather than under-protected.
	ErrInsufficientTargets = fragment.ErrInsufficientTargets
	// ErrUnrecoverable mirrors fragment.ErrUnrecoverable: too many fragments
	// are gone to serve the object.
	ErrUnrecoverable = fragment.ErrUnrecoverable
)

Sentinel errors of the coordinator.

View Source 
var ErrSchemeRejected = errors.New("scheme rejected")

ErrSchemeRejected matches (via errors.Is) any error SetBucketScheme returns because the requested scheme does not parse or the current topology cannot host it. It is a client error: the request named a scheme this cluster cannot serve, not a control-plane failure.

Functions

This section is empty.

Types

type BucketInfo 

type BucketInfo struct {
	Version int       `json:"version"`
	Name    string    `json:"name"`
	ACL     fs.ACL    `json:"acl,omitempty"`
	Created time.Time `json:"created"`
	// Scheme overrides the cluster default replication scheme for this
	// bucket's objects ("rf2.5", "rf3", "ec:k,m"); empty applies the default.
	// Changing it affects new writes immediately; existing objects follow
	// through scheme conversion in repair/rebalance (ROADMAP Phase 8).
	Scheme string `json:"scheme,omitempty"`
}

BucketInfo is the replicated bucket record: the cluster's source of truth that a bucket exists, plus its bucket-level S3 state. True cluster-wide create/delete linearizability arrives with the etcd control plane; until then concurrent conflicting mutations from different nodes follow last-write-wins per target, like object sidecars.

type BucketTotals added in v0.11.0 

type BucketTotals struct {
	Objects int64
	Bytes   int64
}

BucketTotals is what one bucket holds.

type Config 

type Config struct {
	// Topology is the cluster topology source. Required.
	Topology TopologySource
	// Peers dials the peer holding a placement target. Required.
	Peers PeerDialer
	// Scheme resolves the replication scheme per bucket; nil applies
	// scheme.Default (RF=2.5) everywhere.
	Scheme SchemeFunc
	// QueueLen bounds the async remainder queue; when the queue is full the
	// remainder is produced synchronously instead (backpressure, never
	// dropped). Defaults to 128.
	QueueLen int
	// OnAsyncError observes failures of async remainder tasks (the write has
	// already been acknowledged at quorum; the object stays readable and the
	// repair worker will complete it). May be nil.
	OnAsyncError func(bucket, key string, err error)
	// Usage receives per-bucket object accounting deltas. Nil disables
	// accounting, which is what a cluster without a control plane (tests,
	// StaticTopology) does — the counters live in etcd.
	Usage UsageObserver
}

Config configures a Coordinator.

type Coordinator 

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

Coordinator is the cluster object data plane: quorum writes, failover reads and deletes of replicated/erasure-coded objects over the peer transport. It is safe for concurrent use.

func New 

func New(cfg Config) (*Coordinator, error)

New builds a Coordinator and starts its async remainder worker.

func (*Coordinator) Bucket 

func (c *Coordinator) Bucket(ctx context.Context, bucket string) (*BucketInfo, error)

Bucket returns the bucket record, or fs.ErrBucketNotFound.

func (*Coordinator) BucketExists 

func (c *Coordinator) BucketExists(ctx context.Context, bucket string) (bool, error)

BucketExists reports whether the bucket record exists.

func (*Coordinator) Close 

func (c *Coordinator) Close() error

Close drains the async queue and stops the worker. The coordinator remains usable for reads; subsequent writes produce their remainder synchronously.

func (*Coordinator) CountObjects added in v0.11.0 

func (c *Coordinator) CountObjects(ctx context.Context) (map[string]BucketTotals, error)

CountObjects totals every committed object in the cluster, per bucket.

This is the authoritative count and the expensive one: it is a scatter-gather over every disk that reads every sidecar, the same walk a listing does, for every bucket at once. One pass answers the whole cluster, which is why the recount is scheduled cluster-wide rather than per bucket.

Objects reachable only through an unreachable node are missed, so a recount taken during an outage undercounts. It is still an improvement on drifted counters, and the next recount corrects it.

func (*Coordinator) CreateBucket 

func (c *Coordinator) CreateBucket(ctx context.Context, bucket string, acl fs.ACL) error

CreateBucket records a new bucket, refusing when it already exists (fs.ErrBucketAlreadyExists). The existence check and the write are not atomic across nodes until the etcd control plane lands; a racing duplicate create converges to a single record.

func (*Coordinator) Delete 

func (c *Coordinator) Delete(ctx context.Context, bucket, key string) error

Delete removes an object: its sidecars first (the commit records — once they are gone the object is unreadable everywhere), then the generation's fragments, best-effort across the placement targets of every remembered epoch (a not-yet-relocated object still has state at the old placement). Deleting an absent object returns ErrNotFound.

func (*Coordinator) DeleteBucket 

func (c *Coordinator) DeleteBucket(ctx context.Context, bucket string) error

DeleteBucket removes a bucket's record from every target. The caller (the fs.Storage layer) is responsible for the emptiness check.

func (*Coordinator) EffectiveScheme 

func (c *Coordinator) EffectiveScheme(ctx context.Context, bucket string) scheme.Scheme

EffectiveScheme resolves the scheme new writes to the bucket use: the bucket record's override when set, the configured default otherwise (also when no bucket record exists — direct coordinator writes). Results are cached for schemeCacheTTL.

Resolution is lenient: when the record is momentarily unreachable the last cached value (expired or not) applies, then the configured default — a write is never refused over the bucket record, it lands durably at the fallback scheme and the next repair pass converts it. Conversion itself uses the strict resolveBucketScheme.

func (*Coordinator) Flush 

func (c *Coordinator) Flush()

Flush blocks until every async task enqueued so far has completed. It is a test and shutdown aid; new writes may enqueue more work concurrently.

func (*Coordinator) Get 

func (c *Coordinator) Get(ctx context.Context, bucket, key string) (*Sidecar, io.ReadCloser, error)

Get opens an object for reading: the sidecar is fetched from the first reachable placement target, then the payload streams from the first available replica (open-time failover) or is gathered from any k EC shards, reconstructing missing ones in memory. Fragment lookups fall back across remembered topology epochs, so an object whose relocation has not finished is still served from its previous placement. It returns ErrNotFound when no target holds a committed sidecar and ErrUnrecoverable when too many fragments are gone.

func (*Coordinator) ListBuckets 

func (c *Coordinator) ListBuckets(ctx context.Context) ([]BucketInfo, error)

ListBuckets gathers the bucket records from every disk in the cluster, deduplicated by name and sorted. Per-target failures are tolerated as long as every record remains reachable on some replica; if every listing fails the error surfaces.

func (*Coordinator) ListObjects 

func (c *Coordinator) ListObjects(ctx context.Context, bucket, prefix string) ([]*Sidecar, error)

ListObjects gathers the objects of a bucket with the given key prefix: every disk in the cluster is scanned for committed sidecars under the bucket's namespace and the results are merged by key, newest write wins (Modified, then generation as the tie-break for equal stamps). The listing is sorted by key.

Per-target failures are tolerated: an object's sidecar is replicated across its placement targets, so a listing stays complete while every object keeps at least one reachable sidecar (the same availability bound as reads). Only when every scan fails does the error surface. Listing reads each sidecar individually; a per-node listing index is a later optimization.

func (*Coordinator) Put 

func (c *Coordinator) Put(ctx context.Context, req *PutRequest) (*Sidecar, error)

Put writes an object at its bucket's scheme, acknowledging only once the synchronous write quorum is durable on distinct failure domains: the first two full replicas for RF=2.5/RF=3, or all k+m shards for EC. The remainder (RF=2.5 parity, the RF=3 third replica) and stale-generation cleanup happen behind the async queue. A write that cannot reach quorum is refused and leaves no new committed state.

func (*Coordinator) QueueDepth 

func (c *Coordinator) QueueDepth() int

QueueDepth reports the number of objects with pending async work: queued or executing replication remainders plus held repair slots. A sustained non-zero depth means the node is behind on producing its writes' trailing replicas/parity — the repair backlog indicator surfaced by the admin API.

func (*Coordinator) SetBucketACL 

func (c *Coordinator) SetBucketACL(ctx context.Context, bucket string, acl fs.ACL) error

SetBucketACL rewrites the bucket record with a new ACL.

func (*Coordinator) SetBucketScheme 

func (c *Coordinator) SetBucketScheme(ctx context.Context, bucket, schemeID string) error

SetBucketScheme rewrites the bucket record with a new object scheme override; empty restores the cluster default. The scheme must parse and the current topology must be able to host it (a bucket must never be switched into a scheme its cluster cannot write). Existing objects converge to the new scheme through repair/rebalance conversion.

func (*Coordinator) Stat 

func (c *Coordinator) Stat(ctx context.Context, bucket, key string) (*Sidecar, error)

Stat returns an object's sidecar without touching payload fragments.

func (*Coordinator) Topology 

func (c *Coordinator) Topology() *cluster.Topology

Topology returns the coordinator's current topology snapshot (for status reporting and readiness checks).

func (*Coordinator) UpdateSidecar 

func (c *Coordinator) UpdateSidecar(ctx context.Context, bucket, key string, mutate func(*Sidecar)) error

UpdateSidecar rewrites an object's committed metadata in place (tags, ACL — anything that does not touch payload fragments): the sidecar is fetched, mutated, and re-replicated to the object's targets, quorum synchronously and the remainder best-effort. Concurrent Put/Update races on the same key are the caller's to serialize (the fs.Storage layer holds a per-key lock); cross-node races follow last-write-wins per target like every sidecar write.

type FragmentWalker added in v0.12.0 

type FragmentWalker interface {
	WalkFragments(ctx context.Context, disk cluster.DiskID, after string, fn func(name string) error) error
}

FragmentWalker streams the fragment names on one of this node's disks, in lexicographic order, without materializing them.

The scrubber used to ask the peer transport for the whole listing, which meant every name on the disk — several per object — held as a string before the first one could be looked at. On a disk holding tens of millions of fragments that is gigabytes of strings, and it was the first thing to fail on a large node.

Order is the only thing the sweep needs: an object's entries are contiguous in it, so each namespace can be handled and dropped before the next begins.

after is a hint. An implementation may prune names at or before it — that is what makes resuming a pass cheap — so the caller applies its own boundary and must not depend on receiving them.

A nil walker falls back to a buffered listing over the transport, which is correct and is what every non-cluster deployment and most tests use.

type HTTPPeers 

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

HTTPPeers is the production PeerDialer: the node's own targets go straight to its local store, every other node through an authenticated transport.Client for its Addr. Clients are cached per address.

func NewHTTPPeers 

func NewHTTPPeers(self cluster.NodeID, local transport.Store, secret transport.Secret, httpClient *http.Client) *HTTPPeers

NewHTTPPeers builds the production dialer. httpClient may be nil for http.DefaultClient.

func (*HTTPPeers) Peer 

func (h *HTTPPeers) Peer(node cluster.Node) (Peer, error)

Peer implements PeerDialer.

type LocalPeer 

type LocalPeer struct {
	Store transport.Store
}

LocalPeer adapts a node-local transport.Store to the Peer interface, bypassing HTTP for the node's own fragments.

func (LocalPeer) Delete 

func (p LocalPeer) Delete(ctx context.Context, disk cluster.DiskID, name string) error

Delete implements Peer.

func (LocalPeer) Get 

func (p LocalPeer) Get(ctx context.Context, disk cluster.DiskID, name string) (io.ReadCloser, int64, error)

Get implements Peer.

func (LocalPeer) List 

func (p LocalPeer) List(ctx context.Context, disk cluster.DiskID, prefix string) ([]string, error)

List implements Peer.

func (LocalPeer) Put 

func (p LocalPeer) Put(ctx context.Context, disk cluster.DiskID, name string, size int64, body io.Reader) error

Put implements Peer. Exactly size bytes are copied; the fragment becomes visible only when the store's writer closes cleanly.

func (LocalPeer) Stat 

func (p LocalPeer) Stat(ctx context.Context, disk cluster.DiskID, name string) (int64, error)

Stat implements Peer.

type NodePlan 

type NodePlan struct {
	// Objects is how many objects have at least one fragment to place on this
	// node.
	Objects int
	// Bytes is the fragment payload volume to place on this node.
	Bytes int64
}

NodePlan is one destination node's share of a rebalance plan.

type Peer 

type Peer interface {
	Put(ctx context.Context, disk cluster.DiskID, name string, size int64, body io.Reader) error
	Get(ctx context.Context, disk cluster.DiskID, name string) (io.ReadCloser, int64, error)
	Stat(ctx context.Context, disk cluster.DiskID, name string) (int64, error)
	Delete(ctx context.Context, disk cluster.DiskID, name string) error
	List(ctx context.Context, disk cluster.DiskID, prefix string) ([]string, error)
}

Peer moves named fragment payloads to and from one node, local or remote. *transport.Client satisfies it for remote nodes.

type PeerDialer 

type PeerDialer interface {
	Peer(node cluster.Node) (Peer, error)
}

PeerDialer resolves a topology node to the Peer that reaches it. Dialing is on the write/read path, so implementations must cache.

type PutRequest 

type PutRequest struct {
	Bucket string
	Key    string
	// Size is the exact body length; the coordinator streams exactly this
	// many bytes.
	Size int64
	Body io.Reader

	Metadata fs.ObjectMetadata
	Tags     []fs.Tag
	ACL      fs.ACL
	Owner    fs.Owner
	// ETag overrides the stored ETag (multipart composite ETags); empty means
	// the content MD5.
	ETag string
}

PutRequest describes one object write.

type RebalanceCursor 

type RebalanceCursor struct {
	Bucket string `json:"bucket"`
	Key    string `json:"key"`
}

RebalanceCursor marks progress through the rebalance walk: every object up to and including (Bucket, Key) — in bucket order, key order within a bucket — has been processed. The zero value means "start from the beginning".

func DecodeRebalanceCursor 

func DecodeRebalanceCursor(s string) (RebalanceCursor, error)

DecodeRebalanceCursor parses a persisted cursor.

func (RebalanceCursor) Encode 

func (c RebalanceCursor) Encode() (string, error)

Encode serializes the cursor for persistence (etcd).

type RebalanceOptions 

type RebalanceOptions struct {
	// Resume skips every object at or before this cursor. Zero value walks
	// everything.
	Resume RebalanceCursor
	// Concurrency is how many objects repair in parallel (default 4). Peer
	// bandwidth is throttled separately; see ThrottledPeers.
	Concurrency int
	// Checkpoint, if set, persists the cursor after each completed batch: every
	// object at or before it is done, so a successor resumes there. A
	// checkpoint error aborts the pass (the runner has likely lost its
	// single-runner slot).
	Checkpoint func(ctx context.Context, cur RebalanceCursor) error
	// OnObject observes each processed object; rep is nil when err is set.
	// May be nil.
	OnObject func(bucket, key string, rep *RepairReport, err error)
}

RebalanceOptions configures one Rebalance pass.

type RebalancePlan 

type RebalancePlan struct {
	// Objects is how many objects were examined.
	Objects int
	// MisplacedObjects counts objects with at least one fragment absent from
	// its current-placement target.
	MisplacedObjects int
	// MisplacedBytes is the total fragment payload volume to move.
	MisplacedBytes int64
	// Unplannable counts objects whose current placement cannot be computed
	// (e.g. the topology cannot host the scheme); rebalance would skip them
	// too.
	Unplannable int
	// Nodes breaks the move volume down by destination node.
	Nodes map[cluster.NodeID]*NodePlan
}

RebalancePlan is the dry-run summary of a rebalance: what data is not yet at the current epoch's placement and where it has to go.

type RebalanceReport 

type RebalanceReport struct {
	// Buckets is how many buckets were walked.
	Buckets int
	// Objects is how many objects were fed through repair.
	Objects int
	// Relocated counts objects where repair changed anything (fragments moved,
	// sidecars rewritten, old copies retired).
	Relocated int
	// Failed counts objects whose repair errored (also reported to OnError and
	// OnObject); the pass continues past them.
	Failed int
	// Totals aggregates the per-object repair actions.
	Totals RepairReport
}

RebalanceReport summarizes one Rebalance pass.

type RepairReport 

type RepairReport struct {
	// RebuiltFragments counts fragments restored (missing, torn or corrupt).
	RebuiltFragments int
	// RewrittenSidecars counts targets whose sidecar was missing or stale and
	// was rewritten to the authoritative record.
	RewrittenSidecars int
	// DeletedStale counts swept names: superseded generations and orphaned
	// fragments on the object's targets.
	DeletedStale int
	// CorruptReplicas counts replicas whose payload failed checksum
	// verification (each is also rebuilt and counted in RebuiltFragments).
	CorruptReplicas int
	// Converted counts objects rewritten to their bucket's current scheme.
	Converted int
	// ECUnverified is set when the EC parity/data consistency check failed:
	// without per-shard digests no victim can be identified, so nothing is
	// rebuilt and the object needs attention.
	ECUnverified bool
}

RepairReport is what one RepairObject pass did.

func (*RepairReport) Changed 

func (r *RepairReport) Changed() bool

Changed reports whether the pass modified anything.

type Repairer 

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

Repairer is the scheme-aware repair worker (ROADMAP Phase 8): it restores an object to its scheme's full protection level — rebuilding lost or corrupt fragments, completing missing sidecar replicas, and sweeping stale generations. The scrubber walks a node's local disks and feeds every object it finds through RepairObject, so a cluster of periodically-scrubbing nodes converges after node/disk loss or missed async remainders.

func NewRepairer 

func NewRepairer(cfg RepairerConfig) (*Repairer, error)

NewRepairer builds a repair worker over the coordinator.

func (*Repairer) PlanRebalance 

func (r *Repairer) PlanRebalance(ctx context.Context) (*RebalancePlan, error)

PlanRebalance computes the dry-run plan: it walks every object like Rebalance would, compares the current epoch's placement against what each target actually holds (a stat per fragment — no payload is read) and totals the objects and bytes each node would receive. Nothing is modified.

func (*Repairer) Rebalance 

func (r *Repairer) Rebalance(ctx context.Context, opts RebalanceOptions) (*RebalanceReport, error)

Rebalance walks every bucket's objects in key order and repairs each at the current epoch's placement — the manual rebalance pass (ROADMAP Phase 8). Relocation is the repair engine's copy → verify → delete: an object is never below its protection level mid-move. Objects are processed in batches of Concurrency with the cursor checkpointed between batches, so a killed runner is resumed by a successor without re-walking finished work (re-repairing an already-healthy object is a no-op).

func (*Repairer) RepairObject 

func (r *Repairer) RepairObject(ctx context.Context, bucket, key string) (*RepairReport, error)

RepairObject restores one object to full protection at the current epoch's placement. It holds the object's async-work slot exclusively, so writes to the key wait for the repair (and the repair never races a pending remainder). Returns ErrNotFound when no committed sidecar is reachable and ErrUnrecoverable when too few fragments survive to rebuild.

func (*Repairer) Scrub 

func (r *Repairer) Scrub(ctx context.Context) (*ScrubReport, error)

Scrub walks this node's local disks and repairs every object found, cluster-wide: a missing remainder, a dead peer's fragment or a stale generation anywhere in the object's placement gets fixed, not just local state. Objects only reachable through other nodes' disks are covered by those nodes' scrubs.

type RepairerConfig 

type RepairerConfig struct {
	// Coordinator is the cluster data plane. Required.
	Coordinator *Coordinator
	// Self is this node's ID; Scrub walks its disks. Required for Scrub.
	Self cluster.NodeID
	// Verify enables checksum verification of replica fragments during
	// repair (recommended for scheduled scrubs).
	Verify bool
	// OnError observes per-object scrub failures. May be nil.
	OnError func(bucket, key string, err error)
	// SweepGrace is how long an unattributed generation (fragments no
	// committed record names — possibly another node's write mid-commit) is
	// left alone before the sweep may delete it. Defaults to
	// DefaultSweepGrace; only tests should lower it.
	SweepGrace time.Duration
	// ScrubState persists per-disk scrub progress so an interrupted pass
	// resumes instead of restarting. Nil disables resuming.
	ScrubState ScrubStateStore
	// Fragments streams this node's disks for the scrub sweep. Nil falls back
	// to a buffered listing over the peer transport, which holds every name on
	// the disk in memory.
	Fragments FragmentWalker
}

RepairerConfig configures a Repairer.

type SchemeFunc 

type SchemeFunc func(bucket string) scheme.Scheme

SchemeFunc resolves the replication scheme for a bucket.

type ScrubReport 

type ScrubReport struct {
	// Objects is how many distinct objects were fed through repair.
	Objects int
	// Repaired counts objects where the pass changed anything.
	Repaired int
	// Failed counts objects whose repair errored (also reported to OnError).
	Failed int
	// UnknownDirs counts object namespaces on local disks with no readable
	// local sidecar — undecidable without cross-checking; left untouched.
	UnknownDirs int
	// Totals aggregates the per-object repair actions.
	Totals RepairReport
}

ScrubReport summarizes one scrub pass over this node's disks.

type ScrubStateStore added in v0.12.0 

type ScrubStateStore interface {
	// LoadScrubState returns the disk's recorded progress. An unknown or
	// unreadable disk returns the zero state, not an error: losing a cursor
	// costs a restarted pass, never correctness.
	LoadScrubState(disk cluster.DiskID) cluster.ScrubState
	// SaveScrubState records progress. Errors are advisory — the caller keeps
	// scrubbing, because failing to save a cursor is not a reason to stop
	// verifying data.
	SaveScrubState(disk cluster.DiskID, state cluster.ScrubState) error
}

ScrubStateStore persists per-disk scrub progress so an interrupted pass resumes instead of restarting.

It is deliberately node-local. A scrub cursor describes disks only this node can read, so no other node could resume from it, and the control plane is the wrong home for state nobody else can use: it would mean a replicated, fsynced write every few hundred objects from every disk of every node, against an etcd budgeted for kilobytes of control-plane state — and it would stop a local durability process from making progress during a control-plane outage.

The interface is declared here, where it is consumed, and satisfied structurally by diskstore, which owns the disk roots. Neither package imports the other.

A nil store disables resuming: every pass then starts from the beginning, which is the behavior this replaced.

type Sidecar 

type Sidecar struct {
	Version int    `json:"version"`
	Bucket  string `json:"bucket"`
	Key     string `json:"key"`

	// Scheme is the replication scheme the object was written with, in its
	// config form ("rf2.5", "rf3", "ec:k,m"). Recording it makes reads immune
	// to later per-bucket scheme changes.
	Scheme string `json:"scheme"`
	// Size is the exact object length, needed to re-plan fragment sizes and
	// to unpad EC reconstructions.
	Size int64 `json:"size"`
	// Generation names the fragment set this sidecar commits.
	Generation string `json:"generation"`
	// Seq is a per-object write sequence (previous committed Seq + 1): the
	// primary "newest wins" ordering for list-merge and repair
	// reconciliation. Wall clocks are too coarse to order two writes of the
	// same key (observed on Windows), so time is only the cross-writer
	// tie-break.
	Seq int64 `json:"seq,omitempty"`
	// Modified is the write time; orders records with equal Seq (concurrent
	// writers that read the same previous state), with the generation string
	// as the final deterministic tie-break.
	Modified time.Time `json:"modified"`

	ETag string `json:"etag,omitempty"`
	// Checksum is the hex MD5 of the full object content (scrubber and
	// verify-on-read input; equal to ETag for single-part writes).
	Checksum string `json:"checksum,omitempty"`

	ContentType        string            `json:"content_type,omitempty"`
	CacheControl       string            `json:"cache_control,omitempty"`
	ContentDisposition string            `json:"content_disposition,omitempty"`
	ContentEncoding    string            `json:"content_encoding,omitempty"`
	UserMetadata       map[string]string `json:"user_metadata,omitempty"`
	Tags               []fs.Tag          `json:"tags,omitempty"`
	ACL                fs.ACL            `json:"acl,omitempty"`
	// Owner is the principal that wrote the object; absent in sidecars written
	// before owners were modeled.
	Owner fs.Owner `json:"owner,omitzero"`
}

Sidecar is the per-object commit record, replicated to the object's placement targets alongside its fragments. It is what makes a generation visible: a fragment set without a committed sidecar does not exist, and the sidecar carries everything the read path needs to re-plan the fragments — the scheme, the exact size and the generation stamp — plus the S3-level metadata the fs.Storage layer serves without touching payload bytes.

func (*Sidecar) ObjectMetadata 

func (sc *Sidecar) ObjectMetadata() fs.ObjectMetadata

ObjectMetadata converts the sidecar's header fields to the domain type.

func (*Sidecar) ParseScheme 

func (sc *Sidecar) ParseScheme() (scheme.Scheme, error)

ParseScheme returns the scheme the object was written with.

func (*Sidecar) Supersedes 

func (sc *Sidecar) Supersedes(other *Sidecar) bool

Supersedes reports whether this record is newer than other: Seq first, then Modified, then Generation — a total, deterministic order.

type StaticTopology 

type StaticTopology struct {
	T *cluster.Topology
}

StaticTopology is a fixed TopologySource for tests and static clusters.

func (StaticTopology) Topology 

func (s StaticTopology) Topology() *cluster.Topology

Topology implements TopologySource.

type Storage 

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

Storage is the cluster-backed fs.Storage: the S3 semantics layer over the Coordinator's replicated data plane and bucket registry. Every storagetest guarantee of the single-node backends applies here too (see the conformance test).

Conditional writes (If-Match / If-None-Match) are serialized by a per-key lock held across the check and the write, which makes them atomic against writers on this node. Cross-node conditional-write linearizability needs a cluster lock and arrives with the etcd control plane; until then S3 clients pinned to a node (or a sticky load balancer) get full CAS semantics. Unconditional writes never take that lock — see PutObject.

func NewStorage 

func NewStorage(c *Coordinator) *Storage

NewStorage wraps a Coordinator in the fs.Storage interface.

func (*Storage) AbortMultipartUpload 

func (s *Storage) AbortMultipartUpload(ctx context.Context, bucket, _, uploadID string) error

AbortMultipartUpload implements fs.Storage.

func (*Storage) BucketACL 

func (s *Storage) BucketACL(ctx context.Context, bucket string) (fs.ACL, error)

BucketACL implements fs.Storage.

func (*Storage) BucketExists 

func (s *Storage) BucketExists(ctx context.Context, bucket string) (bool, error)

BucketExists implements fs.Storage.

func (*Storage) CompleteMultipartUpload 

func (s *Storage) CompleteMultipartUpload(ctx context.Context, req *fs.CompleteMultipartUploadRequest) (*fs.CompleteMultipartUploadResponse, error)

CompleteMultipartUpload implements fs.Storage: the selected parts stream into one quorum-replicated object write with the composite S3 ETag, then the upload state is deleted.

func (*Storage) CreateBucket 

func (s *Storage) CreateBucket(ctx context.Context, bucket string) error

CreateBucket implements fs.Storage.

func (*Storage) CreateMultipartUpload 

func (s *Storage) CreateMultipartUpload(ctx context.Context, req *fs.CreateMultipartUploadRequest) (*fs.MultipartUpload, error)

CreateMultipartUpload implements fs.Storage.

func (*Storage) DeleteBucket 

func (s *Storage) DeleteBucket(ctx context.Context, bucket string) error

DeleteBucket implements fs.Storage, refusing to delete a non-empty bucket.

func (*Storage) DeleteObject 

func (s *Storage) DeleteObject(ctx context.Context, bucket, key string) error

DeleteObject implements fs.Storage.

func (*Storage) DeleteObjectTagging 

func (s *Storage) DeleteObjectTagging(ctx context.Context, bucket, key string) error

DeleteObjectTagging implements fs.Storage.

func (*Storage) GetObject 

func (s *Storage) GetObject(ctx context.Context, bucket, key string) (*fs.GetObjectResponse, error)

GetObject implements fs.Storage.

func (*Storage) GetObjectTagging 

func (s *Storage) GetObjectTagging(ctx context.Context, bucket, key string) ([]fs.Tag, error)

GetObjectTagging implements fs.Storage.

func (*Storage) ListBuckets 

func (s *Storage) ListBuckets(ctx context.Context) ([]fs.Bucket, error)

ListBuckets implements fs.Storage.

func (*Storage) ListMultipartUploads 

func (s *Storage) ListMultipartUploads(ctx context.Context, bucket string) ([]fs.MultipartUpload, error)

ListMultipartUploads implements fs.Storage.

func (*Storage) ListObjects 

func (s *Storage) ListObjects(ctx context.Context, bucket, prefix string) ([]fs.Object, error)

ListObjects implements fs.Storage.

func (*Storage) ListParts 

func (s *Storage) ListParts(ctx context.Context, bucket, key, uploadID string) ([]fs.Part, error)

ListParts implements fs.Storage.

func (*Storage) ObjectACL 

func (s *Storage) ObjectACL(ctx context.Context, bucket, key string) (fs.ACL, error)

ObjectACL implements fs.Storage.

func (*Storage) ObjectOwner added in v0.10.0 

func (s *Storage) ObjectOwner(ctx context.Context, bucket, key string) (fs.Owner, error)

ObjectOwner implements fs.Storage.

func (*Storage) PutObject 

func (s *Storage) PutObject(ctx context.Context, req *fs.PutObjectRequest) (*fs.PutObjectResponse, error)

PutObject implements fs.Storage. The conditional check and the write happen under the object's key lock, so concurrent conditional PUTs on this node resolve to a single winner.

func (*Storage) PutObjectTagging 

func (s *Storage) PutObjectTagging(ctx context.Context, bucket, key string, tags []fs.Tag) error

PutObjectTagging implements fs.Storage.

func (*Storage) SetBucketACL 

func (s *Storage) SetBucketACL(ctx context.Context, bucket string, acl fs.ACL) error

SetBucketACL implements fs.Storage.

func (*Storage) SetObjectACL added in v0.10.0 

func (s *Storage) SetObjectACL(ctx context.Context, bucket, key string, acl fs.ACL) error

SetObjectACL implements fs.Storage.

func (*Storage) UploadPart 

func (s *Storage) UploadPart(ctx context.Context, req *fs.UploadPartRequest) (*fs.Part, error)

UploadPart implements fs.Storage. Re-uploading a part number replaces it.

type ThrottledPeers 

type ThrottledPeers struct {
	// Dialer is the wrapped dialer. Required.
	Dialer PeerDialer
	// Limiter is the shared byte-rate limit; its burst caps the per-Read chunk.
	// Required.
	Limiter *rate.Limiter
}

ThrottledPeers wraps a PeerDialer with a shared bandwidth limit on fragment reads. Every byte a rebalance or repair pass moves is first read from a source peer, so limiting Get streams bounds the total data-movement rate of the process; metadata calls (Stat, List, Delete) stay unthrottled.

func (*ThrottledPeers) Peer 

func (t *ThrottledPeers) Peer(node cluster.Node) (Peer, error)

Peer implements PeerDialer.

type TopologySource 

type TopologySource interface {
	Topology() *cluster.Topology
}

TopologySource provides the current cluster topology snapshot. The etcd control plane implements it by caching its watch; tests and single-process clusters use StaticTopology.

type UsageObserver added in v0.11.0 

type UsageObserver interface {
	Observe(bucket string, objects, bytes int64)
}

UsageObserver receives per-bucket object accounting: how many objects and bytes a completed operation added or removed.

It is called after the operation has committed, from the request goroutine, and must not block on anything slow — accounting is bookkeeping about a write that already succeeded, so it can never be allowed to fail one. The etcd implementation batches deltas and flushes them on its own schedule.

Deltas are best-effort by construction. A node that dies between committing a write and reporting it leaves the counters short, and nothing here pretends otherwise: CountObjects is the authority, and a periodic recount is what makes the record true again.

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.