From 4206edeb6d51de1d1fac792e14e535fc4c897a5a Mon Sep 17 00:00:00 2001 From: Matt Turner Date: Fri, 23 Oct 2026 19:29:51 -0400 Subject: [PATCH 1/2] store: record which scans are of the trunk Which scans are PR scans was known when they were stored but not kept, so repack had to infer it from the shape of the parent graph. Keep it in a new trunk column, set by Put, and use it in repack. A store written before the column existed is backfilled on open with the inference repack used. On a copy of the production store it marks as trunk exactly the scans on gentoo-ci.git's master. Signed-off-by: Matt Turner --- cmd/gentoo-ci-http/repack.go | 45 ++++------------- internal/store/info.go | 3 +- internal/store/store.go | 93 +++++++++++++++++++++++++++++------- internal/store/store_test.go | 85 +++++++++++++++++++++++++++++++- 4 files changed, 172 insertions(+), 54 deletions(-) diff --git ./cmd/gentoo-ci-http/repack.go ./cmd/gentoo-ci-http/repack.go index 3f30295..8a8edd4 100644 --- ./cmd/gentoo-ci-http/repack.go +++ ./cmd/gentoo-ci-http/repack.go @@ -64,7 +64,12 @@ func cmdRepack(args []string) { // Oldest first, which is the order scans are stored in. all := from.RecentScans(n) slices.Reverse(all) - trunk := trunkOf(all) + trunk := 0 + for _, info := range all { + if info.Trunk { + trunk++ + } + } slog.Info("copying", slog.Int("scans", n), slog.String("from", *src), slog.String("to", *dst)) @@ -90,17 +95,16 @@ func cmdRepack(args []string) { } slog.Info("repacking", - slog.Int("scans", len(pending)), slog.Int("trunk", len(trunk)), + slog.Int("scans", len(pending)), slog.Int("trunk", trunk), slog.Int("candidates", *candidates)) start := time.Now() reparented, shrunk := 0, int64(0) for i, info := range pending { - onTrunk := trunk[info.Sha] // Trunk keeps the scan it was measured against; only PR scans, placed // against whatever looked closest at upload time, are ranked again. parent := info.Parent - if !onTrunk { + if !info.Trunk { var named []string if info.Parent != "" { named = []string{info.Parent} @@ -112,7 +116,7 @@ func cmdRepack(args []string) { // against it now, and an ancestor's depth may have shifted. A PR scan // is a leaf nothing builds on, so it stays a delta. was := info.Bytes - if _, _, err := to.Reparent(info.Sha, parent, onTrunk); err != nil { + if _, _, err := to.Reparent(info.Sha, parent, info.Trunk); err != nil { slog.Error("reparenting", slog.String("scan", info.Sha), slog.Any("err", err)) os.Exit(1) } @@ -154,37 +158,6 @@ func cmdRepack(args []string) { report(before, to.Summary()) } -// trunkOf marks the scans on the longest chain through the parent graph. -// -// Which scans were PR scans is not recorded, but it shows in the shape of the -// graph: trunk scans chain one after another and so form by far the longest -// path, and everything hanging off that path is a PR scan. On the production -// store this agrees with the loader to within one scan. -// -// It assumes a single trunk; a store built from two unrelated histories would -// keep only the longer one. all must be oldest first and complete, so every -// parent is seen before its children. -func trunkOf(all []store.ScanInfo) map[string]bool { - depth := make(map[string]int, len(all)) - parent := make(map[string]string, len(all)) - best, bestDepth := "", 0 - for _, info := range all { - d := 1 - if info.Parent != "" { - d = depth[info.Parent] + 1 - } - depth[info.Sha], parent[info.Sha] = d, info.Parent - if d > bestDepth { - bestDepth, best = d, info.Sha - } - } - trunk := make(map[string]bool, bestDepth) - for cur := best; cur != ""; cur = parent[cur] { - trunk[cur] = true - } - return trunk -} - // report prints what the repack changed, each metric side by side. func report(before, after store.Summary) { line := func(name, was, now string) { diff --git ./internal/store/info.go ./internal/store/info.go index 5fb3550..faf7484 100644 --- ./internal/store/info.go +++ ./internal/store/info.go @@ -17,6 +17,7 @@ type ScanInfo struct { Added int Removed int Snapshot bool + Trunk bool // a scan of the trunk, not of a PR/branch Depth int Bytes int } @@ -26,7 +27,7 @@ func (s *Store) infoAt(i int) ScanInfo { info := ScanInfo{ Sha: r.sha, Time: time.Unix(r.ts, 0), Results: int(r.nResults), Added: int(r.nAdded), Removed: int(r.nRemoved), - Snapshot: r.isSnapshot, Depth: int(r.depth), Bytes: int(r.blobLen), + Snapshot: r.isSnapshot, Trunk: r.trunk, Depth: int(r.depth), Bytes: int(r.blobLen), } if r.parent >= 0 && int(r.parent) < len(s.scans) { info.Parent = s.scans[r.parent].sha diff --git ./internal/store/store.go ./internal/store/store.go index 5a5b4ee..8d82a25 100644 --- ./internal/store/store.go +++ ./internal/store/store.go @@ -129,6 +129,7 @@ type scanRec struct { nAdded int32 nRemoved int32 isSnapshot bool + trunk bool depth int32 blobLen int32 } @@ -251,16 +252,74 @@ func (s *Store) initSchema() error { is_snapshot INTEGER NOT NULL, depth INTEGER NOT NULL, payload BLOB NOT NULL, - sketch BLOB + sketch BLOB, + trunk INTEGER NOT NULL DEFAULT 0 )`, } { if _, err := s.db.Exec(q); err != nil { return fmt.Errorf("%.50s: %w", q, err) } } + var n int + if err := s.db.QueryRow( + `SELECT COUNT(*) FROM pragma_table_info('scans') WHERE name='trunk'`).Scan(&n); err != nil { + return err + } + if n == 0 { + if err := s.addTrunk(); err != nil { + return fmt.Errorf("adding scans.trunk: %w", err) + } + } return nil } +// addTrunk adds scans.trunk to a store written before scans recorded it, and +// fills it in the same transaction so the column is never left unset. Which +// scans are trunk is not known for certain, but shows in the shape of the +// graph: trunk scans chain one after another and so form by far the longest +// path, and everything hanging off that path is a PR scan. +// +// It assumes a single trunk; a store built from two unrelated histories would +// keep only the longer one. +func (s *Store) addTrunk() error { + tx, err := s.db.Begin() + if err != nil { + return err + } + defer tx.Rollback() + if _, err := tx.Exec(`ALTER TABLE scans ADD COLUMN trunk INTEGER NOT NULL DEFAULT 0`); err != nil { + return err + } + rows, err := tx.Query(`SELECT id, COALESCE(parent,-1) FROM scans ORDER BY id`) + if err != nil { + return err + } + defer rows.Close() + // Scans are stored after their parents, so one pass finds every depth. + depth, parent := map[int64]int{}, map[int64]int64{} + best := int64(-1) + for rows.Next() { + var id, p int64 + if err := rows.Scan(&id, &p); err != nil { + return err + } + depth[id], parent[id] = depth[p]+1, p + if best < 0 || depth[id] > depth[best] { + best = id + } + } + if err := rows.Err(); err != nil { + return err + } + rows.Close() + for id := best; id >= 0; id = parent[id] { + if _, err := tx.Exec(`UPDATE scans SET trunk=1 WHERE id=?`, id); err != nil { + return err + } + } + return tx.Commit() +} + // resolveSnapshotEvery mirrors the file backend: the interval belongs to the // data, so it is stored and reused unless explicitly overridden. func (s *Store) resolveSnapshotEvery(want int) error { @@ -308,7 +367,7 @@ func (s *Store) load() error { if rows, err = s.db.Query( `SELECT id, sha, COALESCE(parent,-1), ts, n_results, n_added, n_removed, - is_snapshot, depth, LENGTH(payload), sketch FROM scans ORDER BY id`); err != nil { + is_snapshot, trunk, depth, LENGTH(payload), sketch FROM scans ORDER BY id`); err != nil { return err } defer rows.Close() @@ -321,7 +380,7 @@ func (s *Store) load() error { var parentRowID int64 var sk []byte if err := rows.Scan(&r.id, &r.sha, &parentRowID, &r.ts, &r.nResults, - &r.nAdded, &r.nRemoved, &r.isSnapshot, &r.depth, &r.blobLen, &sk); err != nil { + &r.nAdded, &r.nRemoved, &r.isSnapshot, &r.trunk, &r.depth, &r.blobLen, &sk); err != nil { return err } r.parent = -1 @@ -659,16 +718,17 @@ func (s *Store) encode(rows, prev delta.Rows, parent int32, allowSnapshot bool) // Put stores a scan against the scan named by fromSha ("" for a root scan). // -// allowSnapshot says whether this scan may be materialised as a full snapshot -// to bound chain depth. Trunk scans should allow it: they are what later scans -// chain through, so a snapshot there bounds everyone's replay. A PR/branch scan -// is a leaf nothing builds on, so shortening its chain buys nothing. +// trunk says whether this is a scan of the trunk rather than of a PR/branch. +// Only a trunk scan may be materialized as a full snapshot to bound chain +// depth: trunk scans are what later scans chain through, so a snapshot there +// bounds everyone's replay. A PR/branch scan is a leaf nothing builds on, so +// shortening its chain buys nothing. // // Either way the smaller encoding wins: a branch scan whose parent is a poor // match, a full repo scan hanging off a partial one say, produces a "delta" // that restates the whole report and comes out larger than a snapshot. A scan // with no parent has nothing to delta against and is snapshotted regardless. -func (s *Store) Put(sha, fromSha string, ts time.Time, rows delta.Rows, allowSnapshot bool) (added, removed int, err error) { +func (s *Store) Put(sha, fromSha string, ts time.Time, rows delta.Rows, trunk bool) (added, removed int, err error) { parent := int32(-1) var parentID any prev := delta.Rows{} @@ -684,15 +744,15 @@ func (s *Store) Put(sha, fromSha string, ts time.Time, rows delta.Rows, allowSna } } - enc := s.encode(rows, prev, parent, allowSnapshot) + enc := s.encode(rows, prev, parent, trunk) addSet, delSet, blob, snapshot, depth := enc.add, enc.del, enc.blob, enc.snapshot, enc.depth sketch := delta.Of(rows) res, err := s.exec(`INSERT INTO scans - (sha,parent,ts,n_results,n_added,n_removed,is_snapshot,depth,payload,sketch) - VALUES(?,?,?,?,?,?,?,?,?,?)`, + (sha,parent,ts,n_results,n_added,n_removed,is_snapshot,depth,payload,sketch,trunk) + VALUES(?,?,?,?,?,?,?,?,?,?,?)`, sha, parentID, ts.Unix(), len(rows), len(addSet), len(delSet), snapshot, depth, blob, - delta.EncodeSketch(sketch)) + delta.EncodeSketch(sketch), trunk) if err != nil { return 0, 0, fmt.Errorf("insert scan: %w", err) } @@ -710,7 +770,7 @@ func (s *Store) Put(sha, fromSha string, ts time.Time, rows delta.Rows, allowSna s.scans = append(s.scans, scanRec{ id: id, sha: sha, parent: parent, ts: ts.Unix(), nResults: int32(len(rows)), nAdded: int32(len(addSet)), nRemoved: int32(len(delSet)), - isSnapshot: snapshot, depth: depth, + isSnapshot: snapshot, trunk: trunk, depth: depth, blobLen: int32(len(blob)), }) s.sketches = append(s.sketches, sketch) @@ -735,11 +795,12 @@ func (s *Store) CopyTo(path string) error { // are stored is. // // fromSha must name a scan stored before this one, since a chain is replayed -// by walking backwards. Returns the new added/removed counts. +// by walking backwards. trunk is as for Put. Returns the new added/removed +// counts. // // Callers that move several scans must work in storage order: a scan's depth // follows its parent's, so a parent has to be settled before its children. -func (s *Store) Reparent(sha, fromSha string, allowSnapshot bool) (added, removed int, err error) { +func (s *Store) Reparent(sha, fromSha string, trunk bool) (added, removed int, err error) { idx, ok := s.byShaIdx[sha] if !ok { return 0, 0, fmt.Errorf("unknown scan %s", sha) @@ -767,7 +828,7 @@ func (s *Store) Reparent(sha, fromSha string, allowSnapshot bool) (added, remove } } - enc := s.encode(rows, prev, parent, allowSnapshot) + enc := s.encode(rows, prev, parent, trunk) rec := &s.scans[idx] if _, err := s.exec(`UPDATE scans SET parent=?, n_added=?, n_removed=?, is_snapshot=?, depth=?, payload=? diff --git ./internal/store/store_test.go ./internal/store/store_test.go index 14d5f43..87514b8 100644 --- ./internal/store/store_test.go +++ ./internal/store/store_test.go @@ -63,7 +63,7 @@ func TestReopenAndReconstruct(t *testing.T) { t.Fatal(err) } if rec := q.scans[q.byShaIdx["pr-9"]]; rec.isSnapshot { - t.Error("branch scan was snapshotted despite allowSnapshot=false") + t.Error("branch scan was snapshotted despite trunk=false") } m := map[any]int{} for k, v := range asMultiset(prResults) { @@ -227,3 +227,86 @@ func TestMessageBlocks(t *testing.T) { t.Fatal(err) } } + +// putChain stores trunk scans t0..t(n-1), one second apart, and a scan pr off +// t1, stored as trunk when prTrunk is set. +func putChain(t *testing.T, s *Store, n int, prTrunk bool) { + t.Helper() + rng := rand.New(rand.NewSource(7)) + rows, err := s.Intern(randomResults(rng, 50)) + if err != nil { + t.Fatal(err) + } + parent := "" + for i := range n { + sha := fmt.Sprintf("t%d", i) + if _, _, err := s.Put(sha, parent, time.Unix(int64(i), 0), rows, true); err != nil { + t.Fatal(err) + } + parent = sha + } + if _, _, err := s.Put("pr", "t1", time.Unix(int64(n), 0), rows, prTrunk); err != nil { + t.Fatal(err) + } +} + +func trunkScans(s *Store) []string { + var out []string + for _, info := range s.RecentScans(s.NumScans()) { + if info.Trunk { + out = append(out, info.Sha) + } + } + return out +} + +func TestTrunkRecorded(t *testing.T) { + path := t.TempDir() + "/scans.db" + s, err := Open(path, 16) + if err != nil { + t.Fatal(err) + } + putChain(t, s, 3, false) + want := []string{"t2", "t1", "t0"} + if got := trunkScans(s); !reflect.DeepEqual(got, want) { + t.Errorf("trunk = %v, want %v", got, want) + } + if err := s.Close(); err != nil { + t.Fatal(err) + } + + if s, err = Open(path, 16); err != nil { + t.Fatal(err) + } + defer s.Close() + if got := trunkScans(s); !reflect.DeepEqual(got, want) { + t.Errorf("after reopening, trunk = %v, want %v", got, want) + } +} + +// A store from before scans.trunk existed gets it from the graph's shape. +func TestTrunkBackfilled(t *testing.T) { + path := t.TempDir() + "/scans.db" + s, err := Open(path, 16) + if err != nil { + t.Fatal(err) + } + putChain(t, s, 3, true) + if err := s.Sync(); err != nil { + t.Fatal(err) + } + if _, err := s.db.Exec(`ALTER TABLE scans DROP COLUMN trunk`); err != nil { + t.Fatal(err) + } + if err := s.Close(); err != nil { + t.Fatal(err) + } + + if s, err = Open(path, 16); err != nil { + t.Fatal(err) + } + defer s.Close() + if got, want := trunkScans(s), []string{"t2", "t1", "t0"}; !reflect.DeepEqual(got, want) { + t.Errorf("trunk = %v, want %v", got, want) + } +} -- 2.55.0