diff --git a/docs/design/2026_04_25_partial_s3_raft_blob_offload.md b/docs/design/2026_04_25_partial_s3_raft_blob_offload.md index 7a8f6c257..6159c591e 100644 --- a/docs/design/2026_04_25_partial_s3_raft_blob_offload.md +++ b/docs/design/2026_04_25_partial_s3_raft_blob_offload.md @@ -633,7 +633,7 @@ capability-advertising peers exists. | M0 | Spike: prove the chunkref + chunkblob keyspaces under a feature flag with 1 % traffic. Measure local Pebble write amp & blob fetch latency. | Implemented. | | M1 | PUT path emits chunkrefs through Raft; chunkblob writes go directly to local Pebble. **`FetchChunkBlob` and `PushChunkBlob` RPCs ship in this milestone** because both M1 PUT (semi-synchronous push) and M1 GET (proxy-on-miss) depend on them — without them M1 GET could only serve local-hit or 503. | Implemented behind the GC-readiness gate. | | M2 | Async fetch worker pool for follower apply (catch-up after a long absence). Independent of M1's synchronous `FetchChunkBlob` use on the GET path. SHA verification + retry from alternate peer on mismatch. | Partially implemented: fetch/backfill helpers exist; production enablement remains gated by M3. | -| M3 | Reference-count + grace-period GC (the queue-based scheme in §3.5). | Open. | +| M3 | Reference-count + grace-period GC (the queue-based scheme in §3.5). | In progress: the §3.5 keyspaces (`!s3\|chunkref-rc\|`, `!s3\|chunkblob-gc-queue\|`) and their codecs are implemented in `internal/s3keys/chunkblob_gc.go`. Queue timestamps are HLC commit timestamps, and `ChunkBlobGCGraceBoundary` converts a wall-clock grace period into that domain so a sweeper never mixes it with Unix nanoseconds. The reference-count mutation planner (`PlanChunkRefRCMutations`) is implemented: it encodes the atomic-pair semantics — increment, decrement-to-zero-and-queue, re-reference-and-dequeue, underflow-fails-closed — as a pure function the chunkref txn will call. The sweeper's decision layer (`ClassifyChunkBlobSweep`) is also implemented: it maps one queue entry plus the reference-count record to reclaim / drop-entry-only / skip, which is the classification the conditional Raft delete in 3(b)(i) is built from. The sweeper loop itself (`ChunkBlobSweeper`) is implemented over narrow injected interfaces, enforcing the 3(b) phase ordering — Raft conditional delete first, local unlink second — and treating a lost conditional delete as a skip rather than a failure. The orphan scan (`ChunkBlobOrphanScanner`) is also implemented, covering both documented sources — a sweeper that crashed between phases, and a PUT that aborted before dispatching its chunkref — gated on the blob's own age so an in-flight upload is never mistaken for an abort. What remains is wiring: the planner into the chunkref txn, and the real store behind the sweeper's and scanner's interfaces. | | M4 | Migrator: rewrite legacy `BlobKey` data in the background. Off by default until M0-M3 burn in for 30 days in production. | Open. | Acceptance criteria for M3 (the milestone that flips `ELASTICKV_S3_BLOB_OFFLOAD=true` by default): diff --git a/internal/s3keys/chunkblob_gc.go b/internal/s3keys/chunkblob_gc.go new file mode 100644 index 000000000..2740475d8 --- /dev/null +++ b/internal/s3keys/chunkblob_gc.go @@ -0,0 +1,220 @@ +package s3keys + +import ( + "bytes" + "encoding/binary" + "encoding/hex" + "time" +) + +// Reference counting and GC eligibility for content-addressed +// chunkblobs (design §3.5). +// +// Two keyspaces, both Raft-replicated: +// +// !s3|chunkref-rc| -> ChunkRefRC +// !s3|chunkblob-gc-queue|| -> 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). +const ( + ChunkRefRCPrefix = "!s3|chunkref-rc|" + ChunkBlobGCQueuePrefix = "!s3|chunkblob-gc-queue|" + + // hlcLogicalBits mirrors kv.HLCLogicalBits. It is duplicated + // rather than imported because internal/s3keys cannot import kv + // without a cycle (kv -> distribution -> s3keys). The external + // test asserts the two stay equal, so the duplication cannot + // drift silently. + hlcLogicalBits = 16 + + // chunkRefRCValueBytes is the fixed width of an encoded + // ChunkRefRC: the count followed by the queue timestamp. + chunkRefRCValueBytes = 2 * u64Bytes + + // chunkBlobGCQueueSeparator delimits the timestamp from the SHA. + // It must sort below every hex digit so the scan-end key built + // from a bare timestamp excludes that timestamp's own entries + // only when intended; '|' (0x7C) is above hex, so the separator + // is chosen to match the surrounding key grammar and the end key + // is built explicitly rather than by string concatenation. + chunkBlobGCQueueSeparator = '|' +) + +var ( + chunkRefRCPrefixBytes = []byte(ChunkRefRCPrefix) + chunkBlobGCQueuePrefixBytes = []byte(ChunkBlobGCQueuePrefix) +) + +// ChunkRefRCKey builds the reference-count key for a content hash. +func ChunkRefRCKey(contentSHA256 [chunkBlobSHA256Bytes]byte) []byte { + out := make([]byte, 0, len(ChunkRefRCPrefix)+chunkBlobSHA256HexBytes) + out = append(out, chunkRefRCPrefixBytes...) + return hex.AppendEncode(out, contentSHA256[:]) +} + +// ParseChunkRefRCKey decodes a reference-count key. +func ParseChunkRefRCKey(key []byte) ([chunkBlobSHA256Bytes]byte, bool) { + var sha [chunkBlobSHA256Bytes]byte + if !bytes.HasPrefix(key, chunkRefRCPrefixBytes) { + return sha, false + } + return decodeSHAHex(key[len(chunkRefRCPrefixBytes):], sha) +} + +// 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. +type ChunkRefRC struct { + Count uint64 + QueuedAtTS uint64 +} + +// Queued reports whether this SHA currently has a GC-queue entry. +func (r ChunkRefRC) Queued() bool { return r.QueuedAtTS != 0 } + +// EncodeChunkRefRC encodes a reference-count record. +func EncodeChunkRefRC(rc ChunkRefRC) []byte { + out := make([]byte, 0, chunkRefRCValueBytes) + out = binary.BigEndian.AppendUint64(out, rc.Count) + return binary.BigEndian.AppendUint64(out, rc.QueuedAtTS) +} + +// 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 DecodeChunkRefRC(value []byte) (ChunkRefRC, bool) { + if len(value) != chunkRefRCValueBytes { + return ChunkRefRC{}, false + } + return ChunkRefRC{ + Count: binary.BigEndian.Uint64(value[:u64Bytes]), + QueuedAtTS: binary.BigEndian.Uint64(value[u64Bytes:]), + }, true +} + +// 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 ChunkBlobGCQueueKey(commitTS uint64, contentSHA256 [chunkBlobSHA256Bytes]byte) []byte { + out := make([]byte, 0, + len(ChunkBlobGCQueuePrefix)+u64Bytes+1+chunkBlobSHA256HexBytes) + out = append(out, chunkBlobGCQueuePrefixBytes...) + out = binary.BigEndian.AppendUint64(out, commitTS) + out = append(out, chunkBlobGCQueueSeparator) + return hex.AppendEncode(out, contentSHA256[:]) +} + +// ParseChunkBlobGCQueueKey decodes an eligibility-queue key into the +// timestamp at which the blob became unreferenced and its content hash. +func ParseChunkBlobGCQueueKey(key []byte) (uint64, [chunkBlobSHA256Bytes]byte, bool) { + var sha [chunkBlobSHA256Bytes]byte + if !bytes.HasPrefix(key, chunkBlobGCQueuePrefixBytes) { + return 0, sha, false + } + rest := key[len(chunkBlobGCQueuePrefixBytes):] + if len(rest) != u64Bytes+1+chunkBlobSHA256HexBytes { + return 0, sha, false + } + if rest[u64Bytes] != chunkBlobGCQueueSeparator { + return 0, sha, false + } + commitTS := binary.BigEndian.Uint64(rest[:u64Bytes]) + sha, ok := decodeSHAHex(rest[u64Bytes+1:], sha) + if !ok { + return 0, sha, false + } + return commitTS, sha, true +} + +// ChunkBlobGCQueueScanStart is the inclusive lower bound for a sweeper +// scan: the start of the whole queue. +func ChunkBlobGCQueueScanStart() []byte { + return append([]byte(nil), chunkBlobGCQueuePrefixBytes...) +} + +// 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 ChunkBlobGCQueueScanEnd(boundaryTS uint64) []byte { + out := make([]byte, 0, len(ChunkBlobGCQueuePrefix)+u64Bytes) + out = append(out, chunkBlobGCQueuePrefixBytes...) + return binary.BigEndian.AppendUint64(out, boundaryTS) +} + +// decodeSHAHex decodes a lowercase hex SHA-256 of the exact expected +// width. Length is checked before decoding so a short or padded key +// cannot decode into a partially-populated digest. +func decodeSHAHex(encoded []byte, sha [chunkBlobSHA256Bytes]byte) ([chunkBlobSHA256Bytes]byte, bool) { + if len(encoded) != chunkBlobSHA256HexBytes { + return sha, false + } + if _, err := hex.Decode(sha[:], encoded); err != nil { + return sha, false + } + return sha, true +} + +// 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 ChunkBlobGCGraceBoundary(nowTS uint64, grace time.Duration) uint64 { + physicalMs := nowTS >> hlcLogicalBits + graceMs := uint64(0) + if ms := grace.Milliseconds(); ms > 0 { + // Guarded above zero, so the conversion cannot go negative. + graceMs = uint64(ms) + } + if graceMs >= physicalMs { + return 0 + } + return (physicalMs - graceMs) << hlcLogicalBits +} diff --git a/internal/s3keys/chunkblob_gc_test.go b/internal/s3keys/chunkblob_gc_test.go new file mode 100644 index 000000000..cdda41a2b --- /dev/null +++ b/internal/s3keys/chunkblob_gc_test.go @@ -0,0 +1,330 @@ +package s3keys_test + +import ( + "bytes" + "crypto/sha256" + "fmt" + "math" + "sort" + "testing" + "time" + + "github.com/bootjp/elastickv/internal/s3keys" + "github.com/bootjp/elastickv/kv" + "github.com/stretchr/testify/require" +) + +func testSHA(seed string) [32]byte { + return sha256.Sum256([]byte(seed)) +} + +func TestChunkRefRCKeyRoundTrip(t *testing.T) { + t.Parallel() + + sha := testSHA("payload") + key := s3keys.ChunkRefRCKey(sha) + require.True(t, bytes.HasPrefix(key, []byte(s3keys.ChunkRefRCPrefix))) + + got, ok := s3keys.ParseChunkRefRCKey(key) + require.True(t, ok) + require.Equal(t, sha, got) +} + +func TestParseChunkRefRCKeyRejectsMalformed(t *testing.T) { + t.Parallel() + + sha := testSHA("payload") + valid := s3keys.ChunkRefRCKey(sha) + + tests := []struct { + name string + key []byte + }{ + {"empty", nil}, + {"wrong prefix", []byte("!s3|chunkblob|" + string(valid[len(s3keys.ChunkRefRCPrefix):]))}, + {"truncated hex", valid[:len(valid)-2]}, + {"padded hex", append(append([]byte(nil), valid...), 'a', 'b')}, + {"non hex", append(append([]byte(nil), valid[:len(valid)-2]...), 'z', 'z')}, + } + for _, tc := range tests { + t.Run(tc.name, func(t *testing.T) { + t.Parallel() + _, ok := s3keys.ParseChunkRefRCKey(tc.key) + require.False(t, ok) + }) + } +} + +// TestChunkRefRCValueFailsClosedOnMalformedValue pins that a corrupt +// counter is not read as zero. Zero means "no live reference", so +// defaulting to it would make a live blob look collectable. +func TestChunkRefRCValueFailsClosedOnMalformedValue(t *testing.T) { + t.Parallel() + + got, ok := s3keys.DecodeChunkRefRC(s3keys.EncodeChunkRefRC(s3keys.ChunkRefRC{Count: 7})) + require.True(t, ok) + require.Equal(t, uint64(7), got.Count) + require.False(t, got.Queued()) + + for _, bad := range [][]byte{nil, {}, {0x01}, make([]byte, 8), make([]byte, 15), make([]byte, 17)} { + _, ok := s3keys.DecodeChunkRefRC(bad) + require.False(t, ok, "a malformed record must not decode to a zero count") + } +} + +// TestChunkRefRCCarriesTheQueueTimestamp pins the field that makes the +// §3.5 re-reference path implementable. +// +// When a SHA is referenced again after its count reached zero, the +// same txn must delete the existing GC-queue entry. That key embeds the +// eligibility timestamp, which the re-referencing txn has no other way +// to learn — so the count record has to carry it. Without it the txn +// cannot name the key it must delete, and a stale queue entry would +// point the sweeper at a blob that is live again. +func TestChunkRefRCCarriesTheQueueTimestamp(t *testing.T) { + t.Parallel() + + sha := testSHA("payload") + const queuedAt = uint64(1_700_000_000_000_000_000) + + // Count dropped to zero: the txn records when, and queues. + zeroed := s3keys.ChunkRefRC{Count: 0, QueuedAtTS: queuedAt} + decoded, ok := s3keys.DecodeChunkRefRC(s3keys.EncodeChunkRefRC(zeroed)) + require.True(t, ok) + require.Zero(t, decoded.Count) + require.True(t, decoded.Queued()) + + // A re-referencing txn can now reconstruct the exact queue key it + // has to delete. + require.Equal(t, + s3keys.ChunkBlobGCQueueKey(queuedAt, sha), + s3keys.ChunkBlobGCQueueKey(decoded.QueuedAtTS, sha), + "the recorded timestamp must reproduce the queue key exactly") + + // Re-referenced: count back above zero, no queue entry. + live := s3keys.ChunkRefRC{Count: 1} + decoded, ok = s3keys.DecodeChunkRefRC(s3keys.EncodeChunkRefRC(live)) + require.True(t, ok) + require.Equal(t, uint64(1), decoded.Count) + require.False(t, decoded.Queued(), + "a live SHA must not claim a queue entry") +} + +func TestChunkBlobGCQueueKeyRoundTrip(t *testing.T) { + t.Parallel() + + sha := testSHA("payload") + const ts = uint64(1_700_000_000_123_456_789) + + key := s3keys.ChunkBlobGCQueueKey(ts, sha) + require.True(t, bytes.HasPrefix(key, []byte(s3keys.ChunkBlobGCQueuePrefix))) + + gotTS, gotSHA, ok := s3keys.ParseChunkBlobGCQueueKey(key) + require.True(t, ok) + require.Equal(t, ts, gotTS) + require.Equal(t, sha, gotSHA) +} + +// TestChunkBlobGCQueueSortsByEligibilityTime is the property the whole +// grace window rests on. A decimal or variable-width timestamp would +// order 9 after 10 and make the boundary scan return the wrong set. +func TestChunkBlobGCQueueSortsByEligibilityTime(t *testing.T) { + t.Parallel() + + timestamps := []uint64{1, 9, 10, 99, 100, 1 << 32, math.MaxUint64 - 1, math.MaxUint64} + keys := make([][]byte, 0, len(timestamps)) + for i, ts := range timestamps { + keys = append(keys, s3keys.ChunkBlobGCQueueKey(ts, testSHA(fmt.Sprintf("blob-%d", i)))) + } + + shuffled := append([][]byte(nil), keys...) + sort.Slice(shuffled, func(i, j int) bool { return bytes.Compare(shuffled[i], shuffled[j]) < 0 }) + + for i, key := range shuffled { + gotTS, _, ok := s3keys.ParseChunkBlobGCQueueKey(key) + require.True(t, ok) + require.Equal(t, timestamps[i], gotTS, + "byte order must match eligibility-time order at position %d", i) + } +} + +// TestChunkBlobGCQueueScanEndIsExclusive pins the grace boundary. +// Callers pass now-grace; an entry stamped exactly at the boundary has +// not yet served the full window and must be excluded. +func TestChunkBlobGCQueueScanEndIsExclusive(t *testing.T) { + t.Parallel() + + const boundary = uint64(1_000) + sha := testSHA("payload") + start := s3keys.ChunkBlobGCQueueScanStart() + end := s3keys.ChunkBlobGCQueueScanEnd(boundary) + + inWindow := s3keys.ChunkBlobGCQueueKey(boundary-1, sha) + atBoundary := s3keys.ChunkBlobGCQueueKey(boundary, sha) + afterBoundary := s3keys.ChunkBlobGCQueueKey(boundary+1, sha) + + require.Negative(t, bytes.Compare(start, inWindow)) + require.Negative(t, bytes.Compare(inWindow, end), + "an entry older than the boundary must fall inside the scan") + require.GreaterOrEqual(t, bytes.Compare(atBoundary, end), 0, + "an entry exactly at the boundary has not served the full grace period") + require.Positive(t, bytes.Compare(afterBoundary, end)) +} + +// TestChunkBlobGCQueueScanStartCoversTheWholeQueue guards the lower +// bound: the earliest possible entry must sort at or after the scan +// start rather than below it. +func TestChunkBlobGCQueueScanStartCoversTheWholeQueue(t *testing.T) { + t.Parallel() + + start := s3keys.ChunkBlobGCQueueScanStart() + earliest := s3keys.ChunkBlobGCQueueKey(1, testSHA("earliest")) + require.LessOrEqual(t, bytes.Compare(start, earliest), 0) + require.Negative(t, bytes.Compare(earliest, s3keys.ChunkBlobGCQueueScanEnd(2))) +} + +// TestHLCLogicalBitsMatchesKV pins the duplicated constant. +// internal/s3keys cannot import kv (kv -> distribution -> s3keys), so +// the shift width is mirrored locally; this external test closes the +// loop so the two cannot drift apart silently and leave the grace +// boundary computing against the wrong field width. +func TestHLCLogicalBitsMatchesKV(t *testing.T) { + t.Parallel() + + // Derived rather than read directly: a commit timestamp whose + // physical half is 1 ms must shift down to exactly 1. + oneMs := uint64(1) << kv.HLCLogicalBits + require.Equal(t, uint64(1), + s3keys.ChunkBlobGCGraceBoundary(oneMs, 0)>>kv.HLCLogicalBits, + "s3keys' mirrored HLC logical width must match kv.HLCLogicalBits") +} + +// TestChunkBlobGCGraceBoundaryWorksInTheHLCDomain pins that the grace +// boundary is computed against HLC commit timestamps, not Unix +// nanoseconds. Subtracting a duration means subtracting milliseconds +// from the PHYSICAL half; treating the whole value as nanoseconds +// would be off by orders of magnitude and either sweep everything +// immediately or never sweep at all. +func TestChunkBlobGCGraceBoundaryWorksInTheHLCDomain(t *testing.T) { + t.Parallel() + + nowMs := uint64(1_700_000_000_000) + nowTS := nowMs << kv.HLCLogicalBits + + boundary := s3keys.ChunkBlobGCGraceBoundary(nowTS, time.Hour) + require.Equal(t, (nowMs-3_600_000)<= boundaryTS { + return ChunkBlobOrphanDecision{Verdict: OrphanKeep, Reason: OrphanReasonWithinGrace} + } + + if !rcFound { + // §3.5 case two: the PUT never dispatched its chunkref, and + // the blob is old enough that it cannot still be in flight. + return ChunkBlobOrphanDecision{ + Verdict: OrphanReclaim, + Reason: OrphanReasonNoReferenceRecord, + } + } + + rc, ok := DecodeChunkRefRC(rcValue) + if !ok { + return ChunkBlobOrphanDecision{Verdict: OrphanKeep, Reason: OrphanReasonRecordUnreadable} + } + if rc.Count > 0 { + return ChunkBlobOrphanDecision{Verdict: OrphanKeep, Reason: OrphanReasonReferenced} + } + if queueFound { + // The sweeper owns this one: it has a live queue entry serving + // its grace window, and reclaiming here would bypass the + // conditional-delete interlock the sweeper relies on. + return ChunkBlobOrphanDecision{Verdict: OrphanKeep, Reason: OrphanReasonQueueOwnsIt} + } + // §3.5 case one: count zero and no queue entry means a sweeper + // removed the entry through Raft and then died before the local + // unlink. + return ChunkBlobOrphanDecision{Verdict: OrphanReclaim, Reason: OrphanReasonSweeperCrashed} +} + +// ChunkBlobOrphanStore is the replicated state the scan consults. +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) +} + +// ChunkBlobOrphanLocalStore is the node-local half. +type ChunkBlobOrphanLocalStore interface { + // ListLocalChunkBlobs enumerates this node's chunkblobs. + ListLocalChunkBlobs(ctx context.Context) ([]LocalChunkBlob, error) + DeleteChunkBlob(ctx context.Context, sha [chunkBlobSHA256Bytes]byte) error +} + +// ChunkBlobOrphanObserver receives per-blob outcomes. +type ChunkBlobOrphanObserver interface { + ObserveChunkBlobOrphan(verdict ChunkBlobOrphanVerdict, reason string) +} + +type nopOrphanObserver struct{} + +func (nopOrphanObserver) ObserveChunkBlobOrphan(ChunkBlobOrphanVerdict, string) {} + +// ChunkBlobOrphanScanner reclaims local blobs no replicated state +// refers to. +type ChunkBlobOrphanScanner struct { + store ChunkBlobOrphanStore + local ChunkBlobOrphanLocalStore + grace time.Duration + interval time.Duration + nowTS func() uint64 + observer ChunkBlobOrphanObserver + logger *slog.Logger +} + +// ChunkBlobOrphanScannerOptions configures NewChunkBlobOrphanScanner. +type ChunkBlobOrphanScannerOptions struct { + Store ChunkBlobOrphanStore + Local ChunkBlobOrphanLocalStore + GracePeriod time.Duration + Interval time.Duration + NowTS func() uint64 + Observer ChunkBlobOrphanObserver + Logger *slog.Logger +} + +func NewChunkBlobOrphanScanner(opts ChunkBlobOrphanScannerOptions) (*ChunkBlobOrphanScanner, error) { + switch { + case opts.Store == nil: + return nil, errors.Wrap(ErrInvalidChunkRefPlan, "orphan scanner requires a replicated store") + case opts.Local == nil: + return nil, errors.Wrap(ErrInvalidChunkRefPlan, "orphan scanner requires a local blob store") + case opts.NowTS == nil: + return nil, errors.Wrap(ErrInvalidChunkRefPlan, "orphan scanner requires an HLC clock") + } + timing := resolveGCLoopTiming( + opts.GracePeriod, opts.Interval, + DefaultChunkBlobOrphanGracePeriod, DefaultChunkBlobOrphanScanInterval, opts.Logger) + observer := opts.Observer + if observer == nil { + observer = nopOrphanObserver{} + } + return &ChunkBlobOrphanScanner{ + store: opts.Store, + local: opts.Local, + grace: timing.grace, + interval: timing.interval, + nowTS: opts.NowTS, + observer: observer, + logger: timing.logger, + }, nil +} + +// Run scans on the configured interval until ctx is cancelled. +func (s *ChunkBlobOrphanScanner) Run(ctx context.Context) error { + if ctx == nil { + return errors.Wrap(ErrInvalidChunkRefPlan, "orphan scanner context is required") + } + ticker := time.NewTicker(s.interval) + defer ticker.Stop() + for { + if sweepCancelled(ctx) { + return nil + } + if err := s.ScanOnce(ctx); err != nil && !sweepCancelled(ctx) { + s.logger.WarnContext(ctx, "chunkblob orphan scan failed", + slog.String("error", err.Error())) + } + select { + case <-ctx.Done(): + return nil + case <-ticker.C: + } + } +} + +// ScanOnce runs one pass over this node's local chunkblobs. +func (s *ChunkBlobOrphanScanner) ScanOnce(ctx context.Context) error { + boundary := ChunkBlobGCGraceBoundary(s.nowTS(), s.grace) + if boundary == 0 { + // No blob can be older than the grace window yet, so every + // local blob could still belong to an upload in progress. + return nil + } + blobs, err := s.local.ListLocalChunkBlobs(ctx) + if err != nil { + return errors.Wrap(err, "orphan scan: list local chunkblobs") + } + for _, blob := range blobs { + if sweepCancelled(ctx) { + break + } + if err := s.scanBlob(ctx, blob, boundary); err != nil { + return err + } + } + return nil +} + +func (s *ChunkBlobOrphanScanner) scanBlob( + ctx context.Context, blob LocalChunkBlob, boundary uint64, +) error { + // The age gate needs no replicated reads, so check it before + // paying for them: on a healthy node most blobs are referenced and + // this keeps the scan's cost proportional to real orphans. + if blob.WrittenAtTS == 0 || blob.WrittenAtTS >= boundary { + s.observer.ObserveChunkBlobOrphan(OrphanKeep, OrphanReasonWithinGrace) + return nil + } + + rcValue, rcFound, err := s.store.ReadChunkRefRC(ctx, blob.ContentSHA256) + if err != nil { + return errors.Wrapf(err, "orphan scan: read reference count for %x", blob.ContentSHA256[:4]) + } + queueFound := false + if rcFound { + // Only consulted when a record exists: with no record at all + // the §3.5 criterion is already satisfied and the extra read + // would be wasted. + queueFound, err = s.store.GCQueueEntryExists(ctx, blob.ContentSHA256) + if err != nil { + return errors.Wrapf(err, "orphan scan: queue lookup for %x", blob.ContentSHA256[:4]) + } + } + + decision := ClassifyChunkBlobOrphan(blob, boundary, rcValue, rcFound, queueFound) + s.observer.ObserveChunkBlobOrphan(decision.Verdict, decision.Reason) + if decision.Verdict != OrphanReclaim { + return nil + } + if err := s.local.DeleteChunkBlob(ctx, blob.ContentSHA256); err != nil { + return errors.Wrapf(err, "orphan scan: delete local blob %x", blob.ContentSHA256[:4]) + } + s.logger.InfoContext(ctx, "chunkblob orphan reclaimed", + slog.String("reason", decision.Reason)) + return nil +} diff --git a/internal/s3keys/chunkblob_orphan_test.go b/internal/s3keys/chunkblob_orphan_test.go new file mode 100644 index 000000000..19e226aa3 --- /dev/null +++ b/internal/s3keys/chunkblob_orphan_test.go @@ -0,0 +1,288 @@ +package s3keys_test + +import ( + "context" + "testing" + "time" + + "github.com/bootjp/elastickv/internal/s3keys" + "github.com/stretchr/testify/require" +) + +const ( + orphanNowMs = uint64(1_700_000_000_000) + orphanNowTS = orphanNowMs << 16 + orphanGrace = 6 * time.Hour + orphanGraceMs = uint64(6 * 60 * 60 * 1000) + orphanBoundary = (orphanNowMs - orphanGraceMs) << 16 +) + +// oldBlob is written comfortably before the grace boundary. +func oldBlob(seed string) s3keys.LocalChunkBlob { + return s3keys.LocalChunkBlob{ + ContentSHA256: testSHA(seed), + WrittenAtTS: orphanBoundary - (1 << 16), + } +} + +// TestClassifyChunkBlobOrphanCoversBothDocumentedSources is the §3.5 +// detection criterion plus the age gate the criterion implies but does +// not state. +func TestClassifyChunkBlobOrphanCoversBothDocumentedSources(t *testing.T) { + t.Parallel() + + zeroRC := s3keys.EncodeChunkRefRC(s3keys.ChunkRefRC{Count: 0}) + liveRC := s3keys.EncodeChunkRefRC(s3keys.ChunkRefRC{Count: 2}) + + tests := []struct { + name string + blob s3keys.LocalChunkBlob + rc []byte + rcFound bool + queueFound bool + wantVerb s3keys.ChunkBlobOrphanVerdict + wantReason string + }{ + { + name: "put aborted before chunkref dispatch", + blob: oldBlob("aborted"), + rcFound: false, + wantVerb: s3keys.OrphanReclaim, + wantReason: s3keys.OrphanReasonNoReferenceRecord, + }, + { + name: "sweeper crashed after the raft phase", + blob: oldBlob("crashed"), + rc: zeroRC, + rcFound: true, + queueFound: false, + wantVerb: s3keys.OrphanReclaim, + wantReason: s3keys.OrphanReasonSweeperCrashed, + }, + { + name: "still referenced", + blob: oldBlob("live"), + rc: liveRC, + rcFound: true, + wantVerb: s3keys.OrphanKeep, + wantReason: s3keys.OrphanReasonReferenced, + }, + { + name: "queue entry still owns it", + blob: oldBlob("queued"), + rc: zeroRC, + rcFound: true, + queueFound: true, + wantVerb: s3keys.OrphanKeep, + wantReason: s3keys.OrphanReasonQueueOwnsIt, + }, + { + name: "unreadable record", + blob: oldBlob("corrupt"), + rc: []byte{0x01}, + rcFound: true, + wantVerb: s3keys.OrphanKeep, + wantReason: s3keys.OrphanReasonRecordUnreadable, + }, + } + + for _, tc := range tests { + t.Run(tc.name, func(t *testing.T) { + t.Parallel() + got := s3keys.ClassifyChunkBlobOrphan( + tc.blob, orphanBoundary, tc.rc, tc.rcFound, tc.queueFound) + require.Equal(t, tc.wantVerb, got.Verdict) + require.Equal(t, tc.wantReason, got.Reason) + }) + } +} + +// TestClassifyChunkBlobOrphanProtectsInFlightUploads is the guard the +// §3.5 text does not spell out but the PUT path requires: chunkblob +// bytes land BEFORE the chunkref commits, so a healthy upload briefly +// looks exactly like the abort case. Without the age gate the scan +// would delete the payload out from under every concurrent PUT. +func TestClassifyChunkBlobOrphanProtectsInFlightUploads(t *testing.T) { + t.Parallel() + + for _, writtenAt := range []uint64{ + orphanBoundary, // exactly at the boundary + orphanBoundary + (1 << 16), // just inside + orphanNowTS, // written this instant + } { + blob := s3keys.LocalChunkBlob{ContentSHA256: testSHA("in-flight"), WrittenAtTS: writtenAt} + got := s3keys.ClassifyChunkBlobOrphan(blob, orphanBoundary, nil, false, false) + require.Equal(t, s3keys.OrphanKeep, got.Verdict, + "a blob written at %d must not be reclaimed", writtenAt) + require.Equal(t, s3keys.OrphanReasonWithinGrace, got.Reason) + } +} + +// TestClassifyChunkBlobOrphanKeepsABlobWithAnUnknownAge pins that a +// missing write timestamp is treated as "too young to judge" rather +// than as epoch-old, which would reclaim it immediately. +func TestClassifyChunkBlobOrphanKeepsABlobWithAnUnknownAge(t *testing.T) { + t.Parallel() + + blob := s3keys.LocalChunkBlob{ContentSHA256: testSHA("no-timestamp")} + got := s3keys.ClassifyChunkBlobOrphan(blob, orphanBoundary, nil, false, false) + require.Equal(t, s3keys.OrphanKeep, got.Verdict) + require.Equal(t, s3keys.OrphanReasonWithinGrace, got.Reason) +} + +// fakeOrphanStore is the replicated half. +type fakeOrphanStore struct { + rc map[[32]byte][]byte + queued map[[32]byte]bool + rcReads int + queueReads int +} + +func (f *fakeOrphanStore) ReadChunkRefRC(_ context.Context, sha [32]byte) ([]byte, bool, error) { + f.rcReads++ + v, ok := f.rc[sha] + return v, ok, nil +} + +func (f *fakeOrphanStore) GCQueueEntryExists(_ context.Context, sha [32]byte) (bool, error) { + f.queueReads++ + return f.queued[sha], nil +} + +type fakeOrphanLocal struct { + blobs []s3keys.LocalChunkBlob + deleted [][32]byte +} + +func (f *fakeOrphanLocal) ListLocalChunkBlobs(_ context.Context) ([]s3keys.LocalChunkBlob, error) { + return f.blobs, nil +} + +func (f *fakeOrphanLocal) DeleteChunkBlob(_ context.Context, sha [32]byte) error { + f.deleted = append(f.deleted, sha) + return nil +} + +func newOrphanScanner(t *testing.T, store *fakeOrphanStore, local *fakeOrphanLocal) *s3keys.ChunkBlobOrphanScanner { + t.Helper() + s, err := s3keys.NewChunkBlobOrphanScanner(s3keys.ChunkBlobOrphanScannerOptions{ + Store: store, + Local: local, + GracePeriod: orphanGrace, + NowTS: func() uint64 { return orphanNowTS }, + }) + require.NoError(t, err) + return s +} + +func TestOrphanScannerReclaimsOnlyTheOrphans(t *testing.T) { + t.Parallel() + + aborted := oldBlob("aborted") + crashed := oldBlob("crashed") + live := oldBlob("live") + queued := oldBlob("queued") + young := s3keys.LocalChunkBlob{ContentSHA256: testSHA("young"), WrittenAtTS: orphanNowTS} + + store := &fakeOrphanStore{ + rc: map[[32]byte][]byte{ + crashed.ContentSHA256: s3keys.EncodeChunkRefRC(s3keys.ChunkRefRC{Count: 0}), + live.ContentSHA256: s3keys.EncodeChunkRefRC(s3keys.ChunkRefRC{Count: 1}), + queued.ContentSHA256: s3keys.EncodeChunkRefRC(s3keys.ChunkRefRC{Count: 0}), + young.ContentSHA256: s3keys.EncodeChunkRefRC(s3keys.ChunkRefRC{Count: 0}), + }, + queued: map[[32]byte]bool{queued.ContentSHA256: true}, + } + local := &fakeOrphanLocal{blobs: []s3keys.LocalChunkBlob{aborted, crashed, live, queued, young}} + + require.NoError(t, newOrphanScanner(t, store, local).ScanOnce(context.Background())) + + require.ElementsMatch(t, + [][32]byte{aborted.ContentSHA256, crashed.ContentSHA256}, + local.deleted, + "only the two §3.5 orphan shapes may be reclaimed") +} + +// TestOrphanScannerSkipsReplicatedReadsForYoungBlobs pins that the age +// gate runs before the replicated lookups. On a healthy node most +// blobs are young or referenced, so paying two reads for each would +// make the scan's cost proportional to total blobs rather than to real +// orphans. +func TestOrphanScannerSkipsReplicatedReadsForYoungBlobs(t *testing.T) { + t.Parallel() + + store := &fakeOrphanStore{} + local := &fakeOrphanLocal{blobs: []s3keys.LocalChunkBlob{ + {ContentSHA256: testSHA("a"), WrittenAtTS: orphanNowTS}, + {ContentSHA256: testSHA("b"), WrittenAtTS: orphanNowTS}, + }} + + require.NoError(t, newOrphanScanner(t, store, local).ScanOnce(context.Background())) + require.Zero(t, store.rcReads, "a young blob must not cost a replicated read") + require.Zero(t, store.queueReads) + require.Empty(t, local.deleted) +} + +// TestOrphanScannerSkipsTheQueueLookupWhenNoRecordExists pins the other +// read-avoidance: with no RC record the §3.5 criterion is already +// satisfied, so the queue lookup would be wasted. +func TestOrphanScannerSkipsTheQueueLookupWhenNoRecordExists(t *testing.T) { + t.Parallel() + + blob := oldBlob("aborted") + store := &fakeOrphanStore{} + local := &fakeOrphanLocal{blobs: []s3keys.LocalChunkBlob{blob}} + + require.NoError(t, newOrphanScanner(t, store, local).ScanOnce(context.Background())) + require.Equal(t, 1, store.rcReads) + require.Zero(t, store.queueReads) + require.Equal(t, [][32]byte{blob.ContentSHA256}, local.deleted) +} + +// TestOrphanScannerReclaimsNothingBeforeTheFirstGraceWindow covers a +// freshly started cluster, where now-grace underflows to the epoch and +// every local blob could still be an upload in progress. +func TestOrphanScannerReclaimsNothingBeforeTheFirstGraceWindow(t *testing.T) { + t.Parallel() + + local := &fakeOrphanLocal{blobs: []s3keys.LocalChunkBlob{oldBlob("whatever")}} + s, err := s3keys.NewChunkBlobOrphanScanner(s3keys.ChunkBlobOrphanScannerOptions{ + Store: &fakeOrphanStore{}, + Local: local, + GracePeriod: orphanGrace, + NowTS: func() uint64 { return uint64(1_000) << 16 }, + }) + require.NoError(t, err) + + require.NoError(t, s.ScanOnce(context.Background())) + require.Empty(t, local.deleted) +} + +func TestNewChunkBlobOrphanScannerValidatesItsCollaborators(t *testing.T) { + t.Parallel() + + valid := s3keys.ChunkBlobOrphanScannerOptions{ + Store: &fakeOrphanStore{}, + Local: &fakeOrphanLocal{}, + NowTS: func() uint64 { return 1 << 16 }, + } + for _, mutate := range []func(*s3keys.ChunkBlobOrphanScannerOptions){ + func(o *s3keys.ChunkBlobOrphanScannerOptions) { o.Store = nil }, + func(o *s3keys.ChunkBlobOrphanScannerOptions) { o.Local = nil }, + func(o *s3keys.ChunkBlobOrphanScannerOptions) { o.NowTS = nil }, + } { + opts := valid + mutate(&opts) + _, err := s3keys.NewChunkBlobOrphanScanner(opts) + require.Error(t, err) + } + _, err := s3keys.NewChunkBlobOrphanScanner(valid) + require.NoError(t, err) +} + +func TestChunkBlobOrphanVerdictStringsAreStable(t *testing.T) { + t.Parallel() + + require.Equal(t, "reclaim", s3keys.OrphanReclaim.String()) + require.Equal(t, "keep", s3keys.OrphanKeep.String()) +} diff --git a/internal/s3keys/chunkblob_rc_plan.go b/internal/s3keys/chunkblob_rc_plan.go new file mode 100644 index 000000000..d18379e99 --- /dev/null +++ b/internal/s3keys/chunkblob_rc_plan.go @@ -0,0 +1,140 @@ +package s3keys + +import "github.com/cockroachdb/errors" + +// Reference-count mutation planning for the §3.5 blob GC. +// +// This is the decision layer the chunkref transaction calls: given the +// reference deltas a txn is about to apply and the reference-count +// records as of its read timestamp, it produces the exact set of +// additional mutations that txn must carry. It is deliberately pure — +// no store, no clock — so the atomic-pair semantics can be tested +// exhaustively without standing up a Raft group. +// +// The §3.5 invariant it enforces: the (chunkref change, RC update) +// pair is the linearisation point for "this blob is now / no longer +// reachable", and the GC queue must reflect *currently* RC==0 rather +// than *ever was* zero. That second clause is what forces a +// re-referencing txn to delete the existing queue entry rather than +// leaving it for the sweeper to re-validate. + +// maxMutationsPerDelta is the most keys one SHA's delta can produce: +// its reference-count record, plus at most one GC-queue insert or +// delete. Named so the pre-size reads as the bound it is. +const maxMutationsPerDelta = 2 + +// 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. +var ErrChunkRefRCUnderflow = errors.New("s3keys: chunkref reference count would underflow") + +// 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 ChunkRefDelta struct { + ContentSHA256 [chunkBlobSHA256Bytes]byte + Added uint64 + Removed uint64 +} + +// 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. +type ChunkRefRCMutation struct { + Key []byte + Value []byte + Delete bool +} + +// 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. +func PlanChunkRefRCMutations( + deltas []ChunkRefDelta, + current map[[chunkBlobSHA256Bytes]byte]ChunkRefRC, + commitTS uint64, +) ([]ChunkRefRCMutation, error) { + if commitTS == 0 { + return nil, errors.Wrap(ErrInvalidChunkRefPlan, "commit timestamp is required") + } + out := make([]ChunkRefRCMutation, 0, len(deltas)*maxMutationsPerDelta) + for _, delta := range deltas { + planned, err := planOneChunkRefDelta(delta, current[delta.ContentSHA256], commitTS) + if err != nil { + return nil, err + } + out = append(out, planned...) + } + return out, nil +} + +// ErrInvalidChunkRefPlan reports a malformed planning request. +var ErrInvalidChunkRefPlan = errors.New("s3keys: invalid chunkref reference-count plan") + +func planOneChunkRefDelta( + delta ChunkRefDelta, existing ChunkRefRC, commitTS uint64, +) ([]ChunkRefRCMutation, error) { + if delta.Added == 0 && delta.Removed == 0 { + // A net-zero delta still must not touch the queue: the blob's + // reachability did not change, so neither should the record. + return nil, nil + } + if delta.Removed > existing.Count+delta.Added { + return nil, errors.Wrapf(ErrChunkRefRCUnderflow, + "sha=%x count=%d added=%d removed=%d", + delta.ContentSHA256[:4], existing.Count, delta.Added, delta.Removed) + } + next := ChunkRefRC{Count: existing.Count + delta.Added - delta.Removed} + + mutations := make([]ChunkRefRCMutation, 0, maxMutationsPerDelta) + switch { + case next.Count == 0: + // Newly unreachable. Record WHEN so the sweeper's grace window + // has a time signal, and queue it. + // + // An already-queued record keeps its original timestamp: the + // blob has been continuously unreachable, and restamping it + // would silently restart a grace period that was already + // running. + if existing.Queued() { + next.QueuedAtTS = existing.QueuedAtTS + } else { + next.QueuedAtTS = commitTS + mutations = append(mutations, ChunkRefRCMutation{ + Key: ChunkBlobGCQueueKey(commitTS, delta.ContentSHA256), + Value: []byte{}, + }) + } + case existing.Queued(): + // Reachable again before the sweeper ran. The queue must + // reflect CURRENTLY RC==0, so the entry goes away in this same + // txn — which is only possible because the record carries the + // timestamp the key was built from. + mutations = append(mutations, ChunkRefRCMutation{ + Key: ChunkBlobGCQueueKey(existing.QueuedAtTS, delta.ContentSHA256), + Delete: true, + }) + } + + return append(mutations, ChunkRefRCMutation{ + Key: ChunkRefRCKey(delta.ContentSHA256), + Value: EncodeChunkRefRC(next), + }), nil +} diff --git a/internal/s3keys/chunkblob_rc_plan_test.go b/internal/s3keys/chunkblob_rc_plan_test.go new file mode 100644 index 000000000..bc936a97e --- /dev/null +++ b/internal/s3keys/chunkblob_rc_plan_test.go @@ -0,0 +1,208 @@ +package s3keys_test + +import ( + "bytes" + "testing" + + "github.com/bootjp/elastickv/internal/s3keys" + "github.com/cockroachdb/errors" + "github.com/stretchr/testify/require" +) + +const planCommitTS = uint64(1_700_000_000_000) << 16 + +// findMutation returns the planned mutation for key, if any. +func findMutation(t *testing.T, plan []s3keys.ChunkRefRCMutation, key []byte) (s3keys.ChunkRefRCMutation, bool) { + t.Helper() + for _, m := range plan { + if bytes.Equal(m.Key, key) { + return m, true + } + } + return s3keys.ChunkRefRCMutation{}, false +} + +func requireRCValue(t *testing.T, plan []s3keys.ChunkRefRCMutation, sha [32]byte, want s3keys.ChunkRefRC) { + t.Helper() + m, ok := findMutation(t, plan, s3keys.ChunkRefRCKey(sha)) + require.True(t, ok, "plan must write the reference-count record") + require.False(t, m.Delete) + got, ok := s3keys.DecodeChunkRefRC(m.Value) + require.True(t, ok) + require.Equal(t, want, got) +} + +// TestPlanFirstReferenceWritesCountWithoutQueueing covers the ordinary +// upload: a blob becomes reachable, so nothing is queued. +func TestPlanFirstReferenceWritesCountWithoutQueueing(t *testing.T) { + t.Parallel() + + sha := testSHA("payload") + plan, err := s3keys.PlanChunkRefRCMutations( + []s3keys.ChunkRefDelta{{ContentSHA256: sha, Added: 1}}, nil, planCommitTS) + require.NoError(t, err) + require.Len(t, plan, 1, "a first reference touches only the count") + requireRCValue(t, plan, sha, s3keys.ChunkRefRC{Count: 1}) +} + +// TestPlanLastReferenceRemovalQueuesWithTheCommitTimestamp is the +// eligibility half of §3.5: the same txn that drives the count to zero +// must record WHEN, because a counter resting at zero carries no time +// signal and the grace window would be unimplementable. +func TestPlanLastReferenceRemovalQueuesWithTheCommitTimestamp(t *testing.T) { + t.Parallel() + + sha := testSHA("payload") + plan, err := s3keys.PlanChunkRefRCMutations( + []s3keys.ChunkRefDelta{{ContentSHA256: sha, Removed: 1}}, + map[[32]byte]s3keys.ChunkRefRC{sha: {Count: 1}}, + planCommitTS) + require.NoError(t, err) + + queueKey := s3keys.ChunkBlobGCQueueKey(planCommitTS, sha) + q, ok := findMutation(t, plan, queueKey) + require.True(t, ok, "dropping to zero must queue the blob") + require.False(t, q.Delete) + + requireRCValue(t, plan, sha, s3keys.ChunkRefRC{Count: 0, QueuedAtTS: planCommitTS}) +} + +// TestPlanReReferenceDeletesTheExistingQueueEntry is the clause that +// forced the timestamp into the RC value: §3.5 requires the queue to +// reflect *currently* RC==0, not *ever was* zero, so a txn that makes a +// blob reachable again must remove the entry in the same txn — which it +// can only name because the record carries the timestamp. +func TestPlanReReferenceDeletesTheExistingQueueEntry(t *testing.T) { + t.Parallel() + + sha := testSHA("payload") + const queuedAt = uint64(1_699_000_000_000) << 16 + + plan, err := s3keys.PlanChunkRefRCMutations( + []s3keys.ChunkRefDelta{{ContentSHA256: sha, Added: 1}}, + map[[32]byte]s3keys.ChunkRefRC{sha: {Count: 0, QueuedAtTS: queuedAt}}, + planCommitTS) + require.NoError(t, err) + + del, ok := findMutation(t, plan, s3keys.ChunkBlobGCQueueKey(queuedAt, sha)) + require.True(t, ok, "re-referencing must delete the stale queue entry") + require.True(t, del.Delete) + require.Nil(t, del.Value) + + // The count record no longer claims a queue entry. + requireRCValue(t, plan, sha, s3keys.ChunkRefRC{Count: 1}) +} + +// TestPlanKeepsTheOriginalEligibilityTimestamp pins that a blob which +// is already queued and stays at zero does NOT get restamped. +// Restamping would silently restart a grace period that was already +// running, so a blob could never age out under repeated no-op txns. +func TestPlanKeepsTheOriginalEligibilityTimestamp(t *testing.T) { + t.Parallel() + + sha := testSHA("payload") + const queuedAt = uint64(1_699_000_000_000) << 16 + + // A txn that adds and removes one reference: nets to zero, and the + // blob was already queued. + plan, err := s3keys.PlanChunkRefRCMutations( + []s3keys.ChunkRefDelta{{ContentSHA256: sha, Added: 1, Removed: 1}}, + map[[32]byte]s3keys.ChunkRefRC{sha: {Count: 0, QueuedAtTS: queuedAt}}, + planCommitTS) + require.NoError(t, err) + + requireRCValue(t, plan, sha, s3keys.ChunkRefRC{Count: 0, QueuedAtTS: queuedAt}) + _, requeued := findMutation(t, plan, s3keys.ChunkBlobGCQueueKey(planCommitTS, sha)) + require.False(t, requeued, "an already-queued blob must keep its original timestamp") +} + +// TestPlanUnderflowFailsClosed pins that a decrement below zero fails +// the txn instead of clamping. Clamping would queue a blob for deletion +// on the strength of a bookkeeping disagreement — a correctness bug +// dressed up as a space reclaim. +func TestPlanUnderflowFailsClosed(t *testing.T) { + t.Parallel() + + sha := testSHA("payload") + tests := []struct { + name string + current map[[32]byte]s3keys.ChunkRefRC + delta s3keys.ChunkRefDelta + }{ + { + name: "no record at all", + current: nil, + delta: s3keys.ChunkRefDelta{ContentSHA256: sha, Removed: 1}, + }, + { + name: "removing more than held", + current: map[[32]byte]s3keys.ChunkRefRC{sha: {Count: 2}}, + delta: s3keys.ChunkRefDelta{ContentSHA256: sha, Removed: 3}, + }, + { + name: "adds do not cover removes", + current: map[[32]byte]s3keys.ChunkRefRC{sha: {Count: 1}}, + delta: s3keys.ChunkRefDelta{ContentSHA256: sha, Added: 1, Removed: 3}, + }, + } + for _, tc := range tests { + t.Run(tc.name, func(t *testing.T) { + t.Parallel() + _, err := s3keys.PlanChunkRefRCMutations( + []s3keys.ChunkRefDelta{tc.delta}, tc.current, planCommitTS) + require.Error(t, err) + require.True(t, errors.Is(err, s3keys.ErrChunkRefRCUnderflow)) + }) + } +} + +// TestPlanNetZeroDeltaIsANoOp pins that a txn which neither adds nor +// removes references leaves the record alone — including its queue +// state, since reachability did not change. +func TestPlanNetZeroDeltaIsANoOp(t *testing.T) { + t.Parallel() + + sha := testSHA("payload") + plan, err := s3keys.PlanChunkRefRCMutations( + []s3keys.ChunkRefDelta{{ContentSHA256: sha}}, + map[[32]byte]s3keys.ChunkRefRC{sha: {Count: 3}}, + planCommitTS) + require.NoError(t, err) + require.Empty(t, plan) +} + +// TestPlanHandlesDedupAcrossMultipleSHAsInOneTxn covers a multipart +// upload touching several chunks at once: each SHA is planned +// independently and a drop to zero for one must not affect another. +func TestPlanHandlesDedupAcrossMultipleSHAsInOneTxn(t *testing.T) { + t.Parallel() + + keep := testSHA("still-referenced") + drop := testSHA("about-to-be-orphaned") + + plan, err := s3keys.PlanChunkRefRCMutations([]s3keys.ChunkRefDelta{ + {ContentSHA256: keep, Added: 1}, + {ContentSHA256: drop, Removed: 1}, + }, map[[32]byte]s3keys.ChunkRefRC{ + keep: {Count: 1}, + drop: {Count: 1}, + }, planCommitTS) + require.NoError(t, err) + + requireRCValue(t, plan, keep, s3keys.ChunkRefRC{Count: 2}) + requireRCValue(t, plan, drop, s3keys.ChunkRefRC{Count: 0, QueuedAtTS: planCommitTS}) + + _, keepQueued := findMutation(t, plan, s3keys.ChunkBlobGCQueueKey(planCommitTS, keep)) + require.False(t, keepQueued, "a still-referenced blob must never be queued") + _, dropQueued := findMutation(t, plan, s3keys.ChunkBlobGCQueueKey(planCommitTS, drop)) + require.True(t, dropQueued) +} + +func TestPlanRequiresACommitTimestamp(t *testing.T) { + t.Parallel() + + _, err := s3keys.PlanChunkRefRCMutations( + []s3keys.ChunkRefDelta{{ContentSHA256: testSHA("payload"), Added: 1}}, nil, 0) + require.Error(t, err) + require.True(t, errors.Is(err, s3keys.ErrInvalidChunkRefPlan)) +} diff --git a/internal/s3keys/chunkblob_sweep_plan.go b/internal/s3keys/chunkblob_sweep_plan.go new file mode 100644 index 000000000..384afb972 --- /dev/null +++ b/internal/s3keys/chunkblob_sweep_plan.go @@ -0,0 +1,126 @@ +package s3keys + +// Sweep classification for the §3.5 blob GC. +// +// This is the decision half of the node-local sweeper: given one queue +// entry and the reference-count record as of the sweeper's read +// timestamp, it says what the sweeper is permitted to do. Pure, so the +// correctness-critical classification is testable without a Raft group +// or a Pebble store; the two-phase execution (Raft conditional delete, +// then local unlink) is the caller's. +// +// Why the classification carries the weight: §3.5 notes that an +// UNCONDITIONAL queue delete would let the sweeper proceed to +// local-delete a chunkblob that is currently live — a correctness bug, +// not a space leak. The verdicts below are what the caller turns into +// a conditional Raft txn, so getting them wrong is exactly that bug. + +// ChunkBlobSweepVerdict is what a sweeper may do with one queue entry. +type ChunkBlobSweepVerdict int + +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 +) + +func (v ChunkBlobSweepVerdict) String() string { + switch v { + case SweepReclaim: + return "reclaim" + case SweepDropQueueEntryOnly: + return "drop_queue_entry_only" + case SweepSkip: + return "skip" + default: + return "unknown" + } +} + +// 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. +type ChunkBlobSweepDecision struct { + Verdict ChunkBlobSweepVerdict + Reason string +} + +// Reasons, a closed set so a sweeper can use them as a metric label. +const ( + SweepReasonUnreferenced = "unreferenced" + SweepReasonReferencedAgain = "referenced_again" + SweepReasonSupersededEntry = "superseded_entry" + SweepReasonRecordUnreadable = "record_unreadable" + SweepReasonRecordNotQueued = "record_not_queued" +) + +// 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. +func ClassifyChunkBlobSweep(entryTS uint64, rcValue []byte, rcFound bool) ChunkBlobSweepDecision { + if !rcFound { + // No reference-count record at all. The blob is unreachable + // through any chunkref, and the queue entry is the only thing + // tracking it — reclaim. This is also the PUT-abort orphan + // shape §3.5 describes, except those never reach the queue and + // are the orphan scan's job instead. + return ChunkBlobSweepDecision{Verdict: SweepReclaim, Reason: SweepReasonUnreferenced} + } + + rc, ok := DecodeChunkRefRC(rcValue) + if !ok { + // A malformed count must never be read as zero: that would + // reclaim a blob on the strength of corruption. Decline and + // leave the entry for an operator. + return ChunkBlobSweepDecision{Verdict: SweepSkip, Reason: SweepReasonRecordUnreadable} + } + + if rc.Count > 0 { + // §3.5(c): referenced again. The entry is stale — either a + // re-reference txn failed to remove it or this sweeper raced + // one. Drop the entry, keep the blob. + return ChunkBlobSweepDecision{ + Verdict: SweepDropQueueEntryOnly, + Reason: SweepReasonReferencedAgain, + } + } + + switch { + case !rc.Queued(): + // Count is zero but the record claims no queue entry. The + // entry cannot be matched to the record, so reclaiming on it + // would be acting on state the record does not corroborate. + return ChunkBlobSweepDecision{Verdict: SweepSkip, Reason: SweepReasonRecordNotQueued} + case rc.QueuedAtTS != entryTS: + // A newer queueing superseded this entry: the blob went + // unreferenced, was referenced again, and went unreferenced + // once more. The later entry owns the grace window, so this + // one is garbage — but the BLOB is not, because the newer + // entry has not served its own grace yet. + return ChunkBlobSweepDecision{ + Verdict: SweepDropQueueEntryOnly, + Reason: SweepReasonSupersededEntry, + } + default: + return ChunkBlobSweepDecision{Verdict: SweepReclaim, Reason: SweepReasonUnreferenced} + } +} diff --git a/internal/s3keys/chunkblob_sweep_plan_test.go b/internal/s3keys/chunkblob_sweep_plan_test.go new file mode 100644 index 000000000..5a40bf8a4 --- /dev/null +++ b/internal/s3keys/chunkblob_sweep_plan_test.go @@ -0,0 +1,121 @@ +package s3keys_test + +import ( + "testing" + + "github.com/bootjp/elastickv/internal/s3keys" + "github.com/stretchr/testify/require" +) + +const sweepEntryTS = uint64(1_700_000_000_000) << 16 + +// TestClassifyChunkBlobSweepCoversEveryRecordShape is the table the +// §3.5 correctness argument rests on. The design is explicit that an +// UNCONDITIONAL queue delete would let the sweeper local-delete a blob +// that is currently live — a correctness bug, not a space leak — so +// these verdicts are what keep that from happening. +func TestClassifyChunkBlobSweepCoversEveryRecordShape(t *testing.T) { + t.Parallel() + + tests := []struct { + name string + rc []byte + found bool + wantVerb s3keys.ChunkBlobSweepVerdict + wantReason string + }{ + { + name: "no record at all", + found: false, + wantVerb: s3keys.SweepReclaim, + wantReason: s3keys.SweepReasonUnreferenced, + }, + { + name: "zero count queued at this entry", + rc: s3keys.EncodeChunkRefRC(s3keys.ChunkRefRC{Count: 0, QueuedAtTS: sweepEntryTS}), + found: true, + wantVerb: s3keys.SweepReclaim, + wantReason: s3keys.SweepReasonUnreferenced, + }, + { + name: "referenced again", + rc: s3keys.EncodeChunkRefRC(s3keys.ChunkRefRC{Count: 1}), + found: true, + wantVerb: s3keys.SweepDropQueueEntryOnly, + wantReason: s3keys.SweepReasonReferencedAgain, + }, + { + name: "superseded by a newer queueing", + rc: s3keys.EncodeChunkRefRC(s3keys.ChunkRefRC{ + Count: 0, QueuedAtTS: sweepEntryTS + (1 << 16), + }), + found: true, + wantVerb: s3keys.SweepDropQueueEntryOnly, + wantReason: s3keys.SweepReasonSupersededEntry, + }, + { + name: "zero count claiming no queue entry", + rc: s3keys.EncodeChunkRefRC(s3keys.ChunkRefRC{Count: 0}), + found: true, + wantVerb: s3keys.SweepSkip, + wantReason: s3keys.SweepReasonRecordNotQueued, + }, + { + name: "malformed record", + rc: []byte{0x01, 0x02}, + found: true, + wantVerb: s3keys.SweepSkip, + wantReason: s3keys.SweepReasonRecordUnreadable, + }, + } + + for _, tc := range tests { + t.Run(tc.name, func(t *testing.T) { + t.Parallel() + got := s3keys.ClassifyChunkBlobSweep(sweepEntryTS, tc.rc, tc.found) + require.Equal(t, tc.wantVerb, got.Verdict, "verdict for %s", tc.name) + require.Equal(t, tc.wantReason, got.Reason) + }) + } +} + +// TestClassifyChunkBlobSweepNeverReclaimsALiveBlob is the single +// property that matters most: no record shape carrying a live +// reference may produce a verdict that deletes the blob. +func TestClassifyChunkBlobSweepNeverReclaimsALiveBlob(t *testing.T) { + t.Parallel() + + for _, count := range []uint64{1, 2, 7, 1 << 20} { + for _, queuedAt := range []uint64{0, sweepEntryTS, sweepEntryTS + (1 << 16)} { + rc := s3keys.EncodeChunkRefRC(s3keys.ChunkRefRC{Count: count, QueuedAtTS: queuedAt}) + got := s3keys.ClassifyChunkBlobSweep(sweepEntryTS, rc, true) + require.NotEqual(t, s3keys.SweepReclaim, got.Verdict, + "count=%d queuedAt=%d must never reclaim", count, queuedAt) + } + } +} + +// TestClassifyChunkBlobSweepNeverReclaimsOnCorruption pins that a +// value the decoder rejects is never read as "count zero". Treating +// corruption as zero would delete live data on the strength of a bad +// byte. +func TestClassifyChunkBlobSweepNeverReclaimsOnCorruption(t *testing.T) { + t.Parallel() + + for _, bad := range [][]byte{{}, {0x00}, make([]byte, 8), make([]byte, 15), make([]byte, 17)} { + got := s3keys.ClassifyChunkBlobSweep(sweepEntryTS, bad, true) + require.Equal(t, s3keys.SweepSkip, got.Verdict) + require.Equal(t, s3keys.SweepReasonRecordUnreadable, got.Reason) + } +} + +// TestChunkBlobSweepVerdictStringsAreStable guards the metric label: +// these strings are a closed set a sweeper can emit directly. +func TestChunkBlobSweepVerdictStringsAreStable(t *testing.T) { + t.Parallel() + + require.Equal(t, "reclaim", s3keys.SweepReclaim.String()) + require.Equal(t, "drop_queue_entry_only", s3keys.SweepDropQueueEntryOnly.String()) + require.Equal(t, "skip", s3keys.SweepSkip.String()) + require.Equal(t, "unknown", s3keys.ChunkBlobSweepVerdict(99).String()) +} diff --git a/internal/s3keys/chunkblob_sweeper.go b/internal/s3keys/chunkblob_sweeper.go new file mode 100644 index 000000000..7f1dc893b --- /dev/null +++ b/internal/s3keys/chunkblob_sweeper.go @@ -0,0 +1,294 @@ +package s3keys + +import ( + "context" + "log/slog" + "time" + + "github.com/cockroachdb/errors" +) + +// 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. +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 +) + +// 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. +var ErrQueueEntryChanged = errors.New("s3keys: gc queue entry changed before the conditional delete") + +// ChunkBlobGCQueueEntry is one entry returned by a queue scan. +type ChunkBlobGCQueueEntry struct { + CommitTS uint64 + ContentSHA256 [chunkBlobSHA256Bytes]byte +} + +// ChunkBlobSweepStore is the replicated half: the GC queue and the +// reference counts, both read and written through Raft. +type ChunkBlobSweepStore interface { + // ScanGCQueue returns every queue entry in [startKey, endKey). + // It must be all-or-error: a partial scan would simply delay + // entries to the next pass, which is safe, but a scan that + // silently truncated mid-range while reporting success would hide + // a persistent backlog. + ScanGCQueue(ctx context.Context, startKey, endKey []byte) ([]ChunkBlobGCQueueEntry, 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 only if + // it still exists AND the reference count is still zero, returning + // ErrQueueEntryChanged otherwise. Concurrent sweepers serialise + // here on the queue key's write-write conflict. + 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 +} + +// ChunkBlobLocalStore is the node-local half: the chunkblob payload in +// Pebble, never written through Raft. +type ChunkBlobLocalStore interface { + DeleteChunkBlob(ctx context.Context, sha [chunkBlobSHA256Bytes]byte) error +} + +// ChunkBlobSweepObserver receives per-entry outcomes. +type ChunkBlobSweepObserver interface { + ObserveChunkBlobSweep(verdict ChunkBlobSweepVerdict, reason string) + ObserveChunkBlobSweepRaceLost() +} + +type nopSweepObserver struct{} + +func (nopSweepObserver) ObserveChunkBlobSweep(ChunkBlobSweepVerdict, string) {} +func (nopSweepObserver) ObserveChunkBlobSweepRaceLost() {} + +// ChunkBlobSweeper reclaims chunkblobs whose references are gone. +type ChunkBlobSweeper struct { + store ChunkBlobSweepStore + local ChunkBlobLocalStore + grace time.Duration + interval time.Duration + nowTS func() uint64 + observer ChunkBlobSweepObserver + logger *slog.Logger +} + +// 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 ChunkBlobSweeperOptions struct { + Store ChunkBlobSweepStore + Local ChunkBlobLocalStore + GracePeriod time.Duration + Interval time.Duration + NowTS func() uint64 + Observer ChunkBlobSweepObserver + Logger *slog.Logger +} + +func NewChunkBlobSweeper(opts ChunkBlobSweeperOptions) (*ChunkBlobSweeper, error) { + switch { + case opts.Store == nil: + return nil, errors.Wrap(ErrInvalidChunkRefPlan, "chunkblob sweeper requires a replicated store") + case opts.Local == nil: + return nil, errors.Wrap(ErrInvalidChunkRefPlan, "chunkblob sweeper requires a local blob store") + case opts.NowTS == nil: + return nil, errors.Wrap(ErrInvalidChunkRefPlan, "chunkblob sweeper requires an HLC clock") + } + timing := resolveGCLoopTiming( + opts.GracePeriod, opts.Interval, + DefaultChunkBlobGCGracePeriod, DefaultChunkBlobGCInterval, opts.Logger) + observer := opts.Observer + if observer == nil { + observer = nopSweepObserver{} + } + return &ChunkBlobSweeper{ + store: opts.Store, + local: opts.Local, + grace: timing.grace, + interval: timing.interval, + nowTS: opts.NowTS, + observer: observer, + logger: timing.logger, + }, nil +} + +// gcLoopTiming is the cadence/logging configuration both GC loops +// share. Factored out because the sweeper and the orphan scanner +// only in their defaults, and duplicating the resolution invites the +// two from drifting apart. +type gcLoopTiming struct { + grace time.Duration + interval time.Duration + logger *slog.Logger +} + +// resolveGCLoopTiming applies the caller's values, falling back to the +// supplied defaults for anything non-positive. +func resolveGCLoopTiming( + grace, interval, defaultGrace, defaultInterval time.Duration, logger *slog.Logger, +) gcLoopTiming { + out := gcLoopTiming{grace: grace, interval: interval, logger: logger} + if out.grace <= 0 { + out.grace = defaultGrace + } + if out.interval <= 0 { + out.interval = defaultInterval + } + if out.logger == nil { + out.logger = slog.Default() + } + return out +} + +// 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 (s *ChunkBlobSweeper) Run(ctx context.Context) error { + if ctx == nil { + return errors.Wrap(ErrInvalidChunkRefPlan, "chunkblob sweeper context is required") + } + ticker := time.NewTicker(s.interval) + defer ticker.Stop() + for { + if sweepCancelled(ctx) { + return nil + } + // A failing pass is logged and retried next tick: the queue is + // durable, so a transient store error costs a delay rather + // than a lost reclaim. Cancellation is handled above and in + // the select, so it is never reported as a sweep failure. + if err := s.SweepOnce(ctx); err != nil && !sweepCancelled(ctx) { + s.logger.WarnContext(ctx, "chunkblob gc sweep failed", + slog.String("error", err.Error())) + } + select { + case <-ctx.Done(): + return nil + case <-ticker.C: + } + } +} + +// SweepOnce runs one pass over the entries whose grace window has +// elapsed. +func (s *ChunkBlobSweeper) SweepOnce(ctx context.Context) error { + boundary := ChunkBlobGCGraceBoundary(s.nowTS(), s.grace) + if boundary == 0 { + // Nothing can have served a full grace window yet. + return nil + } + entries, err := s.store.ScanGCQueue(ctx, + ChunkBlobGCQueueScanStart(), ChunkBlobGCQueueScanEnd(boundary)) + if err != nil { + return errors.Wrap(err, "chunkblob gc: scan queue") + } + for _, entry := range entries { + // Stop cleanly on cancellation: the remaining entries stay + // queued and the next pass picks them up, so an interrupted + // sweep costs a delay rather than a lost reclaim. + if sweepCancelled(ctx) { + break + } + if err := s.sweepEntry(ctx, entry); err != nil { + return err + } + } + return nil +} + +// sweepEntry classifies and executes one entry. +func (s *ChunkBlobSweeper) sweepEntry(ctx context.Context, entry ChunkBlobGCQueueEntry) error { + rcValue, found, err := s.store.ReadChunkRefRC(ctx, entry.ContentSHA256) + if err != nil { + return errors.Wrapf(err, "chunkblob gc: read reference count for %x", entry.ContentSHA256[:4]) + } + decision := ClassifyChunkBlobSweep(entry.CommitTS, rcValue, found) + s.observer.ObserveChunkBlobSweep(decision.Verdict, decision.Reason) + + switch decision.Verdict { + case SweepSkip: + s.logger.WarnContext(ctx, "chunkblob gc declined to sweep", + slog.String("reason", decision.Reason)) + return nil + case SweepDropQueueEntryOnly: + if err := s.store.DeleteGCQueueEntry(ctx, entry); err != nil { + return errors.Wrapf(err, "chunkblob gc: drop stale queue entry for %x", entry.ContentSHA256[:4]) + } + return nil + case SweepReclaim: + return s.reclaim(ctx, entry) + default: + return nil + } +} + +// reclaim runs the two phases in the order §3.5 mandates: the Raft +// conditional delete FIRST, the local unlink second. +// +// The ordering is the load-bearing detail. Local-first would leave a +// crash window in which the blob is gone locally but the queue entry +// survives, so every later pass re-attempts a no-op local delete and +// the entry never clears without manual intervention. Raft-first +// inverts that into a bounded local space leak — the entry is gone but +// the blob is still on disk — which the orphan scan reclaims. +func (s *ChunkBlobSweeper) reclaim(ctx context.Context, entry ChunkBlobGCQueueEntry) error { + if err := s.store.DeleteGCQueueEntryIfUnreferenced(ctx, entry); err != nil { + if errors.Is(err, ErrQueueEntryChanged) { + // Another sweeper won, or a re-reference txn committed + // between the classification and here. Either way the blob + // may now be live: do NOT touch it. + s.observer.ObserveChunkBlobSweepRaceLost() + return nil + } + return errors.Wrapf(err, "chunkblob gc: conditional queue delete for %x", entry.ContentSHA256[:4]) + } + // Reaching here means the conditional delete committed, which + // implies the reference count was zero at its read timestamp and + // stayed zero through its commit window — the blob is genuinely + // unreachable. + if err := s.local.DeleteChunkBlob(ctx, entry.ContentSHA256); err != nil { + return errors.Wrapf(err, "chunkblob gc: local delete for %x", entry.ContentSHA256[:4]) + } + return nil +} + +// sweepCancelled reports whether ctx is done. It exists as a bool +// predicate so the cancellation checks above read as control flow +// rather than as error handling — cancellation is an orderly stop, and +// the remaining queue entries are picked up by the next pass. +func sweepCancelled(ctx context.Context) bool { + return ctx.Err() != nil +} diff --git a/internal/s3keys/chunkblob_sweeper_test.go b/internal/s3keys/chunkblob_sweeper_test.go new file mode 100644 index 000000000..3cda0d032 --- /dev/null +++ b/internal/s3keys/chunkblob_sweeper_test.go @@ -0,0 +1,258 @@ +package s3keys_test + +import ( + "context" + "testing" + "time" + + "github.com/bootjp/elastickv/internal/s3keys" + "github.com/cockroachdb/errors" + "github.com/stretchr/testify/require" +) + +// fakeSweepStore records the order of every replicated operation so a +// test can assert the §3.5 phase ordering rather than just the effects. +type fakeSweepStore struct { + entries []s3keys.ChunkBlobGCQueueEntry + rc map[[32]byte][]byte + + calls []string + scanErr error + condDeleteErr error + unconDeletes int + condDeletes int +} + +func (f *fakeSweepStore) ScanGCQueue(_ context.Context, _, _ []byte) ([]s3keys.ChunkBlobGCQueueEntry, error) { + f.calls = append(f.calls, "scan") + if f.scanErr != nil { + return nil, f.scanErr + } + return f.entries, nil +} + +func (f *fakeSweepStore) ReadChunkRefRC(_ context.Context, sha [32]byte) ([]byte, bool, error) { + f.calls = append(f.calls, "read-rc") + v, ok := f.rc[sha] + return v, ok, nil +} + +func (f *fakeSweepStore) DeleteGCQueueEntryIfUnreferenced(_ context.Context, _ s3keys.ChunkBlobGCQueueEntry) error { + f.calls = append(f.calls, "raft-conditional-delete") + f.condDeletes++ + return f.condDeleteErr +} + +func (f *fakeSweepStore) DeleteGCQueueEntry(_ context.Context, _ s3keys.ChunkBlobGCQueueEntry) error { + f.calls = append(f.calls, "raft-unconditional-delete") + f.unconDeletes++ + return nil +} + +type fakeLocalStore struct { + calls *[]string + deletes [][32]byte +} + +func (f *fakeLocalStore) DeleteChunkBlob(_ context.Context, sha [32]byte) error { + *f.calls = append(*f.calls, "local-delete") + f.deletes = append(f.deletes, sha) + return nil +} + +func newSweeperFixture(t *testing.T, store *fakeSweepStore) (*s3keys.ChunkBlobSweeper, *fakeLocalStore) { + t.Helper() + local := &fakeLocalStore{calls: &store.calls} + // An HLC "now" far enough ahead that every fixture entry has served + // its grace window. + nowTS := (uint64(1_700_000_000_000) + 7_200_000) << 16 + sweeper, err := s3keys.NewChunkBlobSweeper(s3keys.ChunkBlobSweeperOptions{ + Store: store, + Local: local, + GracePeriod: time.Hour, + NowTS: func() uint64 { return nowTS }, + }) + require.NoError(t, err) + return sweeper, local +} + +func queuedEntry(sha [32]byte) s3keys.ChunkBlobGCQueueEntry { + return s3keys.ChunkBlobGCQueueEntry{ + CommitTS: uint64(1_700_000_000_000) << 16, + ContentSHA256: sha, + } +} + +// TestSweeperRunsTheRaftPhaseBeforeTheLocalDelete is the §3.5 phase +// ordering. Local-first would leave a crash window where the blob is +// gone locally but the queue entry survives, so every later pass +// re-attempts a no-op local delete and the entry never clears without +// manual intervention. Raft-first inverts that into a bounded local +// space leak the orphan scan reclaims. +func TestSweeperRunsTheRaftPhaseBeforeTheLocalDelete(t *testing.T) { + t.Parallel() + + sha := testSHA("orphaned") + entry := queuedEntry(sha) + store := &fakeSweepStore{ + entries: []s3keys.ChunkBlobGCQueueEntry{entry}, + rc: map[[32]byte][]byte{ + sha: s3keys.EncodeChunkRefRC(s3keys.ChunkRefRC{Count: 0, QueuedAtTS: entry.CommitTS}), + }, + } + sweeper, local := newSweeperFixture(t, store) + + require.NoError(t, sweeper.SweepOnce(context.Background())) + + require.Equal(t, + []string{"scan", "read-rc", "raft-conditional-delete", "local-delete"}, + store.calls, + "the replicated conditional delete must commit before the local unlink") + require.Equal(t, [][32]byte{sha}, local.deletes) +} + +// TestSweeperDoesNotTouchTheBlobWhenItLosesTheRace is the correctness +// property §3.5 calls out explicitly: an unconditional delete would +// silently succeed on an already-absent entry and let the sweeper +// local-delete a blob that is currently live. +func TestSweeperDoesNotTouchTheBlobWhenItLosesTheRace(t *testing.T) { + t.Parallel() + + sha := testSHA("contended") + entry := queuedEntry(sha) + store := &fakeSweepStore{ + entries: []s3keys.ChunkBlobGCQueueEntry{entry}, + rc: map[[32]byte][]byte{ + sha: s3keys.EncodeChunkRefRC(s3keys.ChunkRefRC{Count: 0, QueuedAtTS: entry.CommitTS}), + }, + condDeleteErr: s3keys.ErrQueueEntryChanged, + } + sweeper, local := newSweeperFixture(t, store) + + require.NoError(t, sweeper.SweepOnce(context.Background()), + "losing the race is a normal outcome, not a sweep failure") + require.Empty(t, local.deletes, + "a lost conditional delete means the blob may be live again; it must not be deleted") + require.NotContains(t, store.calls, "local-delete") +} + +// TestSweeperDropsAStaleEntryWithoutDeletingTheBlob covers §3.5(c): +// the blob is referenced again, so only the entry goes. +func TestSweeperDropsAStaleEntryWithoutDeletingTheBlob(t *testing.T) { + t.Parallel() + + sha := testSHA("referenced-again") + entry := queuedEntry(sha) + store := &fakeSweepStore{ + entries: []s3keys.ChunkBlobGCQueueEntry{entry}, + rc: map[[32]byte][]byte{ + sha: s3keys.EncodeChunkRefRC(s3keys.ChunkRefRC{Count: 1}), + }, + } + sweeper, local := newSweeperFixture(t, store) + + require.NoError(t, sweeper.SweepOnce(context.Background())) + require.Equal(t, 1, store.unconDeletes) + require.Zero(t, store.condDeletes) + require.Empty(t, local.deletes, "a referenced blob must survive") +} + +// TestSweeperDeclinesOnAnUnreadableRecord pins that corruption stops +// the sweep for that entry rather than reclaiming on a guess: neither +// the queue entry nor the blob is touched. +func TestSweeperDeclinesOnAnUnreadableRecord(t *testing.T) { + t.Parallel() + + sha := testSHA("corrupt") + store := &fakeSweepStore{ + entries: []s3keys.ChunkBlobGCQueueEntry{queuedEntry(sha)}, + rc: map[[32]byte][]byte{sha: {0xAA}}, + } + sweeper, local := newSweeperFixture(t, store) + + require.NoError(t, sweeper.SweepOnce(context.Background())) + require.Zero(t, store.condDeletes) + require.Zero(t, store.unconDeletes) + require.Empty(t, local.deletes) +} + +// TestSweeperSkipsEntriesInsideTheGraceWindow pins that the scan +// boundary is applied: an entry stamped now has not served its grace. +func TestSweeperSkipsEntriesInsideTheGraceWindow(t *testing.T) { + t.Parallel() + + nowMs := uint64(1_700_000_000_000) + nowTS := nowMs << 16 + store := &fakeSweepStore{} + local := &fakeLocalStore{calls: &store.calls} + sweeper, err := s3keys.NewChunkBlobSweeper(s3keys.ChunkBlobSweeperOptions{ + Store: store, + Local: local, + GracePeriod: time.Hour, + NowTS: func() uint64 { return nowTS }, + }) + require.NoError(t, err) + + require.NoError(t, sweeper.SweepOnce(context.Background())) + // The scan happened, but bounded to entries older than now-1h. + require.Equal(t, []string{"scan"}, store.calls) +} + +// TestSweeperSweepsNothingBeforeTheFirstGraceWindowElapses covers a +// freshly started cluster, where now-grace underflows to the epoch. +func TestSweeperSweepsNothingBeforeTheFirstGraceWindowElapses(t *testing.T) { + t.Parallel() + + store := &fakeSweepStore{} + local := &fakeLocalStore{calls: &store.calls} + sweeper, err := s3keys.NewChunkBlobSweeper(s3keys.ChunkBlobSweeperOptions{ + Store: store, + Local: local, + GracePeriod: time.Hour, + NowTS: func() uint64 { return uint64(1_000) << 16 }, + }) + require.NoError(t, err) + + require.NoError(t, sweeper.SweepOnce(context.Background())) + require.Empty(t, store.calls, "nothing can have served a grace window yet") +} + +func TestSweeperPropagatesAScanFailure(t *testing.T) { + t.Parallel() + + boom := errors.New("store unavailable") + store := &fakeSweepStore{scanErr: boom} + sweeper, local := newSweeperFixture(t, store) + + err := sweeper.SweepOnce(context.Background()) + require.ErrorIs(t, err, boom) + require.Empty(t, local.deletes) +} + +func TestNewChunkBlobSweeperValidatesItsCollaborators(t *testing.T) { + t.Parallel() + + valid := s3keys.ChunkBlobSweeperOptions{ + Store: &fakeSweepStore{}, + Local: &fakeLocalStore{calls: &[]string{}}, + NowTS: func() uint64 { return 1 << 16 }, + } + + noStore := valid + noStore.Store = nil + _, err := s3keys.NewChunkBlobSweeper(noStore) + require.Error(t, err) + + noLocal := valid + noLocal.Local = nil + _, err = s3keys.NewChunkBlobSweeper(noLocal) + require.Error(t, err) + + noClock := valid + noClock.NowTS = nil + _, err = s3keys.NewChunkBlobSweeper(noClock) + require.Error(t, err, "an HLC clock is required; a wall clock would be the wrong domain") + + _, err = s3keys.NewChunkBlobSweeper(valid) + require.NoError(t, err) +}