package ingest
import (
"context"
"log"
"time"
"craftdesign.group/quarry/internal/index"
"craftdesign.group/quarry/internal/zitcli"
)
// lineCountBatch is how many check-ins' counts one backfill
// transaction writes.
const lineCountBatch = 500
// backfillLineCounts fills the line counts an index folded before they
// existed (#349) is missing: one pass of zit's crosslink, whose counts
// a rebuild would have taken, written only into rows that have none.
// Once per process, in the background, on a rebuild slot so a fleet of
// stores does not all count at once — the first crosslink with counts
// pays one diff per changed file. It never holds the store's round
// lock: live folds count their own check-ins meanwhile, and the fill
// skips any row that already has counts.
func (p *Pipeline) backfillLineCounts() {
if p.ix.LineCountsDone() || !p.countsStarted.CompareAndSwap(false, true) {
return
}
go func() {
release, err := acquireRebuild(p.code, p.stop)
if err != nil {
return // stopping
}
defer release()
start := time.Now()
n, err := fillLineCounts(p.cli, p.ix)
if err != nil {
// Left unmarked, so the next process tries again.
log.Printf("ingest: %s: line counts: %v", p.code, err)
return
}
if err := p.ix.MarkLineCountsDone(); err != nil {
log.Printf("ingest: %s: mark line counts: %v", p.code, err)
return
}
log.Printf("ingest: %s: line counts for %d check-ins in %.1fs", p.code, n, time.Since(start).Seconds())
}()
}
// fillLineCounts streams crosslink and writes its counts into ix.
func fillLineCounts(cli *zitcli.Client, ix *index.Index) (int, error) {
n := 0
batch := map[string]map[string]index.LineCount{}
flush := func() error {
if len(batch) == 0 {
return nil
}
err := ix.FillLineCounts(batch)
clear(batch)
return err
}
err := cli.Crosslink(context.Background(), func(r zitcli.CrosslinkRecord) error {
lines := rootless(r.Parents, indexLines(r.Lines))
if len(lines) == 0 {
return nil
}
batch[r.Hash] = lines
n++
if len(batch) >= lineCountBatch {
return flush()
}
return nil
})
if err != nil {
return n, err
}
return n, flush()
}
// indexLines converts crosslink's counts to the index's.
func indexLines(lines map[string]zitcli.LineCount) map[string]index.LineCount {
if len(lines) == 0 {
return nil
}
out := make(map[string]index.LineCount, len(lines))
for path, lc := range lines {
out[path] = index.LineCount{Added: lc.Added, Removed: lc.Removed}
}
return out
}
// rootless drops the counts of a check-in with no parent. Crosslink
// counts one as all lines added, but zit diff --checkin refuses it, so
// a live fold cannot count it: dropping crosslink's count too keeps a
// rebuilt index and a live one agreeing. A repository's first check-in
// shows its files alone, which is all a reader needs of it.
func rootless(parents []string, lines map[string]index.LineCount) map[string]index.LineCount {
if len(parents) == 0 {
return nil
}
return lines
}