diff --git a/ARCHITECTURE.md b/ARCHITECTURE.md index 00fa7bd7..7244ccd7 100644 --- a/ARCHITECTURE.md +++ b/ARCHITECTURE.md @@ -186,6 +186,11 @@ degrades into a silently partial result. part swept at the next open — never rows that are committed but unresolvable. The commit is a `backend.CompareAndSwap` against the version the writer read, so two writers over one prefix (a shared store) cannot overwrite each other's entries; the loser reloads and retries. +- **A part leaves the index's `Entries` only into `Removed` or into `Wanted`** — a tombstone (a + deliberate deletion) or a repair obligation, never silently. An owner that cannot open a part + drops it and records the want in the *same* commit, so the two halves cannot come apart; and only + a definitive `backend.ErrNotExist` qualifies, because a backend that merely failed to answer says + nothing about whether the data is there (`backend/ARCH.md`, `engine/ARCH.md`). - **A part id is globally unique** (both engines). A part's backend key is `{enginePrefix}/{partid}`, where `partid` is a minted ULID-shaped id (`internal/partid`), not a per-node counter: replicas of a shard can share one object store under one prefix, so only an id no local state participates in diff --git a/engine/ARCH.md b/engine/ARCH.md index c2d2ca9b..e2146fa6 100644 --- a/engine/ARCH.md +++ b/engine/ARCH.md @@ -201,6 +201,41 @@ name; a directory is a part's when its name parses as a part id. The sweep assum the prefix, so a replica's `RefreshReplica` skips it: the owner's in-flight part is not in the index yet. + +## A part the owner cannot read becomes a want, not a removal + +`LoadParts` used to fail the whole engine on the first part it could not open, which turns one lost +part into a node that will not start. It now drops that part from `Entries` and records a +`bucketindex.Want` naming it (`backend/ARCH.md`, "`Entries → Removed | Wanted`"), in **one** +compare-and-swap: the drop and the obligation are the same commit, so no crash can land the drop +without the want and a lost race leaves neither — the retry re-reads and re-derives both from the +same evidence. Nothing consumes wants yet; they are recorded and durable. + +**Only `backend.ErrNotExist` may become a want.** Every other failure — a timeout, a canceled +context, a full disk, a denied request, a throttled bucket — leaves the part's existence unknown, +and a want is a statement that it is gone. Recording one on an unknown would drop a live part from +the index and start a repair for data that was never missing; over a shared backend a single +transient fault touches every part at once, so the index would be stripped wholesale. So a +non-absence error fails the load exactly as it did before (`partGone`, the counterpart of +ClickHouse's `isRetryableException` rethrow in `checkDataPart`). + +The trigger stays narrow in the other direction too: a part whose objects are **present but +unreadable** — a corrupt manifest, a truncated column — is not a want. It is a different failure +with a different remedy, and widening the trigger is how a repair path turns into a +data-destruction path. It still fails the load. + +Two consequences fall out of the entry no longer being there: + +- The gone part is absent from `indexed`, so the next commit's diff does not also call it a + *removal*. A loss restated as a deliberate deletion is exactly the ambiguity wants exist to + remove. +- `foreignEntries` refuses to adopt it back from a rival's index, and the orphan sweep spares its + surviving objects. They are the remains of a part repair is owed, not the residue of a failed + flush, and deleting them destroys the evidence before repair can see it. + +Only an owner does this — the replica path (`RefreshReplica`, no sweep) still fails, because +recording a want is committing an index, which is the owner's to write. + ## Disk pressure closes the ingest path A flush that cannot fit is not a retryable failure. It detaches the head, fails part-way, burns a diff --git a/engine/engine.go b/engine/engine.go index dbbf969e..16eb5482 100644 --- a/engine/engine.go +++ b/engine/engine.go @@ -195,6 +195,12 @@ type Engine struct { // replica tell a compaction from a loss. indexed map[string]struct{} removals []bucketindex.Removal + // wants are the repair obligations the last bucket index this engine wrote carried, and + // pendingWants the ones a load discovered and the next commit must add — see [bucketindex.Want]. + // Keeping the discovery pending rather than committing it on the spot is what makes dropping + // the entry and recording the want one CAS commit instead of two. + wants []bucketindex.Want + pendingWants []bucketindex.Want // indexVersion is the backend version of the bucket index this engine last read or committed — // the token its next commit conditions on, so a rewrite that another writer got in front of is // refused rather than silently overwriting it (#392). diff --git a/engine/index.go b/engine/index.go index f374a725..dbb2cc44 100644 --- a/engine/index.go +++ b/engine/index.go @@ -49,7 +49,7 @@ func (e *Engine) updateIndexLocked(ctx context.Context) error { } for range indexCommitAttempts { - version, err := e.nextIndexLocked().Save(ctx, e.cfg.Backend, e.indexKey(), e.indexVersion) + version, err := e.nextIndexLocked(ctx).Save(ctx, e.cfg.Backend, e.indexKey(), e.indexVersion) if err == nil { e.indexVersion = version @@ -88,7 +88,7 @@ func (e *Engine) adoptIndexLocked(ctx context.Context) error { // flushes and nothing another writer committed can say anything about it. e.epochs, e.anonEpoch = ix.Epochs, ix.FlushedEpoch - e.foreign = foreignEntries(ix.Entries, e.indexed, e.removals) + e.foreign = foreignEntries(ix.Entries, e.indexed, e.removals, e.wants) e.openForeignLocked(ctx) return nil @@ -161,7 +161,7 @@ func (e *Engine) readablePartsLocked() []*part { // nextIndexLocked builds the index state this engine wants committed and advances the // bookkeeping the next one is diffed against. Caller holds e.mu. -func (e *Engine) nextIndexLocked() *bucketindex.Index { +func (e *Engine) nextIndexLocked(ctx context.Context) *bucketindex.Index { // The generation advances on every write, including one that only removes parts — which is // the whole point of it, since that is exactly the rewrite the part names cannot express. e.generation = e.generation.Next(e.term()) @@ -203,6 +203,27 @@ func (e *Engine) nextIndexLocked() *bucketindex.Index { ix.Removed = e.removals e.indexed = live + // The same commit that drops an unreadable part from Entries states the obligation to get it + // back: a part leaves Entries only into Removed or into Wanted, and one CAS commit carries + // both halves, so no crash can land the drop without the want. + ix.Wanted = slices.Clone(e.wants) + + for _, w := range e.pendingWants { + w.Generation = e.generation + ix.RecordWant(w) + } + + e.pendingWants = nil + + kept, dropped := bucketindex.TrimWants(ix.Wanted, ix.Entries, bucketindex.MaxWants) + if len(dropped) > 0 { + zctx.From(ctx).Warn("outstanding repairs exceed the index bound", + zap.String("prefix", e.cfg.Prefix), zap.Int("dropped", len(dropped)), + zap.Int("kept", len(kept))) + } + + ix.Wanted, e.wants = kept, kept + return ix } @@ -211,7 +232,8 @@ func (e *Engine) nextIndexLocked() *bucketindex.Index { // still live, so this engine's job is to carry them across its own commits — an entry dropped // here leaves durable part objects unreferenced, and the next open-time orphan sweep deletes them. func foreignEntries( - entries []bucketindex.Entry, indexed map[string]struct{}, removals []bucketindex.Removal, + entries []bucketindex.Entry, indexed map[string]struct{}, + removals []bucketindex.Removal, wants []bucketindex.Want, ) []bucketindex.Entry { var out []bucketindex.Entry @@ -224,6 +246,13 @@ func foreignEntries( continue } + // A part this engine already found unreadable is not adopted back: carrying its entry + // forward would put it in Entries again and discharge the want that names it, leaving the + // index pointing at bytes no one has. + if slices.ContainsFunc(wants, func(w bucketindex.Want) bool { return w.Prefix == e.Prefix }) { + continue + } + out = append(out, e) } @@ -269,10 +298,21 @@ func (e *Engine) loadPartsLocked(ctx context.Context, sweep bool) error { parts := make([]*part, 0, len(ix.Entries)) + var lost []bucketindex.Want + for _, ent := range ix.Entries { p, err := openPart(ctx, e.cfg.Backend, ent.Prefix) if err != nil { - return errors.Wrapf(err, "open part %q", ent.Prefix) + if !sweep || !partGone(err) { + return errors.Wrapf(err, "open part %q", ent.Prefix) + } + + zctx.From(ctx).Error("part named by the index is gone; recording a repair", + zap.String("prefix", ent.Prefix), zap.Error(err)) + + lost = append(lost, bucketindex.Want{Prefix: ent.Prefix, Blocks: ent.Blocks}) + + continue } p.minTime, p.maxTime = ent.MinTime, ent.MaxTime @@ -294,10 +334,13 @@ func (e *Engine) loadPartsLocked(ctx context.Context, sweep bool) error { e.epochs, e.anonEpoch = ix.Epochs, ix.FlushedEpoch e.generation = ix.Generation e.removals = ix.Removed - e.indexed = make(map[string]struct{}, len(ix.Entries)) + e.wants, e.pendingWants = ix.Wanted, lost + e.indexed = make(map[string]struct{}, len(parts)) - for _, entry := range ix.Entries { - e.indexed[entry.Prefix] = struct{}{} + // Only the parts that opened: a lost one left out of e.indexed is what keeps the commit below + // from also calling it a removal, which would restate a loss as a deliberate deletion. + for _, p := range parts { + e.indexed[p.prefix] = struct{}{} } // New head records belong to the generation past the recovered watermark; replay (which the @@ -312,6 +355,14 @@ func (e *Engine) loadPartsLocked(ctx context.Context, sweep bool) error { } } + // One commit drops the gone parts from Entries and states the wants that replace them. It runs + // before the rest of the load so the obligation is durable even if identity recovery fails. + if len(lost) > 0 { + if err := e.updateIndexLocked(ctx); err != nil { + return errors.Wrap(err, "record repair wants") + } + } + complete, err := e.loadIdentitiesLocked(ctx, parts) if err != nil { return err diff --git a/engine/lost.go b/engine/lost.go new file mode 100644 index 00000000..c4ebdfb3 --- /dev/null +++ b/engine/lost.go @@ -0,0 +1,14 @@ +package engine + +import ( + "github.com/go-faster/errors" + + "github.com/oteldb/storage/backend" +) + +// partGone reports whether err is the backend saying the object is not there, as opposed to +// saying nothing at all. Only that answer may turn an index entry into a repair obligation: +// every other failure — a timeout, a canceled context, a full disk, a denied request, a +// throttled bucket — leaves the part's existence unknown, and treating one as loss would strip +// live parts out of the index, shard-wide when the backend is shared. +func partGone(err error) bool { return errors.Is(err, backend.ErrNotExist) } diff --git a/engine/lostpart_test.go b/engine/lostpart_test.go new file mode 100644 index 00000000..5dff9972 --- /dev/null +++ b/engine/lostpart_test.go @@ -0,0 +1,462 @@ +package engine_test + +import ( + "context" + "maps" + "math/rand/v2" + "slices" + "strings" + "sync/atomic" + "testing" + + "github.com/go-faster/errors" + "github.com/stretchr/testify/assert" + "github.com/stretchr/testify/require" + + "github.com/oteldb/storage/backend" + "github.com/oteldb/storage/backend/bucketindex" + "github.com/oteldb/storage/engine" + "github.com/oteldb/storage/internal/partid" + "github.com/oteldb/storage/query/fetch" + "github.com/oteldb/storage/signal" +) + +const lostPrefix = "default/metrics" + +func lostIndexKey() string { return lostPrefix + "/" + bucketindex.Object } + +func prefixes(entries []bucketindex.Entry) []string { + out := make([]string, len(entries)) + for i, e := range entries { + out[i] = e.Prefix + } + + return out +} + +func wantPrefixes(wants []bucketindex.Want) []string { + out := make([]string, len(wants)) + for i, w := range wants { + out[i] = w.Prefix + } + + return out +} + +// diskPartKeys returns the backend objects of the part with the given id. +func diskPartKeys(ctx context.Context, t *testing.T, be backend.Backend, id string) []string { + t.Helper() + + keys, err := be.List(ctx, lostPrefix+"/"+id+"/") + require.NoError(t, err) + + return keys +} + +// diskPartIDs returns the sorted ids of the parts that have objects under the engine prefix — part ids +// are minted, so a test cannot name them up front and asserts over this set instead. +func diskPartIDs(ctx context.Context, t *testing.T, be backend.Backend) []string { + t.Helper() + + keys, err := be.List(ctx, lostPrefix+"/") + require.NoError(t, err) + + seen := make(map[string]struct{}, len(keys)) + + for _, k := range keys { + if dir, _, ok := strings.Cut(strings.TrimPrefix(k, lostPrefix+"/"), "/"); ok && partid.Valid(dir) { + seen[dir] = struct{}{} + } + } + + return slices.Sorted(maps.Keys(seen)) +} + +// erasePart deletes every backend object of the part with the given id, the disk failure a repair +// exists for. +func erasePart(ctx context.Context, t *testing.T, be backend.Backend, id string) { + t.Helper() + + for _, k := range diskPartKeys(ctx, t, be, id) { + require.NoError(t, be.Delete(ctx, k)) + } +} + +// failReads wraps a backend and answers Read for keys with a given suffix with err — an injected +// *transient* failure when err is anything but [backend.ErrNotExist]. +type failReads struct { + backend.Backend + + armed atomic.Bool + only string + err error +} + +func (f *failReads) Read(ctx context.Context, key string) ([]byte, error) { + if f.armed.Load() && strings.HasSuffix(key, f.only) { + return nil, f.err + } + + return f.Backend.Read(ctx, key) +} + +// countCAS wraps a backend and counts the compare-and-swaps against one key, optionally refusing +// them all so a caller sees a permanently lost race. +type countCAS struct { + backend.Backend + + key string + n atomic.Int64 + refuse atomic.Bool +} + +func (c *countCAS) CompareAndSwap( + ctx context.Context, key string, expected backend.Version, data []byte, +) (backend.Version, bool, error) { + if key != c.key { + return c.Backend.CompareAndSwap(ctx, key, expected, data) + } + + c.n.Add(1) + + if c.refuse.Load() { + return backend.VersionAbsent, false, nil + } + + return c.Backend.CompareAndSwap(ctx, key, expected, data) +} + +func newLostEngine(be backend.Backend) *engine.Engine { + return engine.New(engine.Config{Backend: be, Prefix: lostPrefix}) +} + +// twoParts flushes two parts and returns their ids in index order. +func twoParts(t *testing.T, e *engine.Engine, be backend.Backend, s signal.Series) []string { + t.Helper() + + ctx := context.Background() + + mustAppend(t, e, s, 100, 1.0) + require.NoError(t, e.Flush(ctx)) + mustAppend(t, e, s, 200, 2.0) + require.NoError(t, e.Flush(ctx)) + + ids := diskPartIDs(ctx, t, be) + require.Len(t, ids, 2) + + return ids +} + +func lostValues(t *testing.T, e *engine.Engine) []float64 { + t.Helper() + + var out []float64 + + for _, b := range fetchAll(t, e, fetch.Request{ + Matchers: []fetch.Matcher{eqMatcher("job", "api")}, + Start: 0, End: 1 << 60, + }) { + out = append(out, b.Values...) + } + + slices.Sort(out) + + return out +} + +// TestGonePartBecomesWant is the whole of step 2: a part whose objects are permanently absent no +// longer fails the load — it leaves Entries into Wanted, the rest of the part set still serves, and +// the obligation is in the committed index. +func TestGonePartBecomesWant(t *testing.T) { + t.Parallel() + + ctx := context.Background() + be := backend.Memory() + s := mkSeries("job", "api") + e := newLostEngine(be) + ids := twoParts(t, e, be, s) + + erasePart(ctx, t, be, ids[0]) + + r := newLostEngine(be) + require.NoError(t, r.LoadParts(ctx), "one gone part must not fail the load") + require.Equal(t, 1, r.PartCount()) + require.Equal(t, []float64{2.0}, lostValues(t, r)) + + ix := loadIndex(t, be, lostPrefix) + require.Equal(t, []string{lostPrefix + "/" + ids[1]}, prefixes(ix.Entries)) + require.Equal(t, []string{lostPrefix + "/" + ids[0]}, wantPrefixes(ix.Wanted)) + require.Empty(t, ix.Removed, "a loss is not a removal") + require.NotZero(t, ix.Wanted[0].Generation, "the want records when it was discovered") +} + +// TestPartiallyGonePartBecomesWant covers the half-erased part: the manifest is there and nothing +// it names is. It cannot answer a read, so it is as lost as one with no objects at all — and +// the open-time orphan sweep spares what is left of it, because those objects are the remains of a +// part repair is owed rather than the residue of a failed flush. +func TestPartiallyGonePartBecomesWant(t *testing.T) { + t.Parallel() + + ctx := context.Background() + be := backend.Memory() + e := newLostEngine(be) + ids := twoParts(t, e, be, mkSeries("job", "api")) + + for _, k := range diskPartKeys(ctx, t, be, ids[0]) { + if !strings.HasSuffix(k, "/manifest") { + require.NoError(t, be.Delete(ctx, k)) + } + } + + r := newLostEngine(be) + require.NoError(t, r.LoadParts(ctx)) + + ix := loadIndex(t, be, lostPrefix) + require.Equal(t, []string{lostPrefix + "/" + ids[0]}, wantPrefixes(ix.Wanted)) + require.Equal(t, []string{lostPrefix + "/" + ids[1]}, prefixes(ix.Entries)) + require.NotEmpty(t, diskPartKeys(ctx, t, be, ids[0]), "a wanted part's remaining objects are not orphans") +} + +// TestTransientOpenFailureRecordsNoWant is the guard the whole step turns on: a backend that could +// not answer says nothing about whether the part exists, so it fails the load exactly as before and +// records nothing. Converting one into a want would drop a live part and start a repair for data +// that was never missing — shard-wide, when many parts share the failing backend. +func TestTransientOpenFailureRecordsNoWant(t *testing.T) { + t.Parallel() + + for _, tc := range []struct { + name string + err error + }{ + {"timeout", errors.New("i/o timeout")}, + {"canceled", context.Canceled}, + {"no space", errors.New("no space left on device")}, + {"denied", errors.New("AccessDenied")}, + {"throttled", errors.New("SlowDown: please reduce your request rate")}, + } { + t.Run(tc.name, func(t *testing.T) { + t.Parallel() + + ctx := context.Background() + be := &failReads{Backend: backend.Memory(), only: "/manifest", err: tc.err} + e := newLostEngine(be) + ids := twoParts(t, e, be, mkSeries("job", "api")) + + before := loadIndex(t, be, lostPrefix) + + be.armed.Store(true) + err := newLostEngine(be).LoadParts(ctx) + be.armed.Store(false) + + after := loadIndex(t, be, lostPrefix) + assert.Empty(t, after.Wanted, "a transient failure is not evidence of loss") + assert.Equal(t, prefixes(before.Entries), prefixes(after.Entries), + "a transient failure must not strip live parts from the index") + assert.Len(t, after.Entries, 2) + assert.Equal(t, before.Generation, after.Generation, "nothing was committed") + assert.NotEmpty(t, diskPartKeys(ctx, t, be, ids[0])) + assert.Error(t, err, "a transient failure must still fail the load") + }) + } +} + +// TestCorruptPartIsNotAWant pins the deliberately narrow trigger: objects that are present but +// unreadable are a different failure with a different remedy, and widening the trigger to cover +// them is how a repair path becomes a data-destruction path. +func TestCorruptPartIsNotAWant(t *testing.T) { + t.Parallel() + + ctx := context.Background() + be := backend.Memory() + e := newLostEngine(be) + ids := twoParts(t, e, be, mkSeries("job", "api")) + + objs := diskPartKeys(ctx, t, be, ids[0]) + i := slices.IndexFunc(objs, func(k string) bool { return strings.HasSuffix(k, "/manifest") }) + require.GreaterOrEqual(t, i, 0) + require.NoError(t, be.Write(ctx, objs[i], []byte("not a manifest"))) + + require.Error(t, newLostEngine(be).LoadParts(ctx)) + require.Empty(t, loadIndex(t, be, lostPrefix).Wanted) +} + +// TestWantIsRecordedInOneCommit verifies the atomicity by construction: dropping the entry and +// recording the want is a single compare-and-swap, so there is no window in which a crash could +// land one without the other. +func TestWantIsRecordedInOneCommit(t *testing.T) { + t.Parallel() + + ctx := context.Background() + be := &countCAS{Backend: backend.Memory(), key: lostIndexKey()} + e := newLostEngine(be) + ids := twoParts(t, e, be, mkSeries("job", "api")) + + erasePart(ctx, t, be, ids[0]) + + be.n.Store(0) + require.NoError(t, newLostEngine(be).LoadParts(ctx)) + + require.Equal(t, int64(1), be.n.Load(), "the drop and the want are one commit, not two") + + ix := loadIndex(t, be, lostPrefix) + require.NotContains(t, prefixes(ix.Entries), lostPrefix+"/"+ids[0]) + require.Contains(t, wantPrefixes(ix.Wanted), lostPrefix+"/"+ids[0]) +} + +// TestFailedWantCommitAppliesNeither verifies a lost race leaves the stored index untouched: the +// entry is still there and no want is, so the retry re-reads and re-derives both halves rather +// than inheriting half of an applied commit. +func TestFailedWantCommitAppliesNeither(t *testing.T) { + t.Parallel() + + ctx := context.Background() + be := &countCAS{Backend: backend.Memory(), key: lostIndexKey()} + e := newLostEngine(be) + ids := twoParts(t, e, be, mkSeries("job", "api")) + + before := loadIndex(t, be, lostPrefix) + erasePart(ctx, t, be, ids[0]) + + be.refuse.Store(true) + require.Error(t, newLostEngine(be).LoadParts(ctx), + "a commit that never lands must not be reported as success") + be.refuse.Store(false) + + after := loadIndex(t, be, lostPrefix) + require.Equal(t, prefixes(before.Entries), prefixes(after.Entries), "the entry is still there") + require.Empty(t, after.Wanted, "and the want is not") + + // The retry re-derives both halves from the same evidence and commits them together. + require.NoError(t, newLostEngine(be).LoadParts(ctx)) + + ix := loadIndex(t, be, lostPrefix) + require.Equal(t, []string{lostPrefix + "/" + ids[1]}, prefixes(ix.Entries)) + require.Equal(t, []string{lostPrefix + "/" + ids[0]}, wantPrefixes(ix.Wanted)) +} + +// TestWantSurvivesLaterCommits verifies an obligation is durable: it is carried through every later +// index this engine writes, so a crash between recording a want and satisfying it is benign. +func TestWantSurvivesLaterCommits(t *testing.T) { + t.Parallel() + + ctx := context.Background() + be := backend.Memory() + s := mkSeries("job", "api") + e := newLostEngine(be) + ids := twoParts(t, e, be, s) + + erasePart(ctx, t, be, ids[0]) + + r := newLostEngine(be) + require.NoError(t, r.LoadParts(ctx)) + + mustAppend(t, r, s, 300, 3.0) + require.NoError(t, r.Flush(ctx)) + require.NoError(t, r.Merge(ctx, 0)) + + require.Equal(t, []string{lostPrefix + "/" + ids[0]}, wantPrefixes(loadIndex(t, be, lostPrefix).Wanted)) + + // And across a restart, which reads it back off the index rather than rediscovering it. + require.NoError(t, newLostEngine(be).LoadParts(ctx)) + require.Equal(t, []string{lostPrefix + "/" + ids[0]}, wantPrefixes(loadIndex(t, be, lostPrefix).Wanted)) +} + +// invariantCAS asserts, on every index commit that lands, that no part vanished from Entries +// without landing in exactly one of Removed and Wanted. It sits on the real commit path, so every +// flush, merge, retention drop and lossy load in a test is checked. +type invariantCAS struct { + backend.Backend + + t *testing.T + key string +} + +func (c *invariantCAS) CompareAndSwap( + ctx context.Context, key string, expected backend.Version, data []byte, +) (backend.Version, bool, error) { + if key != c.key { + return c.Backend.CompareAndSwap(ctx, key, expected, data) + } + + prev, err := bucketindex.Load(ctx, c.Backend, key) + require.NoError(c.t, err) + + version, ok, err := c.Backend.CompareAndSwap(ctx, key, expected, data) + if err != nil || !ok { + return version, ok, err + } + + next, decErr := bucketindex.Decode(data) + require.NoError(c.t, decErr) + assertLeavesIntoOneList(c.t, prev, next) + + return version, ok, err +} + +func assertLeavesIntoOneList(t *testing.T, prev, next *bucketindex.Index) { + t.Helper() + + live := prefixes(next.Entries) + + for _, p := range prefixes(prev.Entries) { + if slices.Contains(live, p) { + continue + } + + removed := slices.ContainsFunc(next.Removed, func(r bucketindex.Removal) bool { return r.Prefix == p }) + wanted := slices.ContainsFunc(next.Wanted, func(w bucketindex.Want) bool { return w.Prefix == p }) + + switch { + case removed && wanted: + require.Fail(t, "a part left Entries into both Removed and Wanted", "part %q", p) + case !removed && !wanted: + require.Fail(t, "a part vanished from Entries into neither Removed nor Wanted", "part %q", p) + } + } +} + +// TestEntriesLeaveOnlyIntoRemovedOrWanted is the design in one line, as a property over the real +// commit path: random append, flush, merge, retention and part-loss sequences, with every committed +// index checked against the one it replaced. +func TestEntriesLeaveOnlyIntoRemovedOrWanted(t *testing.T) { + t.Parallel() + + ctx := context.Background() + + for seed := range uint64(24) { + t.Run("", func(t *testing.T) { + t.Parallel() + + rng := rand.New(rand.NewPCG(seed, 0x5eed)) + be := &invariantCAS{Backend: backend.Memory(), t: t, key: lostIndexKey()} + s := mkSeries("job", "api") + e := newLostEngine(be) + ts := int64(100) + + for step := range 12 { + ts += 100 + + switch rng.IntN(5) { + case 0, 1: + mustAppend(t, e, s, ts, float64(step)) + require.NoError(t, e.Flush(ctx)) + case 2: + require.NoError(t, e.Merge(ctx, 0)) + case 3: + // Retention: parts wholly below the horizon are dropped, not merged. + require.NoError(t, e.Merge(ctx, ts-150)) + case 4: + ids := diskPartIDs(ctx, t, be) + if len(ids) == 0 { + continue + } + + erasePart(ctx, t, be, ids[rng.IntN(len(ids))]) + + e = newLostEngine(be) + require.NoError(t, e.LoadParts(ctx), "step %d", step) + } + } + }) + } +} diff --git a/engine/reclaim.go b/engine/reclaim.go index a6add99a..d9842058 100644 --- a/engine/reclaim.go +++ b/engine/reclaim.go @@ -86,6 +86,17 @@ func (e *Engine) sweepOrphansLocked(ctx context.Context) error { live[p.prefix] = struct{}{} } + // A wanted part's objects are not orphans, they are what is left of it. Deleting the readable + // remainder of a part repair is about to fetch destroys evidence and buys nothing: the want + // bounds how long it is kept, and a discharged want puts the prefix back in the live set. + for _, w := range e.wants { + live[w.Prefix] = struct{}{} + } + + for _, w := range e.pendingWants { + live[w.Prefix] = struct{}{} + } + var orphans []string for _, k := range keys { diff --git a/recordengine/ARCH.md b/recordengine/ARCH.md index 5d267ae7..dc28f452 100644 --- a/recordengine/ARCH.md +++ b/recordengine/ARCH.md @@ -226,6 +226,12 @@ Reuse would be unsound here for one extra reason: two of a part's objects are co is skipped when the rows carry no record attributes, the `sym-*.bin` sidecars when there is no side data — so a new part would silently adopt a failed attempt's. +A part the owner cannot open is handled identically: only `backend.ErrNotExist` drops it from +`Entries` and records a `bucketindex.Want` in the same compare-and-swap, every other error still +fails the load, and the sweep spares a wanted part's remaining objects +([`../engine/ARCH.md`](../engine/ARCH.md), "A part the owner cannot read becomes a want, not a +removal"). + ## Lifecycle guards Flush and merge run under one `flushMu`, and `Reset` takes it too, with the same rationale and the same diff --git a/recordengine/engine.go b/recordengine/engine.go index c0b39be6..711ff6a7 100644 --- a/recordengine/engine.go +++ b/recordengine/engine.go @@ -168,6 +168,12 @@ type Engine struct { // replica tell a compaction from a loss. indexed map[string]struct{} removals []bucketindex.Removal + // wants are the repair obligations the last bucket index this engine wrote carried, and + // pendingWants the ones a load discovered and the next commit must add — see [bucketindex.Want]. + // Keeping the discovery pending rather than committing it on the spot is what makes dropping + // the entry and recording the want one CAS commit instead of two. + wants []bucketindex.Want + pendingWants []bucketindex.Want // indexVersion is the backend version of the bucket index this engine last read or committed — // the token its next commit conditions on, so a rewrite that another writer got in front of is // refused rather than silently overwriting it (#392). diff --git a/recordengine/index.go b/recordengine/index.go index b1ef1a15..61148889 100644 --- a/recordengine/index.go +++ b/recordengine/index.go @@ -42,7 +42,7 @@ func (e *Engine) updateIndexLocked(ctx context.Context) error { } for range indexCommitAttempts { - version, err := e.nextIndexLocked().Save(ctx, e.cfg.Backend, e.indexKey(), e.indexVersion) + version, err := e.nextIndexLocked(ctx).Save(ctx, e.cfg.Backend, e.indexKey(), e.indexVersion) if err == nil { e.indexVersion = version @@ -81,7 +81,7 @@ func (e *Engine) adoptIndexLocked(ctx context.Context) error { // flushes and nothing another writer committed can say anything about it. e.epochs, e.anonEpoch = ix.Epochs, ix.FlushedEpoch - e.foreign = foreignEntries(ix.Entries, e.indexed, e.removals) + e.foreign = foreignEntries(ix.Entries, e.indexed, e.removals, e.wants) e.openForeignLocked(ctx) return nil @@ -154,7 +154,7 @@ func (e *Engine) readablePartsLocked() []*part { // nextIndexLocked builds the index state this engine wants committed and advances the // bookkeeping the next one is diffed against. Caller holds e.mu. -func (e *Engine) nextIndexLocked() *bucketindex.Index { +func (e *Engine) nextIndexLocked(ctx context.Context) *bucketindex.Index { // The generation advances on every write, including one that only removes parts — which is // the whole point of it, since that is exactly the rewrite the part names cannot express. e.generation = e.generation.Next(e.term()) @@ -196,6 +196,27 @@ func (e *Engine) nextIndexLocked() *bucketindex.Index { ix.Removed = e.removals e.indexed = live + // The same commit that drops an unreadable part from Entries states the obligation to get it + // back: a part leaves Entries only into Removed or into Wanted, and one CAS commit carries + // both halves, so no crash can land the drop without the want. + ix.Wanted = slices.Clone(e.wants) + + for _, w := range e.pendingWants { + w.Generation = e.generation + ix.RecordWant(w) + } + + e.pendingWants = nil + + kept, dropped := bucketindex.TrimWants(ix.Wanted, ix.Entries, bucketindex.MaxWants) + if len(dropped) > 0 { + zctx.From(ctx).Warn("outstanding repairs exceed the index bound", + zap.String("prefix", e.cfg.Prefix), zap.Int("dropped", len(dropped)), + zap.Int("kept", len(kept))) + } + + ix.Wanted, e.wants = kept, kept + return ix } @@ -204,7 +225,8 @@ func (e *Engine) nextIndexLocked() *bucketindex.Index { // still live, so this engine's job is to carry them across its own commits — an entry dropped // here leaves durable part objects unreferenced, and the next open-time orphan sweep deletes them. func foreignEntries( - entries []bucketindex.Entry, indexed map[string]struct{}, removals []bucketindex.Removal, + entries []bucketindex.Entry, indexed map[string]struct{}, + removals []bucketindex.Removal, wants []bucketindex.Want, ) []bucketindex.Entry { var out []bucketindex.Entry @@ -217,6 +239,13 @@ func foreignEntries( continue } + // A part this engine already found unreadable is not adopted back: carrying its entry + // forward would put it in Entries again and discharge the want that names it, leaving the + // index pointing at bytes no one has. + if slices.ContainsFunc(wants, func(w bucketindex.Want) bool { return w.Prefix == e.Prefix }) { + continue + } + out = append(out, e) } @@ -257,10 +286,21 @@ func (e *Engine) loadPartsLocked(ctx context.Context, sweep bool) error { parts := make([]*part, 0, len(ix.Entries)) + var lost []bucketindex.Want + for _, ent := range ix.Entries { p, err := openPart(ctx, e.cfg.Backend, e.cfg.Schema, ent.Prefix) if err != nil { - return errors.Wrapf(err, "open part %q", ent.Prefix) + if !sweep || !partGone(err) { + return errors.Wrapf(err, "open part %q", ent.Prefix) + } + + zctx.From(ctx).Error("part named by the index is gone; recording a repair", + zap.String("prefix", ent.Prefix), zap.Error(err)) + + lost = append(lost, bucketindex.Want{Prefix: ent.Prefix, Blocks: ent.Blocks}) + + continue } p.minTime, p.maxTime = ent.MinTime, ent.MaxTime @@ -282,10 +322,13 @@ func (e *Engine) loadPartsLocked(ctx context.Context, sweep bool) error { e.epochs, e.anonEpoch = ix.Epochs, ix.FlushedEpoch e.generation = ix.Generation e.removals = ix.Removed - e.indexed = make(map[string]struct{}, len(ix.Entries)) + e.wants, e.pendingWants = ix.Wanted, lost + e.indexed = make(map[string]struct{}, len(parts)) - for _, entry := range ix.Entries { - e.indexed[entry.Prefix] = struct{}{} + // Only the parts that opened: a lost one left out of e.indexed is what keeps the commit below + // from also calling it a removal, which would restate a loss as a deliberate deletion. + for _, p := range parts { + e.indexed[p.prefix] = struct{}{} } if sweep { @@ -294,6 +337,14 @@ func (e *Engine) loadPartsLocked(ctx context.Context, sweep bool) error { } } + // One commit drops the gone parts from Entries and states the wants that replace them. It runs + // before the rest of the load so the obligation is durable even if identity recovery fails. + if len(lost) > 0 { + if err := e.updateIndexLocked(ctx); err != nil { + return errors.Wrap(err, "record repair wants") + } + } + // New head records belong to the generation past the recovered watermark; replay (which the // facade runs next) then skips everything the loaded parts already hold. if e.cfg.WAL != nil { diff --git a/recordengine/lost.go b/recordengine/lost.go new file mode 100644 index 00000000..28ca1285 --- /dev/null +++ b/recordengine/lost.go @@ -0,0 +1,14 @@ +package recordengine + +import ( + "github.com/go-faster/errors" + + "github.com/oteldb/storage/backend" +) + +// partGone reports whether err is the backend saying the object is not there, as opposed to +// saying nothing at all. Only that answer may turn an index entry into a repair obligation: +// every other failure — a timeout, a canceled context, a full disk, a denied request, a +// throttled bucket — leaves the part's existence unknown, and treating one as loss would strip +// live parts out of the index, shard-wide when the backend is shared. +func partGone(err error) bool { return errors.Is(err, backend.ErrNotExist) } diff --git a/recordengine/lostpart_test.go b/recordengine/lostpart_test.go new file mode 100644 index 00000000..4f6278a0 --- /dev/null +++ b/recordengine/lostpart_test.go @@ -0,0 +1,451 @@ +package recordengine_test + +import ( + "context" + "maps" + "math/rand/v2" + "slices" + "strings" + "sync/atomic" + "testing" + + "github.com/go-faster/errors" + "github.com/stretchr/testify/assert" + "github.com/stretchr/testify/require" + + "github.com/oteldb/storage/backend" + "github.com/oteldb/storage/backend/bucketindex" + "github.com/oteldb/storage/internal/partid" + "github.com/oteldb/storage/recordengine" +) + +const enginePrefix = "t/recs" + +func indexKey() string { return enginePrefix + "/" + bucketindex.Object } + +// loadIndex returns the committed bucket index under the engine prefix. +func loadIndex(t *testing.T, be backend.Backend) *bucketindex.Index { + t.Helper() + + ix, err := bucketindex.Load(context.Background(), be, indexKey()) + require.NoError(t, err) + + return ix +} + +func prefixes(entries []bucketindex.Entry) []string { + out := make([]string, len(entries)) + for i, e := range entries { + out[i] = e.Prefix + } + + return out +} + +func wantPrefixes(wants []bucketindex.Want) []string { + out := make([]string, len(wants)) + for i, w := range wants { + out[i] = w.Prefix + } + + return out +} + +// diskPartKeys returns the backend objects of the part with the given id. +func diskPartKeys(ctx context.Context, t *testing.T, be backend.Backend, id string) []string { + t.Helper() + + keys, err := be.List(ctx, enginePrefix+"/"+id+"/") + require.NoError(t, err) + + return keys +} + +// diskPartIDs returns the sorted ids of the parts that have objects under the engine prefix — part ids +// are minted, so a test cannot name them up front and asserts over this set instead. +func diskPartIDs(ctx context.Context, t *testing.T, be backend.Backend) []string { + t.Helper() + + keys, err := be.List(ctx, enginePrefix+"/") + require.NoError(t, err) + + seen := make(map[string]struct{}, len(keys)) + + for _, k := range keys { + if dir, _, ok := strings.Cut(strings.TrimPrefix(k, enginePrefix+"/"), "/"); ok && partid.Valid(dir) { + seen[dir] = struct{}{} + } + } + + return slices.Sorted(maps.Keys(seen)) +} + +// erasePart deletes every backend object of the part with the given id, the disk failure a repair +// exists for. +func erasePart(ctx context.Context, t *testing.T, be backend.Backend, id string) { + t.Helper() + + for _, k := range diskPartKeys(ctx, t, be, id) { + require.NoError(t, be.Delete(ctx, k)) + } +} + +// failReads wraps a backend and answers Read for keys with a given suffix with err — an injected +// *transient* failure when err is anything but [backend.ErrNotExist]. +type failReads struct { + backend.Backend + + armed atomic.Bool + only string + err error +} + +func (f *failReads) Read(ctx context.Context, key string) ([]byte, error) { + if f.armed.Load() && strings.HasSuffix(key, f.only) { + return nil, f.err + } + + return f.Backend.Read(ctx, key) +} + +// countCAS wraps a backend and counts the compare-and-swaps against one key, optionally refusing +// them all so a caller sees a permanently lost race. +type countCAS struct { + backend.Backend + + key string + n atomic.Int64 + refuse atomic.Bool +} + +func (c *countCAS) CompareAndSwap( + ctx context.Context, key string, expected backend.Version, data []byte, +) (backend.Version, bool, error) { + if key != c.key { + return c.Backend.CompareAndSwap(ctx, key, expected, data) + } + + c.n.Add(1) + + if c.refuse.Load() { + return backend.VersionAbsent, false, nil + } + + return c.Backend.CompareAndSwap(ctx, key, expected, data) +} + +// twoParts flushes two parts and returns their ids in index order. +func twoParts(t *testing.T, e *recordengine.Engine, be backend.Backend) []string { + t.Helper() + + ctx := context.Background() + + ingest(t, e, mkBatch("api", rrec{ts: 100, body: "p1"})) + require.NoError(t, e.Flush(ctx)) + ingest(t, e, mkBatch("api", rrec{ts: 200, body: "p2"})) + require.NoError(t, e.Flush(ctx)) + + ids := diskPartIDs(ctx, t, be) + require.Len(t, ids, 2) + + return ids +} + +// TestGonePartBecomesWant is the whole of step 2: a part whose objects are permanently absent no +// longer fails the load — it leaves Entries into Wanted, the rest of the part set still serves, and +// the obligation is in the committed index. +func TestGonePartBecomesWant(t *testing.T) { + t.Parallel() + + ctx := context.Background() + be := backend.Memory() + e := newEngine(t, be) + ids := twoParts(t, e, be) + + erasePart(ctx, t, be, ids[0]) + + r := newEngine(t, be) + require.NoError(t, r.LoadParts(ctx), "one gone part must not fail the load") + require.Equal(t, 1, r.PartCount()) + require.Equal(t, []string{"p2"}, streamBodies(t, r)) + + ix := loadIndex(t, be) + require.Equal(t, []string{enginePrefix + "/" + ids[1]}, prefixes(ix.Entries)) + require.Equal(t, []string{enginePrefix + "/" + ids[0]}, wantPrefixes(ix.Wanted)) + require.Empty(t, ix.Removed, "a loss is not a removal") + require.NotZero(t, ix.Wanted[0].Generation, "the want records when it was discovered") +} + +// TestPartiallyGonePartBecomesWant covers the half-erased part: the manifest is there and the +// column data is not. It cannot answer a read, so it is as lost as one with no objects at all — and +// the open-time orphan sweep spares what is left of it, because those objects are the remains of a +// part repair is owed rather than the residue of a failed flush. +func TestPartiallyGonePartBecomesWant(t *testing.T) { + t.Parallel() + + ctx := context.Background() + be := backend.Memory() + e := newEngine(t, be) + ids := twoParts(t, e, be) + + for _, k := range diskPartKeys(ctx, t, be, ids[0]) { + if strings.Contains(k, "/c/") { + require.NoError(t, be.Delete(ctx, k)) + } + } + + r := newEngine(t, be) + require.NoError(t, r.LoadParts(ctx)) + + ix := loadIndex(t, be) + require.Equal(t, []string{enginePrefix + "/" + ids[0]}, wantPrefixes(ix.Wanted)) + require.Equal(t, []string{enginePrefix + "/" + ids[1]}, prefixes(ix.Entries)) + require.NotEmpty(t, diskPartKeys(ctx, t, be, ids[0]), "a wanted part's remaining objects are not orphans") +} + +// TestTransientOpenFailureRecordsNoWant is the guard the whole step turns on: a backend that could +// not answer says nothing about whether the part exists, so it fails the load exactly as before and +// records nothing. Converting one into a want would drop a live part and start a repair for data +// that was never missing — shard-wide, when many parts share the failing backend. +func TestTransientOpenFailureRecordsNoWant(t *testing.T) { + t.Parallel() + + for _, tc := range []struct { + name string + err error + }{ + {"timeout", errors.New("i/o timeout")}, + {"canceled", context.Canceled}, + {"no space", errors.New("no space left on device")}, + {"denied", errors.New("AccessDenied")}, + {"throttled", errors.New("SlowDown: please reduce your request rate")}, + } { + t.Run(tc.name, func(t *testing.T) { + t.Parallel() + + ctx := context.Background() + be := &failReads{Backend: backend.Memory(), only: "/manifest", err: tc.err} + e := newEngine(t, be) + ids := twoParts(t, e, be) + + before := loadIndex(t, be) + + be.armed.Store(true) + r := newEngine(t, be) + err := r.LoadParts(ctx) + be.armed.Store(false) + + after := loadIndex(t, be) + assert.Empty(t, after.Wanted, "a transient failure is not evidence of loss") + assert.Equal(t, prefixes(before.Entries), prefixes(after.Entries), + "a transient failure must not strip live parts from the index") + assert.Len(t, after.Entries, 2) + assert.Equal(t, before.Generation, after.Generation, "nothing was committed") + assert.NotEmpty(t, diskPartKeys(ctx, t, be, ids[0])) + assert.Error(t, err, "a transient failure must still fail the load") + }) + } +} + +// TestCorruptPartIsNotAWant pins the deliberately narrow trigger: objects that are present but +// unreadable are a different failure with a different remedy, and widening the trigger to cover +// them is how a repair path becomes a data-destruction path. +func TestCorruptPartIsNotAWant(t *testing.T) { + t.Parallel() + + ctx := context.Background() + be := backend.Memory() + e := newEngine(t, be) + ids := twoParts(t, e, be) + + objs := diskPartKeys(ctx, t, be, ids[0]) + i := slices.IndexFunc(objs, func(k string) bool { return strings.HasSuffix(k, "/manifest") }) + require.GreaterOrEqual(t, i, 0) + require.NoError(t, be.Write(ctx, objs[i], []byte("not a manifest"))) + + r := newEngine(t, be) + require.Error(t, r.LoadParts(ctx)) + require.Empty(t, loadIndex(t, be).Wanted) +} + +// TestWantIsRecordedInOneCommit verifies the atomicity by construction: dropping the entry and +// recording the want is a single compare-and-swap, so there is no window in which a crash could +// land one without the other. +func TestWantIsRecordedInOneCommit(t *testing.T) { + t.Parallel() + + ctx := context.Background() + be := &countCAS{Backend: backend.Memory(), key: indexKey()} + e := newEngine(t, be) + ids := twoParts(t, e, be) + + erasePart(ctx, t, be, ids[0]) + + be.n.Store(0) + r := newEngine(t, be) + require.NoError(t, r.LoadParts(ctx)) + + require.Equal(t, int64(1), be.n.Load(), "the drop and the want are one commit, not two") + + ix := loadIndex(t, be) + require.NotContains(t, prefixes(ix.Entries), enginePrefix+"/"+ids[0]) + require.Contains(t, wantPrefixes(ix.Wanted), enginePrefix+"/"+ids[0]) +} + +// TestFailedWantCommitAppliesNeither verifies a lost race leaves the stored index untouched: the +// entry is still there and no want is, so the retry re-reads and re-derives both halves rather +// than inheriting half of an applied commit. +func TestFailedWantCommitAppliesNeither(t *testing.T) { + t.Parallel() + + ctx := context.Background() + be := &countCAS{Backend: backend.Memory(), key: indexKey()} + e := newEngine(t, be) + ids := twoParts(t, e, be) + + before := loadIndex(t, be) + erasePart(ctx, t, be, ids[0]) + + be.refuse.Store(true) + r := newEngine(t, be) + require.Error(t, r.LoadParts(ctx), "a commit that never lands must not be reported as success") + be.refuse.Store(false) + + after := loadIndex(t, be) + require.Equal(t, prefixes(before.Entries), prefixes(after.Entries), "the entry is still there") + require.Empty(t, after.Wanted, "and the want is not") + + // The retry re-derives both halves from the same evidence and commits them together. + r2 := newEngine(t, be) + require.NoError(t, r2.LoadParts(ctx)) + + ix := loadIndex(t, be) + require.Equal(t, []string{enginePrefix + "/" + ids[1]}, prefixes(ix.Entries)) + require.Equal(t, []string{enginePrefix + "/" + ids[0]}, wantPrefixes(ix.Wanted)) +} + +// TestWantSurvivesLaterCommits verifies an obligation is durable: it is carried through every later +// index this engine writes, so a crash between recording a want and satisfying it is benign. +func TestWantSurvivesLaterCommits(t *testing.T) { + t.Parallel() + + ctx := context.Background() + be := backend.Memory() + e := newEngine(t, be) + ids := twoParts(t, e, be) + + erasePart(ctx, t, be, ids[0]) + + r := newEngine(t, be) + require.NoError(t, r.LoadParts(ctx)) + + ingest(t, r, mkBatch("api", rrec{ts: 300, body: "p3"})) + require.NoError(t, r.Flush(ctx)) + require.NoError(t, r.Merge(ctx, 0)) + + require.Equal(t, []string{enginePrefix + "/" + ids[0]}, wantPrefixes(loadIndex(t, be).Wanted)) + + // And across a restart, which reads it back off the index rather than rediscovering it. + r2 := newEngine(t, be) + require.NoError(t, r2.LoadParts(ctx)) + require.Equal(t, []string{enginePrefix + "/" + ids[0]}, wantPrefixes(loadIndex(t, be).Wanted)) +} + +// invariantCAS asserts, on every index commit that lands, that no part vanished from Entries +// without landing in exactly one of Removed and Wanted. It sits on the real commit path, so every +// flush, merge, retention drop and lossy load in a test is checked. +type invariantCAS struct { + backend.Backend + + t *testing.T + key string +} + +func (c *invariantCAS) CompareAndSwap( + ctx context.Context, key string, expected backend.Version, data []byte, +) (backend.Version, bool, error) { + if key != c.key { + return c.Backend.CompareAndSwap(ctx, key, expected, data) + } + + prev, err := bucketindex.Load(ctx, c.Backend, key) + require.NoError(c.t, err) + + version, ok, err := c.Backend.CompareAndSwap(ctx, key, expected, data) + if err != nil || !ok { + return version, ok, err + } + + next, decErr := bucketindex.Decode(data) + require.NoError(c.t, decErr) + assertLeavesIntoOneList(c.t, prev, next) + + return version, ok, err +} + +func assertLeavesIntoOneList(t *testing.T, prev, next *bucketindex.Index) { + t.Helper() + + live := prefixes(next.Entries) + + for _, p := range prefixes(prev.Entries) { + if slices.Contains(live, p) { + continue + } + + removed := slices.ContainsFunc(next.Removed, func(r bucketindex.Removal) bool { return r.Prefix == p }) + wanted := slices.ContainsFunc(next.Wanted, func(w bucketindex.Want) bool { return w.Prefix == p }) + + switch { + case removed && wanted: + require.Fail(t, "a part left Entries into both Removed and Wanted", "part %q", p) + case !removed && !wanted: + require.Fail(t, "a part vanished from Entries into neither Removed nor Wanted", "part %q", p) + } + } +} + +// TestEntriesLeaveOnlyIntoRemovedOrWanted is the design in one line, as a property over the real +// commit path: random ingest, flush, merge, retention and part-loss sequences, with every committed +// index checked against the one it replaced. +func TestEntriesLeaveOnlyIntoRemovedOrWanted(t *testing.T) { + t.Parallel() + + ctx := context.Background() + + for seed := range uint64(24) { + t.Run("", func(t *testing.T) { + t.Parallel() + + rng := rand.New(rand.NewPCG(seed, 0x5eed)) + be := &invariantCAS{Backend: backend.Memory(), t: t, key: indexKey()} + e := newEngine(t, be) + ts := int64(100) + + for step := range 12 { + ts += 100 + + switch rng.IntN(5) { + case 0, 1: + ingest(t, e, mkBatch("api", rrec{ts: ts, body: "r"})) + require.NoError(t, e.Flush(ctx)) + case 2: + require.NoError(t, e.Merge(ctx, 0)) + case 3: + // Retention: parts wholly below the horizon are dropped, not merged. + require.NoError(t, e.Merge(ctx, ts-150)) + case 4: + ids := diskPartIDs(ctx, t, be) + if len(ids) == 0 { + continue + } + + erasePart(ctx, t, be, ids[rng.IntN(len(ids))]) + + e = newEngine(t, be) + require.NoError(t, e.LoadParts(ctx), "step %d", step) + } + } + }) + } +} diff --git a/recordengine/reclaim.go b/recordengine/reclaim.go index c3b9679b..8b354f81 100644 --- a/recordengine/reclaim.go +++ b/recordengine/reclaim.go @@ -85,6 +85,17 @@ func (e *Engine) sweepOrphansLocked(ctx context.Context) error { live[p.prefix] = struct{}{} } + // A wanted part's objects are not orphans, they are what is left of it. Deleting the readable + // remainder of a part repair is about to fetch destroys evidence and buys nothing: the want + // bounds how long it is kept, and a discharged want puts the prefix back in the live set. + for _, w := range e.wants { + live[w.Prefix] = struct{}{} + } + + for _, w := range e.pendingWants { + live[w.Prefix] = struct{}{} + } + var orphans []string for _, k := range keys {