feat(engine): record a want for a part the owner cannot read

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

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

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).

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

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) }