s3keys

package
v0.0.0-...-3ab43b7 Latest Latest
Warning

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

 Go to latest Published: Nov 14, 2026 License: AGPL-3.0 Imports: 10 Imported by: 0

Documentation

Index

Constants

View Source 
const (
	ChunkRefRCPrefix       = "!s3|chunkref-rc|"
	ChunkBlobGCQueuePrefix = "!s3|chunkblob-gc-queue|"
)

Reference counting and GC eligibility for content-addressed chunkblobs (design §3.5).

Two keyspaces, both Raft-replicated: 

!s3|chunkref-rc|<sha-hex>                    -> ChunkRefRC
!s3|chunkblob-gc-queue|<commitTS>|<sha-hex>  -> empty

Every timestamp here is an elastickv HLC commit timestamp — (UnixMilli << HLCLogicalBits) | logical — NOT Unix nanoseconds. The queue key is built from the commitTS of the txn that drove the reference count to zero, so the sweeper's grace boundary has to be expressed in the same domain. Mixing the two silently produces a boundary off by roughly six orders of magnitude, which would either sweep everything immediately or never sweep at all.

The queue key carries the commit timestamp in its NAME rather than its value, and that is the whole point: a counter sitting at zero records *that* a blob became reclaimable but not *when*, so the grace window would be unimplementable. Big-endian fixed-width encoding makes the queue sort by eligibility time, so a sweeper finds everything past the grace boundary with one range scan ending at ChunkBlobGCQueueScanEnd(now - grace).

View Source 
const (
	// DefaultChunkBlobOrphanScanInterval is the §3.5 proposed cadence.
	// Deliberately far longer than the sweep interval: this is a
	// safety net for crash paths, not a routine reclaim.
	DefaultChunkBlobOrphanScanInterval = time.Hour

	// DefaultChunkBlobOrphanGracePeriod is how long a local blob must
	// have existed before an absent RC entry is read as an abort
	// rather than an upload in progress. It must exceed the longest
	// plausible interval between writing chunkblob bytes and
	// committing the chunkref that references them.
	DefaultChunkBlobOrphanGracePeriod = 6 * time.Hour
)

The §3.5 orphan scan: the safety net behind two paths the queue scan structurally cannot see.

  • A sweeper that crashed between the Raft conditional delete and the local unlink. The queue entry is gone, so no queue scan will ever revisit it, but the blob is still on disk.
  • A PUT that wrote the chunkblob to local Pebble and then aborted before dispatching its chunkref (admission 503, client disconnect, push quorum failure, context cancel). Neither an RC entry nor a queue entry was ever written.

Because the second case is indistinguishable from an IN-FLIGHT PUT — the bytes land before the chunkref commits, so a healthy upload briefly has no RC entry at all — the scan is gated on the blob's own age. Without that gate it would delete the payload out from under every concurrent upload.

View Source 
const (
	OrphanReasonNoReferenceRecord = "no_reference_record"
	OrphanReasonSweeperCrashed    = "sweeper_crashed"
	OrphanReasonWithinGrace       = "within_grace"
	OrphanReasonReferenced        = "referenced"
	OrphanReasonQueueOwnsIt       = "queue_owns_it"
	OrphanReasonRecordUnreadable  = "record_unreadable"
	OrphanReasonPending           = "pending"
	OrphanReasonPayloadChanged    = "payload_changed"
)

Reasons, a closed set suitable for a metric label.

View Source 
const (
	SweepReasonUnreferenced     = "unreferenced"
	SweepReasonReferencedAgain  = "referenced_again"
	SweepReasonSupersededEntry  = "superseded_entry"
	SweepReasonRecordUnreadable = "record_unreadable"
	SweepReasonRecordNotQueued  = "record_not_queued"
)

Reasons, a closed set so a sweeper can use them as a metric label.

View Source 
const (
	// DefaultChunkBlobGCInterval is the §3.5 proposed sweep cadence.
	DefaultChunkBlobGCInterval = 5 * time.Minute

	// DefaultChunkBlobGCGracePeriod is how long a blob must sit
	// unreferenced before it may be reclaimed. It has to exceed the
	// longest window in which a live reader could still be holding a
	// chunkref it read before the dereferencing txn committed.
	DefaultChunkBlobGCGracePeriod = time.Hour
)

The §3.5 node-local sweeper loop.

Each node runs this independently; correctness across nodes comes from the Raft-replicated queue key, whose single-writer-per-key property serialises concurrent sweepers — only the sweeper whose conditional delete commits proceeds to the local phase.

The collaborators are narrow interfaces rather than a store handle so the loop's ordering guarantees are testable without a Raft group or Pebble.

View Source 
const (
	BucketMetaPrefix       = "!s3|bucket|meta|"
	BucketGenerationPrefix = "!s3|bucket|gen|"
	ObjectManifestPrefix   = "!s3|obj|head|"
	UploadMetaPrefix       = "!s3|upload|meta|"
	UploadPartPrefix       = "!s3|upload|part|"
	BlobPrefix             = "!s3|blob|"
	ChunkRefPrefix         = "!s3|chunkref|"
	ChunkBlobPrefix        = "!s3|chunkblob|"
	GCUploadPrefix         = "!s3|gc|upload|"
	RoutePrefix            = "!s3route|"
)
View Source 
const BucketCleanupPrefix = "!s3|bucket|cleanup|"

BucketCleanupPrefix marks bucket generations whose post-delete safety net has not finished. The marker is committed with bucket metadata deletion.

View Source 
const DefaultChunkBlobGCPageSize = 1024

DefaultChunkBlobGCPageSize bounds one queue-scan page.

Chosen so the sweeper's memory is independent of the backlog: a million-entry queue becomes slow to drain rather than impossible to load.

View Source 
const MaxRetainedSweepFailures = 64

maxRetainedSweepFailures caps how many per-entry errors one pass keeps for its joined return value. Every failure is still counted; only the detail is sampled, because a backlog large enough to matter is also large enough for one error per entry to be the memory problem rather than the diagnosis. Exported so a test can assert the cap actually bites rather than hard-coding a number that could drift past the fixture size.

Variables

View Source 
var ErrChunkRefPeerFieldTooLong = errors.New("s3 chunkref replica field is too long")

ErrChunkRefPeerFieldTooLong is returned when a replica node ID or address cannot be represented by the V2 chunkref format.

View Source 
var ErrChunkRefRCUnderflow = errors.New("s3keys: chunkref reference count would underflow")

ErrChunkRefRCUnderflow reports a decrement that would drive a reference count below zero.

It fails the txn rather than clamping. A count that underflows means the caller's view of which chunkrefs exist disagrees with the stored record, and clamping to zero would queue a blob for deletion on the strength of that disagreement — turning a bookkeeping bug into data loss.

View Source 
var ErrChunkRefReplicaSetTooLarge = errors.New("s3 chunkref replica set is too large")

ErrChunkRefReplicaSetTooLarge is returned when a write-time replica set cannot be represented by the V2 chunkref format.

View Source 
var ErrChunkRefSourcePeerTooLong = errors.New("s3 chunkref source peer is too long")

ErrChunkRefSourcePeerTooLong is returned when a ChunkRefValue source peer cannot be encoded into the fixed-width value header.

View Source 
var ErrInvalidChunkRefPlan = errors.New("s3keys: invalid chunkref reference-count plan")

ErrInvalidChunkRefPlan reports a malformed planning request.

View Source 
var ErrQueueEntryChanged = errors.New("s3keys: gc queue entry changed before the conditional delete")

ErrQueueEntryChanged reports that a conditional queue delete lost its precondition: the entry was already gone, or the reference count is no longer zero.

This is the load-bearing error of the whole design. §3.5 is explicit that an UNCONDITIONAL delete would silently succeed on an already-absent entry and let the sweeper go on to local-delete a chunkblob that is currently live. Receiving this means another actor won the race and the sweeper MUST NOT touch the blob.

Functions

func BlobKey 

func BlobKey(bucket string, generation uint64, object string, uploadID string, partNo uint64, chunkNo uint64) []byte

func BlobPrefixForBucket 

func BlobPrefixForBucket(bucket string, generation uint64) []byte

func BlobPrefixForUpload 

func BlobPrefixForUpload(bucket string, generation uint64, object string, uploadID string) []byte

BlobPrefixForUpload returns the key prefix that covers all blob chunks for a specific multipart upload. Used for cleanup of all chunks in an upload.

func BucketCleanupKey 

func BucketCleanupKey(bucket string, generation uint64) []byte

func BucketGenerationKey 

func BucketGenerationKey(bucket string) []byte

func BucketGenerationRoutePrefixForCleanupPrefix 

func BucketGenerationRoutePrefixForCleanupPrefix(prefix []byte) ([]byte, bool)

func BucketMetaKey 

func BucketMetaKey(bucket string) []byte

func BucketScopedRoutePrefix 

func BucketScopedRoutePrefix(prefix []byte) ([]byte, bool)

BucketScopedRoutePrefix projects a bucket-scoped raw family prefix -- the shape bucketScopedPrefix builds, family bytes followed by the bucket segment and the generation u64 -- onto the object route prefix every key underneath it routes through.

It reports false unless the prefix ends exactly at the generation. A shorter prefix can still span more than one bucket or generation, so it has no single route range to project onto, and a longer one reaches into the object segment this function does not parse.

func ChunkBlobGCGraceBoundary 

func ChunkBlobGCGraceBoundary(nowTS uint64, grace time.Duration) uint64

ChunkBlobGCGraceBoundary converts a wall-clock grace period into the HLC boundary timestamp a sweeper passes to ChunkBlobGCQueueScanEnd.

It exists so callers never have to open-code the HLC layout, which is where the domain confusion would creep in: the queue keys carry HLC commit timestamps, so subtracting a duration means subtracting milliseconds from the PHYSICAL half, not nanoseconds from the whole value.

A grace period that reaches back past the epoch clamps to zero rather than wrapping, so an absurd configuration sweeps nothing instead of sweeping everything.

func ChunkBlobGCQueueKey 

func ChunkBlobGCQueueKey(commitTS uint64, contentSHA256 [chunkBlobSHA256Bytes]byte) []byte

ChunkBlobGCQueueKey builds the eligibility-queue key for a content hash that became unreferenced at commitTS.

The timestamp is fixed-width big-endian so the queue sorts by eligibility time; a decimal or variable-width encoding would order 9 after 10 and silently break the grace-boundary scan. A zero commitTS is rejected by the caller contract: ChunkRefRC uses zero as its "no queue entry" sentinel, so a record genuinely queued at timestamp zero would report Queued() == false and its entry would become unreachable. A real HLC commit timestamp is never zero — the physical half is Unix milliseconds — so this costs nothing.

func ChunkBlobGCQueueScanEnd 

func ChunkBlobGCQueueScanEnd(boundaryTS uint64) []byte

ChunkBlobGCQueueScanEnd is the EXCLUSIVE upper bound for a sweeper scan covering everything that became eligible strictly before boundaryTS.

boundaryTS is an HLC commit timestamp, not a Unix nanosecond count. Build it with ChunkBlobGCGraceBoundary rather than from time.Now().UnixNano(), which is a different domain entirely.

Exclusivity matters: passing the current timestamp would sweep a blob that became eligible this instant, skipping the grace window entirely. An entry stamped exactly at the boundary is excluded — it has not yet served the full grace.

func ChunkBlobGCQueueScanStart 

func ChunkBlobGCQueueScanStart() []byte

ChunkBlobGCQueueScanStart is the inclusive lower bound for a sweeper scan: the start of the whole queue.

func ChunkBlobKey 

func ChunkBlobKey(contentSHA256 [chunkBlobSHA256Bytes]byte) []byte

ChunkBlobKey builds a content-addressed blob key from a SHA-256 digest.

func ChunkRefKey 

func ChunkRefKey(bucket string, generation uint64, object string, uploadID string, partNo uint64, chunkNo uint64) []byte

func ChunkRefPrefixForBucket 

func ChunkRefPrefixForBucket(bucket string, generation uint64) []byte

func ChunkRefPrefixForUpload 

func ChunkRefPrefixForUpload(bucket string, generation uint64, object string, uploadID string) []byte

ChunkRefPrefixForUpload returns the key prefix that covers all offload chunk references for a specific multipart upload.

func ChunkRefRCKey 

func ChunkRefRCKey(contentSHA256 [chunkBlobSHA256Bytes]byte) []byte

ChunkRefRCKey builds the reference-count key for a content hash.

func EncodeChunkRefRC 

func EncodeChunkRefRC(rc ChunkRefRC) []byte

EncodeChunkRefRC encodes a reference-count record.

func EncodeChunkRefValue 

func EncodeChunkRefValue(ref ChunkRefValue) ([]byte, error)

EncodeChunkRefValue encodes a chunk reference value using a versioned binary shape so future offload metadata can fail closed on unknown versions.

func EncodeSegment 

func EncodeSegment(raw []byte) []byte

func EncodeSegmentPrefix 

func EncodeSegmentPrefix(raw []byte) []byte

func ExtractRouteKey 

func ExtractRouteKey(key []byte) []byte

func GCUploadKey 

func GCUploadKey(bucket string, generation uint64, object string, uploadID string) []byte

func GCUploadPrefixForBucket 

func GCUploadPrefixForBucket(bucket string, generation uint64) []byte

func ManifestScanRouteBounds 

func ManifestScanRouteBounds(start []byte, end []byte) ([]byte, []byte, bool)

func ObjectManifestKey 

func ObjectManifestKey(bucket string, generation uint64, object string) []byte

func ObjectManifestPrefixForBucket 

func ObjectManifestPrefixForBucket(bucket string, generation uint64) []byte

func ObjectManifestScanStart 

func ObjectManifestScanStart(bucket string, generation uint64, objectPrefix string) []byte

func ParseBlobKey 

func ParseBlobKey(key []byte) (bucket string, generation uint64, object string, uploadID string, partNo uint64, chunkNo uint64, partVersion uint64, ok bool)

ParseBlobKey decodes a !s3|blob| key into its components. Both the 6-component form produced by BlobKey and the 7-component form produced by VersionedBlobKey (with partVersion != 0) are supported; partVersion is reported as zero for the un-versioned form.

Returns ok=false on any parse failure (truncated key, malformed segment, junk trailer). Used by the offline backup decoder (internal/backup) to route blob chunks to their assembled object, and by future readers that need to walk the blob keyspace without holding a live cluster.

func ParseBucketCleanupKey 

func ParseBucketCleanupKey(key []byte) (string, uint64, bool)

func ParseBucketGenerationKey 

func ParseBucketGenerationKey(key []byte) (string, bool)

func ParseBucketMetaKey 

func ParseBucketMetaKey(key []byte) (string, bool)

func ParseChunkBlobGCQueueKey 

func ParseChunkBlobGCQueueKey(key []byte) (uint64, [chunkBlobSHA256Bytes]byte, bool)

ParseChunkBlobGCQueueKey decodes an eligibility-queue key into the timestamp at which the blob became unreferenced and its content hash.

func ParseChunkBlobKey 

func ParseChunkBlobKey(key []byte) ([chunkBlobSHA256Bytes]byte, bool)

ParseChunkBlobKey decodes a !s3|chunkblob| key into its SHA-256 digest.

func ParseChunkRefKey 

func ParseChunkRefKey(key []byte) (bucket string, generation uint64, object string, uploadID string, partNo uint64, chunkNo uint64, ok bool)

ParseChunkRefKey decodes a !s3|chunkref| key into its object-part location.

func ParseChunkRefRCKey 

func ParseChunkRefRCKey(key []byte) ([chunkBlobSHA256Bytes]byte, bool)

ParseChunkRefRCKey decodes a reference-count key.

func ParseObjectManifestKey 

func ParseObjectManifestKey(key []byte) (bucket string, generation uint64, object string, ok bool)

func ParseUploadPartKey 

func ParseUploadPartKey(key []byte) (bucket string, generation uint64, object string, uploadID string, partNo uint64, ok bool)

ParseUploadPartKey extracts bucket, generation, object, uploadID, and partNo from a part descriptor key.

func ParseVersionedChunkRefKey 

func ParseVersionedChunkRefKey(key []byte) (bucket string, generation uint64, object string, uploadID string, partNo uint64, chunkNo uint64, partVersion uint64, ok bool)

ParseVersionedChunkRefKey decodes both the original key shape and the attempt-versioned shape. partVersion is zero for original keys.

func RouteKey 

func RouteKey(bucket string, generation uint64, object string) []byte

func RoutePrefixForBucket 

func RoutePrefixForBucket(bucket string, generation uint64) []byte

func RoutePrefixForBucketAnyGeneration 

func RoutePrefixForBucketAnyGeneration(bucket string) []byte

func UploadMetaKey 

func UploadMetaKey(bucket string, generation uint64, object string, uploadID string) []byte

func UploadMetaPrefixForBucket 

func UploadMetaPrefixForBucket(bucket string, generation uint64) []byte

UploadMetaPrefixForBucket / UploadPartPrefixForBucket / BlobPrefixForBucket / ChunkRefPrefixForBucket / GCUploadPrefixForBucket / RoutePrefixForBucket each isolate to a single bucket+generation tuple. Used by AdminDeleteBucket's DEL_PREFIX safety net (design doc 2026_04_28_implemented_admin_delete_bucket_safety_net.md): the bucket- must-be-empty empty-probe is racy against concurrent PutObject, so the delete commit also DEL_PREFIXes every per-bucket key family to sweep anything that snuck in during the readTS → commitTS window.

Each prefix encodes the bucket segment then the generation u64 in the same shape as the per-key constructors that build entries under each family. Re-creating the bucket bumps generation so old orphans under the old generation prefix stay isolated from the new bucket.

func UploadPartKey 

func UploadPartKey(bucket string, generation uint64, object string, uploadID string, partNo uint64) []byte

func UploadPartPrefixForBucket 

func UploadPartPrefixForBucket(bucket string, generation uint64) []byte

func UploadPartPrefixForUpload 

func UploadPartPrefixForUpload(bucket string, generation uint64, object string, uploadID string) []byte

UploadPartPrefixForUpload returns the key prefix that covers all part descriptors for a specific multipart upload. Used to scan/list parts for ListParts and cleanup.

func VersionedBlobKey 

func VersionedBlobKey(bucket string, generation uint64, object string, uploadID string, partNo uint64, chunkNo uint64, partVersion uint64) []byte

VersionedBlobKey returns the blob key for a specific part attempt identified by partVersion (typically the part's commit timestamp). When partVersion is 0 the result is identical to BlobKey, preserving backward compatibility with data written before versioned keys were introduced. When partVersion is non-zero the key includes an extra u64 suffix so concurrent re-uploads of the same part (e.g. retry or parallel UploadPart for the same partNo) write to a distinct key space, making blob data immutable per attempt.

func VersionedChunkRefKey 

func VersionedChunkRefKey(bucket string, generation uint64, object string, uploadID string, partNo uint64, chunkNo uint64, partVersion uint64) []byte

VersionedChunkRefKey returns the chunk reference key for one immutable part attempt. A zero version preserves the original key shape for existing data and single PUTs whose upload ID is already unique.

Types

type ChunkBlobGCQueueEntry 

type ChunkBlobGCQueueEntry struct {
	CommitTS      uint64
	ContentSHA256 [chunkBlobSHA256Bytes]byte
}

ChunkBlobGCQueueEntry is one entry returned by a queue scan.

type ChunkBlobLocalStore 

type ChunkBlobLocalStore interface {
	// ChunkBlobWrittenAt returns when this node wrote the payload, as an
	// HLC timestamp, and whether it is present. Read BEFORE the Raft
	// phase so the unlink afterwards can be conditioned on the state the
	// sweep actually observed.
	ChunkBlobWrittenAt(ctx context.Context, sha [chunkBlobSHA256Bytes]byte) (uint64, bool, error)

	// DeleteChunkBlobIfUnchanged unlinks the blob only if it is still the
	// payload written at writtenAtTS, reporting false when it is not.
	//
	// Conditional because DeleteGCQueueEntryIfUnreferenced guarantees a
	// zero reference count only up to ITS OWN commit. A PUT that reuses
	// the SHA immediately afterwards re-anchors the payload and commits a
	// reference, and an unconditional unlink then removes bytes that PUT
	// has already acknowledged as durable. Same contract as
	// ChunkBlobOrphanLocalStore: the comparison and the unlink must be
	// atomic with respect to the local writer.
	DeleteChunkBlobIfUnchanged(
		ctx context.Context, sha [chunkBlobSHA256Bytes]byte, writtenAtTS uint64,
	) (bool, error)
}

ChunkBlobLocalStore is the node-local half: the chunkblob payload in Pebble, never written through Raft.

type ChunkBlobOrphanDecision 

type ChunkBlobOrphanDecision struct {
	Verdict ChunkBlobOrphanVerdict
	Reason  string
}

ChunkBlobOrphanDecision is a verdict plus its reason.

func ClassifyChunkBlobOrphan 

func ClassifyChunkBlobOrphan(
	blob LocalChunkBlob, boundaryTS uint64, rcValue []byte, rcFound, queueFound bool,
) ChunkBlobOrphanDecision

ClassifyChunkBlobOrphan applies the §3.5 detection criterion plus the age gate.

rcValue/rcFound are the raw reference-count record; queueFound reports whether a GC queue entry exists for this SHA. Raw bytes again, so an undecodable record stays distinguishable from an absent one — the first means reachability is unknown and the scan must decline, the second is the legitimate never-referenced state this scan exists to clean up.

type ChunkBlobOrphanLocalStore 

type ChunkBlobOrphanLocalStore interface {
	// ListLocalChunkBlobs enumerates this node's chunkblobs.
	ListLocalChunkBlobs(ctx context.Context) ([]LocalChunkBlob, error)
	// DeleteChunkBlobIfUnchanged unlinks the blob only if it is still the
	// payload written at writtenAtTS, reporting false when it is not.
	//
	// Conditional, not a plain delete, because a PUT that reuses this SHA
	// re-anchors the local payload before committing its chunkref: a
	// rewrite therefore moves WrittenAtTS and this refuses, sparing bytes
	// the PUT has already acknowledged as durable. An unconditional unlink
	// had no interlock at all -- the scan's reads are from before the PUT
	// committed, so it would remove a now-live payload.
	//
	// The implementation must compare and unlink atomically with respect to
	// the local writer; a read-then-delete would reopen the same window it
	// exists to close.
	DeleteChunkBlobIfUnchanged(
		ctx context.Context, sha [chunkBlobSHA256Bytes]byte, writtenAtTS uint64,
	) (bool, error)
}

ChunkBlobOrphanLocalStore is the node-local half.

type ChunkBlobOrphanObserver 

type ChunkBlobOrphanObserver interface {
	ObserveChunkBlobOrphan(verdict ChunkBlobOrphanVerdict, reason string)
}

ChunkBlobOrphanObserver receives per-blob outcomes.

type ChunkBlobOrphanScanner 

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

ChunkBlobOrphanScanner reclaims local blobs no replicated state refers to.

func NewChunkBlobOrphanScanner 

func NewChunkBlobOrphanScanner(opts ChunkBlobOrphanScannerOptions) (*ChunkBlobOrphanScanner, error)

func (*ChunkBlobOrphanScanner) MarkedOrphans 

func (s *ChunkBlobOrphanScanner) MarkedOrphans() [][chunkBlobSHA256Bytes]byte

MarkedOrphans returns the SHAs currently held under a sweep mark. Exposed for tests and operator tooling; the set is per-process.

func (*ChunkBlobOrphanScanner) Run 

func (s *ChunkBlobOrphanScanner) Run(ctx context.Context) error

Run scans on the configured interval until ctx is cancelled.

func (*ChunkBlobOrphanScanner) ScanOnce 

func (s *ChunkBlobOrphanScanner) ScanOnce(ctx context.Context) error

ScanOnce runs one pass over this node's local chunkblobs.

type ChunkBlobOrphanScannerOptions 

type ChunkBlobOrphanScannerOptions struct {
	Store       ChunkBlobOrphanStore
	Local       ChunkBlobOrphanLocalStore
	GracePeriod time.Duration
	Interval    time.Duration
	NowTS       func() uint64
	Observer    ChunkBlobOrphanObserver
	Logger      *slog.Logger
}

ChunkBlobOrphanScannerOptions configures NewChunkBlobOrphanScanner.

type ChunkBlobOrphanStore 

type ChunkBlobOrphanStore interface {
	ReadChunkRefRC(ctx context.Context, sha [chunkBlobSHA256Bytes]byte) ([]byte, bool, error)
	// GCQueueEntryExists reports whether any queue entry references
	// this SHA. The scan only needs presence, not the timestamp, so
	// this stays a cheaper question than a range scan.
	GCQueueEntryExists(ctx context.Context, sha [chunkBlobSHA256Bytes]byte) (bool, error)
}

ChunkBlobOrphanStore is the replicated state the scan consults.

type ChunkBlobOrphanVerdict 

type ChunkBlobOrphanVerdict int

ChunkBlobOrphanVerdict is what the scan may do with one local blob. 

const (
	// OrphanKeep leaves the blob alone.
	OrphanKeep ChunkBlobOrphanVerdict = iota
	// OrphanReclaim deletes the local blob. There is no Raft phase:
	// by construction an orphan has no replicated state referring to
	// it, which is exactly what makes it an orphan.
	OrphanReclaim
)

func (ChunkBlobOrphanVerdict) String 

func (v ChunkBlobOrphanVerdict) String() string

type ChunkBlobSweepDecision 

type ChunkBlobSweepDecision struct {
	Verdict ChunkBlobSweepVerdict
	Reason  string
}

ChunkBlobSweepDecision is a verdict plus the reason behind it, so a sweeper can log and meter why a blob was or was not reclaimed without re-deriving the logic.

func ClassifyChunkBlobSweep 

func ClassifyChunkBlobSweep(entryTS uint64, rcValue []byte, rcFound bool) ChunkBlobSweepDecision

ClassifyChunkBlobSweep decides the fate of the queue entry stamped entryTS for this SHA.

rcValue is the raw stored reference-count value, and rcFound reports whether the key existed. The raw bytes are taken rather than a decoded record so an undecodable value is distinguishable from an absent one: the first means the sweeper cannot reason about reachability and must decline, while the second is a legitimate "never referenced" state.

type ChunkBlobSweepObserver 

type ChunkBlobSweepObserver interface {
	ObserveChunkBlobSweep(verdict ChunkBlobSweepVerdict, reason string)
	ObserveChunkBlobSweepRaceLost()
}

ChunkBlobSweepObserver receives per-entry outcomes.

type ChunkBlobSweepStore 

type ChunkBlobSweepStore interface {
	// ScanGCQueue returns up to limit queue entries from [startKey, endKey),
	// plus the key to resume from when more remain.
	//
	// Paged, because the eligible range is unbounded: an outage or a large
	// object-deletion workload can leave millions of expired entries, and
	// materialising them all before the sweeper reclaims even the first one
	// can OOM the process -- leaving the backlog permanently untouched,
	// which is the opposite of what a GC pass is for.
	//
	// Each PAGE must still be all-or-error. A short page is fine -- the
	// remaining entries are picked up by the continuation key or the next
	// pass -- but a page that silently truncated while reporting success
	// would hide a persistent backlog.
	//
	// nextStartKey is nil when the range is exhausted.
	ScanGCQueue(
		ctx context.Context, startKey, endKey []byte, limit int,
	) (entries []ChunkBlobGCQueueEntry, nextStartKey []byte, err error)

	// ReadChunkRefRC returns the raw reference-count value and
	// whether the key exists. Raw bytes, so the classifier can tell
	// an undecodable record from an absent one.
	ReadChunkRefRC(ctx context.Context, sha [chunkBlobSHA256Bytes]byte) ([]byte, bool, error)

	// DeleteGCQueueEntryIfUnreferenced deletes the queue entry AND the
	// zero-count reference record, in ONE txn, only if the entry still
	// exists and the count is still zero; ErrQueueEntryChanged otherwise.
	// Concurrent sweepers serialise here on the queue key's write-write
	// conflict.
	//
	// The RC record goes with the entry because nothing else ever removed
	// it: the planner persists a count-zero record when the last reference
	// drops, and reclamation deleted only the queue entry and the local
	// payload. A workload that creates and deletes unique chunks therefore
	// left one permanent Raft-replicated key per content hash, so the live
	// MVCC state and every snapshot grew without bound despite blob GC
	// working correctly.
	//
	// Atomically with the entry, not as a second txn: a crash between two
	// txns would leave either a queue entry for a blob with no RC record,
	// or the same unbounded leak this removes.
	DeleteGCQueueEntryIfUnreferenced(ctx context.Context, entry ChunkBlobGCQueueEntry) error

	// DeleteGCQueueEntry deletes the entry unconditionally. Used only
	// for the §3.5(c) stale-entry path, where the blob is explicitly
	// being left in place.
	DeleteGCQueueEntry(ctx context.Context, entry ChunkBlobGCQueueEntry) error

	// ClearGCQueueMetadata deletes the entry AND clears QueuedAtTS on the
	// live RC record in ONE transaction.
	//
	// Atomic because the two halves are only consistent together. Between
	// them the record would claim a queue entry that does not exist, and a
	// dereference landing in that window takes
	// PlanChunkRefRCMutations' already-queued branch: it preserves the
	// stale timestamp and writes no replacement queue key, stranding the
	// blob outside the queue permanently.
	ClearGCQueueMetadata(ctx context.Context, entry ChunkBlobGCQueueEntry) error
}

ChunkBlobSweepStore is the replicated half: the GC queue and the reference counts, both read and written through Raft.

type ChunkBlobSweepVerdict 

type ChunkBlobSweepVerdict int

ChunkBlobSweepVerdict is what a sweeper may do with one queue entry. 

const (
	// SweepSkip leaves both the queue entry and the blob alone. Used
	// when the record cannot be trusted, so the sweeper declines
	// rather than guessing.
	SweepSkip ChunkBlobSweepVerdict = iota

	// SweepReclaim deletes the queue entry (conditionally, through
	// Raft) and then the local chunkblob. Only reachable when the
	// reference count is zero AND the record still points at THIS
	// queue entry.
	SweepReclaim

	// SweepDropQueueEntryOnly deletes the queue entry and leaves the
	// chunkblob in place. This is §3.5(c): the entry is stale, either
	// because the blob is referenced again or because a newer entry
	// supersedes this one.
	SweepDropQueueEntryOnly

	// SweepClearStaleQueueMetadata deletes the queue entry AND clears
	// QueuedAtTS on the live RC record, atomically.
	//
	// Dropping only the key leaves the record claiming a queue entry that
	// no longer exists, and PlanChunkRefRCMutations reads that claim: when
	// the last reference is later removed it takes the `existing.Queued()`
	// branch, preserves the stale timestamp, and inserts NO replacement
	// queue key. The blob then never re-enters the queue, so the sweeper
	// never sees it again and only the orphan scan can reclaim it -- on its
	// own independently configurable interval rather than the GC grace
	// period this keyspace exists to enforce.
	//
	// Distinct from SweepDropQueueEntryOnly because a SUPERSEDED entry must
	// keep its timestamp: there the record points at a newer entry that is
	// still serving its own grace window.
	SweepClearStaleQueueMetadata
)

func (ChunkBlobSweepVerdict) String 

func (v ChunkBlobSweepVerdict) String() string

type ChunkBlobSweeper 

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

ChunkBlobSweeper reclaims chunkblobs whose references are gone.

func NewChunkBlobSweeper 

func NewChunkBlobSweeper(opts ChunkBlobSweeperOptions) (*ChunkBlobSweeper, error)

func (*ChunkBlobSweeper) Run 

func (s *ChunkBlobSweeper) Run(ctx context.Context) error

Run sweeps on the configured interval until ctx is cancelled.

A failing pass is retried next tick rather than tearing the loop down: the queue is durable, so a transient store error costs a delay, never a lost reclaim.

func (*ChunkBlobSweeper) SweepOnce 

func (s *ChunkBlobSweeper) SweepOnce(ctx context.Context) error

SweepOnce runs one pass over the entries whose grace window has elapsed.

type ChunkBlobSweeperOptions 

type ChunkBlobSweeperOptions struct {
	Store       ChunkBlobSweepStore
	Local       ChunkBlobLocalStore
	GracePeriod time.Duration
	Interval    time.Duration
	NowTS       func() uint64
	Observer    ChunkBlobSweepObserver
	Logger      *slog.Logger
	// PageSize bounds one queue-scan page; zero uses
	// DefaultChunkBlobGCPageSize.
	PageSize int
}

ChunkBlobSweeperOptions configures NewChunkBlobSweeper.

NowTS returns the current HLC timestamp — not a wall clock — because the queue keys are stamped with commit timestamps and the grace boundary must be computed in that domain.

type ChunkRefDelta 

type ChunkRefDelta struct {
	ContentSHA256 [chunkBlobSHA256Bytes]byte
	Added         uint64
	Removed       uint64
}

ChunkRefDelta is one SHA's reference-count change within a txn.

Added and Removed are counted separately rather than pre-netted so a txn that both adds and removes references to the same content — a part rewritten to identical bytes — is expressed honestly and nets to zero here instead of at the call site.

type ChunkRefPeer 

type ChunkRefPeer struct {
	NodeID  string
	Address string
}

ChunkRefPeer records a peer that durably acknowledged the chunkblob before its reference became reachable through Raft.

type ChunkRefRC 

type ChunkRefRC struct {
	Count      uint64
	QueuedAtTS uint64
}

ChunkRefRC is the reference-count record for one content hash.

QueuedAtTS carries the timestamp of this SHA's GC-queue entry, or zero when it has none. It is part of the VALUE because §3.5 requires a txn that re-references a SHA to delete the queue entry atomically with incrementing the count — and the queue key embeds the eligibility timestamp, which that txn has no other way to learn. Without it the re-referencing txn cannot name the key it must delete, leaving a stale entry that points the sweeper at a blob which is once again live.

func DecodeChunkRefRC 

func DecodeChunkRefRC(value []byte) (ChunkRefRC, bool)

DecodeChunkRefRC decodes a reference-count record. A missing key and an explicit zero count are equivalent to the caller — both mean "no live reference" — but a malformed value is not, so it fails closed rather than defaulting to zero and making a live blob look collectable.

func (ChunkRefRC) Queued 

func (r ChunkRefRC) Queued() bool

Queued reports whether this SHA currently has a GC-queue entry.

type ChunkRefRCMutation 

type ChunkRefRCMutation struct {
	Key    []byte
	Value  []byte
	Delete bool
}

ChunkRefRCMutation is one key the txn must write or delete.

Value is nil for a delete. Callers apply these alongside their own chunkref mutations in the SAME txn; applying them separately would break the linearisation point the design depends on.

func PlanChunkRefRCMutations 

func PlanChunkRefRCMutations(
	deltas []ChunkRefDelta,
	current map[[chunkBlobSHA256Bytes]byte]ChunkRefRC,
	commitTS uint64,
) ([]ChunkRefRCMutation, error)

PlanChunkRefRCMutations computes the reference-count and GC-queue mutations a txn must carry for the given deltas.

current maps a SHA to its reference-count record as of the txn's read timestamp; a SHA absent from the map is treated as count zero with no queue entry, which is the correct reading of a missing key.

commitTS is the txn's commit timestamp, used as the eligibility timestamp for any SHA this txn drives to zero. It must be an HLC commit timestamp — see ChunkBlobGCQueueKey.

type ChunkRefValue 

type ChunkRefValue struct {
	ContentSHA256 [chunkBlobSHA256Bytes]byte
	Size          uint64
	SourcePeer    string
	ReplicaPeers  []ChunkRefPeer
}

ChunkRefValue is the stored value for a !s3|chunkref| entry. It points from an object part chunk to the content-addressed !s3|chunkblob| payload.

func DecodeChunkRefValue 

func DecodeChunkRefValue(value []byte) (ChunkRefValue, bool)

DecodeChunkRefValue decodes a chunk reference value. It returns ok=false for unknown versions, truncated values, or trailing-length mismatches.

type LocalChunkBlob 

type LocalChunkBlob struct {
	ContentSHA256 [chunkBlobSHA256Bytes]byte
	// WrittenAtTS is when this node wrote the payload, as an HLC
	// timestamp. It is the blob's own age, not a reference timestamp:
	// the whole point of the grace gate is that a young blob with no
	// RC entry is an upload in progress.
	WrittenAtTS uint64
}

LocalChunkBlob is one blob found on local disk.

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.
⌂ 换网址