fix(bucketindex): keep split lineage and rollup layouts in the index

ARCHITECTURE.md

@@ -261,9 +261,9 @@ to fail over to: a read overlapping a want fails.
   reads as unknown, never raw) / marks (`OTMK`) / column object framing (the
   trailer-dictionary bytes column; the leading layout still read) and key layout, the attribute
   hash+binary encoding (the SeriesID pre-image), symbol table (`OTSY`), the bucket index (format
-  v7: entries with their term-scoped block sets, split claims and writing term, tombstones, wants,
-  per-writer flush watermarks and the block high-water mark — readable back to v1, not writable by
-  older nodes), WAL record framing
+  v8: entries with their term-scoped block sets, split claims, writing term and rollup layout,
+  tombstones, wants, per-writer flush watermarks, the block high-water mark and the lineage catalog —
+  readable back to v1, not writable by older nodes), WAL record framing
   (additive record types), record-key footer (`OTKY`), the metric part column layout
   (`[series:int128, ts:int64, value:float64]` + optional `sf:float64`), and the profile
   symbol-store sidecar (`OTSP`, version 2: a compressed body with its raw length; version 1 still

backend/ARCH.md

@@ -206,6 +206,44 @@ a demonstration.
   short of reading the rows says how. A part covering only some of a group's claim therefore
   overlaps every member while replacing none; no single-part rule resolves it, and repair refuses
   to publish one beside the other (`engine/ARCH.md`, "Repair").
+  **A group's claim outlives its members in the lineage catalog** (`Index.Catalog`, format v8). A
+  merge that consumes a whole group folds the claim into its output's blocks, and a group split
+  again leaves the outer claim on no entry at all — yet a want for an inner member can be answered
+  by a successor of the outer ancestry only through that claim. So every commit records the claims
+  its entries and wants carry (`RecordLineage`), and `Index.Lineage` — the catalog plus the claims
+  still on entries and wants — is what every identity question is answered by: `Satisfying`,
+  `Discharging`, `TrimWants`, `TrimHoles` (so `Revokes`), joint coverage (`Covered`, which realizes
+  any group the lineage records, transitively), `Subsumed` (every group the lineage records that the
+  parts complete, not only the groups some entry still carries) and `Missing` (the members of every
+  group derived from a want's rows that no later group consumed, so a member split again is asked for
+  through its own split). A joint answer names any part descending through a recorded split, not
+  only one carrying a claim: a merge that consumed a whole inner group folded its claim away. The shared repair pass carries the engine's catalog through every index it builds, down to
+  commit admission (`internal/repair`). A claim is a fact about rows wherever it was recorded, so
+  lineages merge freely: a rebase unions the rival's catalog, a replica installing a peer's index
+  keeps its own, and `cluster/partsync` answers a want with the wanting node's lineage beside the
+  peer's (`SatisfyingWith`), since the peer may never have seen either split. Only a committed
+  claim is recorded: a group run a lost CAS attempt allocated is handed out again, maybe to other
+  parts. `MaxLineage` (= `MaxRemovals`) is a target, not a cap: `TrimCatalog` ages the lowest groups
+  out first but never one the ancestry of an outstanding want, a hole or a live entry passes through,
+  since dropping it could turn a want a successor still answers into absence and then a hole. A live
+  entry reaches its whole ancestry, not only its own claim: once it is lost, its want carries only that
+  claim, and an outer ancestor's successor answers it only through the rest. The walk shares one
+  visited set across the whole part set, so each group is followed once: trimming a catalog 200 over
+  the target with 600 live parts over 4200 claims takes 2.7ms, against 1.7ms following only wants and
+  holes. One closure per part costs 6.6ms there, and one closure over the union of every part's blocks
+  costs 390ms, since growing an interval of thousands of runs rebuilds it on each step. Only
+  unreachable claims go; a catalog every claim of which is reached stays above the target, and the
+  engine warns once.
+  **Every identity question over a lineage goes through one `Relater`** (and `Index.Relations` over
+  an index), which indexes the claims' block runs and computes each closure — what a set of blocks
+  holds, derives from, or completes — by a work queue that visits only the claims a newly gained run
+  meets, and remembers it. A batch then pays for each part's closure once: a commit's `TrimWants`,
+  a repair pass, partsync's answers to one peer. The naive fixed point rescans the whole lineage until
+  nothing changes, per candidate per want, which a nested catalog near the target turns into seconds
+  under the state lock: `TrimWants` of 256 nested wants over a 4000-claim catalog and 100 live parts
+  took 6.7s that way and takes 10ms this way.
+  **An entry carries its part's rollup layout** (`Entry.Rollup`, format v8; nil is unknown), so a
+  writer checks a rival's layout without opening the rival's part (`engine/ARCH.md`, "Merge").
   **Allocation is the shard owner's alone, and scoped to its tenure**: `Index.NextBlock(term)` is one
   above `AllocatedBlocks`, the persisted high-water mark, and above every block the live entries,
   wants and group runs still name — within `term`, starting at `(term, 1)` for a tenure the index has
@@ -290,21 +328,27 @@ a demonstration.
 - **A hole is revocable.** The commit is not cross-replica atomic (unlike ClickHouse's
   `createEmptyPartInsteadOfLost`, which checks non-existence on every replica in one transaction),
   so an owner may acknowledge a loss while a peer still holds the data. `Revokes` therefore replaces
-  a hole with any data-bearing entry at the same prefix or any successor containing its blocks, and
-  `TrimHoles` runs on every commit — so a repair fetch, a rival's adopted entry and a merge all
+  a hole with any data-bearing entry at the same prefix or any successor containing its blocks,
+  `TrimHoles` also with a complete split group jointly covering it (the answer `Satisfying` gives the
+  hole's want), and `TrimHoles` runs on every commit — so a repair fetch, a rival's adopted entry and a merge all
   revoke it as a side effect. `NextBlock` counts a hole's interval, because the part it stands for
   may yet come back and two parts must never claim one identity.
 - **`LostParts` is monotone and cluster-visible.** It is carried forward across commits and raised
   to the maximum on rebase, never lowered — a revoked hole leaves the count where it was. Data loss
   is a fact, not a level: a per-node gauge that a restart or a successful repair clears erases the
   only record that a range of a shard was ever acknowledged as gone. It follows ClickHouse's
   `/lost_part_count`.
-- **Format v7 is a hard read break.** `Decode` rejects any version above the one it knows, so
-  reading is backward compatible (v1–v6 still decode, unset fields ordering below everything a
-  writer produces) but **writing is not**: a node on pre-v7 code fails on the first v7 index it
+- **Format v8 is a hard read break.** `Decode` rejects any version above the one it knows, so
+  reading is backward compatible (v1–v7 still decode, unset fields ordering below everything a
+  writer produces) but **writing is not**: a node on pre-v8 code fails on the first v8 index it
   reads. Every node that reads a given index must be upgraded together. The path differs by
   deployment and both matter — on a shared backend every node reads the same index object, and in
   `PrivateBackend` mode `cluster/partsync` reads its peers' indexes.
+  **A v7 index knows no layout and no catalog.** Every entry's layout is unknown until a writer
+  opens the part and its next commit records the manifest's; the first v8 commit catalogs the claims
+  its entries and wants still carry. A layout rides behind a flag bit in the entry's flag word, so
+  an unknown one costs nothing; the catalog is canonical like the rest — sorted, deduplicated,
+  bounded — and `Decode` rejects any other form.
   **A v6 index is term 0 throughout**: its blocks, its entries' and wants' writing terms and its
   high-water mark. A clustered tenure then allocates from `(term, 1)` above all of it, and a writer
   with no cluster continues term 0's numbering where v6 left it. An interval's lowest block number

backend/bucketindex/bench_layout_test.go

@@ -0,0 +1,55 @@
+package bucketindex_test
+
+import (
+	"testing"
+
+	"github.com/oteldb/storage/backend/bucketindex"
+)
+
+// benchIndexWithLineage is [benchIndex] as a metric shard's index is: every part records its rollup
+// layout, and the catalog holds a long history of split groups.
+func benchIndexWithLineage() *bucketindex.Index {
+	ix := benchIndex()
+
+	for i := range ix.Entries {
+		ix.Entries[i].Rollup = &bucketindex.Rollup{Tiers: []bucketindex.RollupTier{
+			{Before: 1 << 50, Interval: 60e9, Agg: 1},
+			{Before: 1 << 49, Interval: 3600e9, Agg: 1},
+		}}
+	}
+
+	for i := range uint64(256) {
+		ix.Catalog = append(ix.Catalog, bucketindex.Claim{
+			Blocks: bucketindex.Blocks(4*i+1, 4*i+2), Group: bucketindex.Blocks(4096+2*i, 4097+2*i),
+		})
+	}
+
+	ix.Catalog = bucketindex.MergeLineage(ix.Catalog)
+
+	return ix
+}
+
+func BenchmarkIndexEncodeWithLineage(b *testing.B) {
+	ix := benchIndexWithLineage()
+	buf := ix.AppendBinary(nil)
+
+	b.SetBytes(int64(len(buf)))
+	b.ReportAllocs()
+
+	for b.Loop() {
+		buf = ix.AppendBinary(buf[:0])
+	}
+}
+
+func BenchmarkIndexDecodeWithLineage(b *testing.B) {
+	data := benchIndexWithLineage().AppendBinary(nil)
+
+	b.SetBytes(int64(len(data)))
+	b.ReportAllocs()
+
+	for b.Loop() {
+		if _, err := bucketindex.Decode(data); err != nil {
+			b.Fatal(err)
+		}
+	}
+}

backend/bucketindex/bench_lineage_test.go

@@ -0,0 +1,176 @@
+package bucketindex_test
+
+import (
+	"fmt"
+	"slices"
+	"testing"
+
+	"github.com/oteldb/storage/backend/bucketindex"
+)
+
+const (
+	benchChains = 500
+	benchDepth  = 8
+)
+
+// nestedLineage is a catalog near [bucketindex.MaxLineage] of deeply nested groups: benchChains
+// ancestors, each split benchDepth times, every split taking the previous split's second member. It
+// returns the catalog, the innermost member of each chain, and each chain's ancestor block.
+func nestedLineage() (catalog []bucketindex.Claim, inner []bucketindex.Want, ancestors []uint64) {
+	next := uint64(1)
+	take := func() uint64 { next++; return next - 1 }
+
+	for c := range benchChains {
+		anc := take()
+		ancestors = append(ancestors, anc)
+		consumed := bucketindex.Blocks(anc)
+
+		var member uint64
+
+		for d := range benchDepth {
+			a, b := take(), take()
+			claim := bucketindex.Claim{Blocks: consumed, Group: bucketindex.Blocks(a, b)}
+			catalog = append(catalog, claim)
+			consumed, member = bucketindex.Blocks(b), b
+
+			if d == benchDepth-1 {
+				inner = append(inner, bucketindex.Want{
+					Prefix: fmt.Sprintf("inner-%04d", c), Blocks: bucketindex.Blocks(member),
+					Claim: claim, Level: uint32(d + 1),
+				})
+			}
+		}
+	}
+
+	return bucketindex.MergeLineage(catalog), inner, ancestors
+}
+
+// nestedLive is a shard's live set over the nested lineage: 100 parts, each a high-level successor
+// of five chains' ancestors — so it answers those chains' inner wants only through the whole chain.
+func nestedLive(ancestors []uint64) []bucketindex.Entry {
+	live := make([]bucketindex.Entry, 0, 100)
+
+	for i := range 100 {
+		live = append(live, bucketindex.Entry{
+			Prefix: fmt.Sprintf("succ-%03d", i),
+			Blocks: bucketindex.Blocks(ancestors[5*i : 5*i+5]...),
+			Level:  benchDepth + 5,
+		})
+	}
+
+	return live
+}
+
+func BenchmarkTrimWantsNestedLineage(b *testing.B) {
+	catalog, inner, ancestors := nestedLineage()
+	live := nestedLive(ancestors)
+
+	for _, n := range []int{1, 16, 256} {
+		b.Run(fmt.Sprintf("wants=%d", n), func(b *testing.B) {
+			wants := inner[:n]
+			b.ReportAllocs()
+
+			for b.Loop() {
+				bucketindex.TrimWants(append([]bucketindex.Want(nil), wants...), live, catalog...)
+			}
+		})
+	}
+}
+
+// BenchmarkSatisfyingWithNestedLineage is a peer answering a batch of repair wants with the wanting
+// node's lineage, as partsync does: the relations built once per peer, then asked per want.
+func BenchmarkSatisfyingWithNestedLineage(b *testing.B) {
+	catalog, inner, ancestors := nestedLineage()
+	peer := &bucketindex.Index{Entries: nestedLive(ancestors)}
+
+	for _, n := range []int{1, 16} {
+		b.Run(fmt.Sprintf("wants=%d", n), func(b *testing.B) {
+			b.ReportAllocs()
+
+			for b.Loop() {
+				rel := peer.Relations(catalog...)
+				for i := range n {
+					rel.Satisfying(inner[i])
+				}
+			}
+		})
+	}
+}
+
+// nestedCompletion is, for the first n chains of [nestedLineage], the parts completing each chain's
+// outermost group: the first member of every split and both innermost members, so every group of
+// those chains is complete, the outer ones only through the ones nested in them.
+func nestedCompletion(n int) []bucketindex.Entry {
+	var out []bucketindex.Entry
+
+	for c := range n {
+		anc := 1 + uint64(c)*(1+2*benchDepth)
+		consumed := bucketindex.Blocks(anc)
+
+		for d := range benchDepth {
+			a := anc + 1 + 2*uint64(d)
+			claim := bucketindex.Claim{Blocks: consumed, Group: bucketindex.Blocks(a, a+1)}
+			consumed = bucketindex.Blocks(a + 1)
+
+			members := []uint64{a}
+			if d == benchDepth-1 {
+				members = append(members, a+1)
+			}
+
+			for _, m := range members {
+				out = append(out, bucketindex.Entry{
+					Prefix: fmt.Sprintf("c%04d-%d", c, m), Blocks: bucketindex.Blocks(m), Claim: claim, Level: uint32(d + 1),
+				})
+			}
+		}
+	}
+
+	return out
+}
+
+// BenchmarkSubsumedNestedLineage is a repair commit's pruning: the parts completing n nested chains
+// judged against the live set over the whole catalog.
+func BenchmarkSubsumedNestedLineage(b *testing.B) {
+	catalog, _, ancestors := nestedLineage()
+	live := nestedLive(ancestors)
+
+	for _, n := range []int{1, 16, 256} {
+		b.Run(fmt.Sprintf("chains=%d", n), func(b *testing.B) {
+			added := nestedCompletion(n)
+			all := append(append([]bucketindex.Entry(nil), live...), added...)
+			b.ReportAllocs()
+
+			for b.Loop() {
+				bucketindex.LineageOf(all).With(catalog...).Relater().Subsumed(live, added)
+			}
+		})
+	}
+}
+
+// BenchmarkRecordLineageSaturated is a commit's catalog upkeep past the target: the nested catalog
+// with 200 newer groups nothing reaches, and a live set of every chain's innermost member beside the
+// successors, so trimming walks each live part's whole ancestry.
+func BenchmarkRecordLineageSaturated(b *testing.B) {
+	catalog, inner, ancestors := nestedLineage()
+
+	for i := range uint64(200) {
+		base := uint64(1 << 20)
+		catalog = append(catalog, bucketindex.Claim{
+			Blocks: bucketindex.Blocks(base + 2*i), Group: bucketindex.Blocks(base + 2*i + 1),
+		})
+	}
+
+	catalog = bucketindex.MergeLineage(catalog)
+	live := nestedLive(ancestors)
+
+	for i := range inner {
+		live = append(live, inner[i].Entry())
+	}
+
+	b.ReportAllocs()
+
+	for b.Loop() {
+		ix := &bucketindex.Index{Entries: live, Catalog: slices.Clone(catalog)}
+		ix.RecordLineage()
+	}
+}

backend/bucketindex/bench_test.go

@@ -1,9 +1,11 @@
 package bucketindex_test
 
 import (
+	"context"
 	"fmt"
 	"testing"
 
+	"github.com/oteldb/storage/backend"
 	"github.com/oteldb/storage/backend/bucketindex"
 )
 
@@ -64,6 +66,28 @@ func BenchmarkIndexDecode(b *testing.B) {
 	}
 }
 
+// BenchmarkIndexLoad is the read a load and every rebase pay: a versioned read of the stored index
+// and its decode.
+func BenchmarkIndexLoad(b *testing.B) {
+	ctx := context.Background()
+	be := backend.Memory()
+	key := "default/metrics/" + bucketindex.Object
+
+	ix := benchIndex()
+	if _, err := ix.Save(ctx, be, key, backend.VersionAbsent); err != nil {
+		b.Fatal(err)
+	}
+
+	b.SetBytes(int64(len(ix.AppendBinary(nil))))
+	b.ReportAllocs()
+
+	for b.Loop() {
+		if _, _, err := bucketindex.LoadVersioned(ctx, be, key); err != nil {
+			b.Fatal(err)
+		}
+	}
+}
+
 // BenchmarkTrimWants is the identity relation a commit runs per want over the live set.
 func BenchmarkTrimWants(b *testing.B) {
 	ix := benchIndex()