package ingest
import (
"context"
"errors"
"fmt"
"log"
"sort"
"time"
"craftdesign.group/quarry/internal/artifact"
"craftdesign.group/quarry/internal/attach"
"craftdesign.group/quarry/internal/index"
"craftdesign.group/quarry/internal/obs"
"craftdesign.group/quarry/internal/zitcli"
)
// fileLists resolves each check-in's changed paths — fossil's mlink
// computation, run at ingest where the artifact plumbing lives.
//
// A delta manifest's F cards ARE its changes. A baseline manifest lists
// every file, so its changes are a diff against the first parent's full
// list. Full lists are kept in a small LRU because parents almost
// always precede children in frame order; a miss (restart, cross-batch
// baseline) falls back to fetching the artifact by hash. When even that
// fails, the check-in simply contributes no rows: the listing shows no
// last-change for those paths rather than a wrong one.
type fileLists struct {
cli *zitcli.Client
cache map[string][]artifact.File
order []string // FIFO eviction; frame order makes recency ≈ arrival
entries int // total cached File entries, against the budget
// spool holds every manifest's RAW bytes for the duration of a
// rebuild. The census is not in parent-before-child order, so
// arrival-order caching misses constantly — measured at one
// subprocess fetch per FOLDED CHECK-IN on tcl, the whole rebuild
// budget. Raw manifests for the largest store are ~200MB, held
// only while the rebuild runs; a spool hit is a parse, not a
// process. Nil outside rebuilds.
spool map[string][]byte
// failed remembers hashes a fetch could not produce. A store's
// missing ancestors are referenced by MANY descendants, and without
// this each of them pays a ~1.3s subprocess to relearn the same
// absence — enough, on a tcl-sized replay, to fall behind the event
// stream's bounded tail and lose artifacts.
failed map[string]bool
}
var errKnownMissing = errors.New("ingest: artifact known missing")
const (
// The cache is budgeted by ENTRY count, not list count: tcl's
// census interleaves ticket artifacts between manifests, and a
// 48-list FIFO thrashed so hard that nearly every check-in of a
// bulk rebuild paid a subprocess fetch (~5 folds/s — hours). A
// million entries is a few hundred MB at worst, holds hundreds of
// recent manifests, and lives only as long as the rebuild does.
fileListEntryBudget = 1_000_000
// A manifest for a 3,000-file tree runs ~300KB; 8MB refuses only
// pathological artifacts.
manifestMax = 8 << 20
)
func newFileLists(cli *zitcli.Client) *fileLists {
return &fileLists{cli: cli, cache: map[string][]artifact.File{}, failed: map[string]bool{}}
}
func (fl *fileLists) put(hash string, files []artifact.File) {
if _, ok := fl.cache[hash]; ok {
return
}
for fl.entries+len(files) > fileListEntryBudget && len(fl.order) > 0 {
victim := fl.order[0]
fl.entries -= len(fl.cache[victim])
delete(fl.cache, victim)
fl.order = fl.order[1:]
}
fl.cache[hash] = files
fl.entries += len(files)
fl.order = append(fl.order, hash)
}
// errTree reports a tree check-in (S cards) where a file list was
// wanted: its files live in sub-manifests this code never reads.
var errTree = errors.New("ingest: a tree check-in's files are zit's to name")
// liveCallTimeout bounds each zit call a live fold makes. The fold runs
// inside an ingest-lane slot, under the store's round lock and before
// its messages are acked, so a call that hangs must end, and the
// check-in it was for is left pending rather than the round stalled.
// A retry of a pending check-in gets longer (retryCallTimeout).
var liveCallTimeout = 10 * time.Second
func liveCtx() (context.Context, context.CancelFunc) {
return callCtx(liveCallTimeout)
}
func callCtx(d time.Duration) (context.Context, context.CancelFunc) {
return context.WithTimeout(obs.WithReason(context.Background(), obs.ReasonIngest), d)
}
// full is the manifest's complete file list, baseline-expanded, sorted
// by name (F cards arrive sorted; the merge preserves that). A tree
// check-in has none to give (errTree).
func (fl *fileLists) full(hash string, art *artifact.Artifact) ([]artifact.File, error) {
if len(art.SubManifests) > 0 {
return nil, errTree
}
if art.Baseline == "" {
return art.Files, nil
}
base, err := fl.byHash(art.Baseline)
if err != nil {
return nil, err
}
return applyDelta(base, art.Files), nil
}
func (fl *fileLists) byHash(hash string) ([]artifact.File, error) {
if files, ok := fl.cache[hash]; ok {
return files, nil
}
if raw, ok := fl.spool[hash]; ok {
if art, err := artifact.Parse(raw); err == nil && art != nil {
files, ferr := fl.full(hash, art)
if ferr == nil {
fl.put(hash, files)
return files, nil
}
}
}
if fl.failed[hash] {
return nil, errKnownMissing
}
ctx, cancel := liveCtx()
raw, err := fl.cli.ArtifactRaw(ctx, hash, manifestMax)
timedOut := ctx.Err() != nil
cancel()
if err != nil {
if timedOut {
return nil, fmt.Errorf("%w: %v", context.DeadlineExceeded, err)
}
fl.failed[hash] = true
return nil, err
}
art, err := artifact.Parse(raw)
if err != nil || art == nil {
return nil, err
}
files, err := fl.full(hash, art)
if err != nil {
return nil, err
}
fl.put(hash, files)
return files, nil
}
// applyDelta merges a baseline's sorted list with a delta's sorted F
// cards: a card with a hash replaces or inserts, one without removes —
// and a rename card removes its OLD name from the baseline implicitly
// (§4.7.6: the four-token F card both places the new name and vacates
// the old one).
func applyDelta(base, delta []artifact.File) []artifact.File {
renamedAway := map[string]bool{}
for _, d := range delta {
if d.OldName != "" {
renamedAway[d.OldName] = true
}
}
out := make([]artifact.File, 0, len(base)+len(delta))
i, j := 0, 0
for i < len(base) || j < len(delta) {
switch {
case j >= len(delta) || (i < len(base) && base[i].Name < delta[j].Name):
if !renamedAway[base[i].Name] {
out = append(out, base[i])
}
i++
case i >= len(base) || base[i].Name > delta[j].Name:
if delta[j].Hash != "" {
out = append(out, delta[j])
}
j++
default: // same name
if delta[j].Hash != "" {
out = append(out, delta[j])
}
i++
j++
}
}
return out
}
// changes is the paths this check-in touched: additions, edits,
// deletions, and both sides of a rename. known is false when they could
// not be learned now but may be later — a tree check-in whose
// sub-manifests zit cannot read yet, or a call out of time — and the
// check-in is then folded as pending (index.Parsed.ChangesPending).
//
// A flat check-in is diffed here, from its manifests. A tree check-in,
// or a flat one whose parent or baseline is a tree, is zit's: quarry
// never walks a tree (see zitChanges).
func (fl *fileLists) changes(hash string, art *artifact.Artifact) (paths []string, known bool) {
paths, err := fl.flatChanges(hash, art)
switch {
case err == nil:
return paths, true
case errors.Is(err, errTree):
return fl.zitChanges(hash, len(art.Parents) == 0)
case errors.Is(err, context.DeadlineExceeded):
log.Printf("ingest: changes for %.12s: %v; left pending", hash, err)
return nil, false
}
// A parent or baseline the store does not hold: the check-in shows
// no changes, as crosslink's "files unknown" does on a rebuild.
if !errors.Is(err, errKnownMissing) {
log.Printf("ingest: changes for %.12s: %v", hash, err)
}
return nil, true
}
// zitChanges takes a check-in's changed paths from zit, as a rebuild
// takes them from crosslink: one call, whatever the tree's size, and
// the same set (zitcli.ChangedFiles). A root's changes are its whole
// file list. A failure, including a sub-manifest not yet in the store
// or a tree zit refuses, leaves the check-in pending: zit names no
// change it cannot read, and neither does the index.
func (fl *fileLists) zitChanges(hash string, root bool) ([]string, bool) {
paths, err := fl.zitChangesWithin(hash, root, liveCallTimeout)
if err != nil {
log.Printf("ingest: changes for %.12s: %v; left pending", hash, err)
return nil, false
}
return paths, true
}
// zitChangesWithin is zitChanges' call, bounded by d. A call that ran
// out of time answers an error wrapping context.DeadlineExceeded.
func (fl *fileLists) zitChangesWithin(hash string, root bool, d time.Duration) ([]string, error) {
ctx, cancel := callCtx(d)
defer cancel()
var paths []string
var err error
if root {
var entries []zitcli.LsEntry
if entries, err = fl.cli.LsNames(ctx, hash); err == nil {
paths = make([]string, len(entries))
for i, e := range entries {
paths[i] = e.Name
}
}
} else {
paths, err = fl.cli.ChangedFiles(ctx, hash)
}
if err != nil && ctx.Err() != nil {
return nil, fmt.Errorf("%w: %v", context.DeadlineExceeded, err)
}
return paths, err
}
func (fl *fileLists) flatChanges(hash string, art *artifact.Artifact) ([]string, error) {
full, err := fl.full(hash, art)
if err != nil {
return nil, err
}
fl.put(hash, full)
var paths []string
if art.Baseline != "" {
// The delta names its changes directly.
for _, f := range art.Files {
paths = append(paths, f.Name)
if f.OldName != "" {
paths = append(paths, f.OldName)
}
}
return paths, nil
}
if len(art.Parents) == 0 {
for _, f := range full {
paths = append(paths, f.Name)
}
return paths, nil
}
parent, err := fl.byHash(art.Parents[0])
if err != nil {
return nil, err
}
return diffLists(parent, full), nil
}
// diffLists is the symmetric difference of two sorted file lists by
// name, treating a hash or permission change as a change.
func diffLists(old, new []artifact.File) []string {
var paths []string
i, j := 0, 0
for i < len(old) || j < len(new) {
switch {
case j >= len(new) || (i < len(old) && old[i].Name < new[j].Name):
paths = append(paths, old[i].Name) // deleted
i++
case i >= len(old) || old[i].Name > new[j].Name:
paths = append(paths, new[j].Name) // added
j++
default:
if old[i].Hash != new[j].Hash || old[i].Perm != new[j].Perm {
paths = append(paths, new[j].Name)
}
i++
j++
}
}
sort.Strings(paths)
return paths
}
// DeriveChanges fills Parsed.Changes for a batch the ARTIFACT way —
// the pre-crosslink derivation, retained as the parity reference: the
// harness proves a crosslink-built index equal to one whose changes
// came from expanding manifests.
func DeriveChanges(cli *zitcli.Client, batch []index.Parsed) {
fl := newFileLists(cli)
fl.spool = map[string][]byte{}
fl.annotate(batch)
}
// annotate fills Parsed.Changes for the check-ins in a batch, and
// Parsed.Stub for the A cards whose source blob is a quarry stub.
func (fl *fileLists) annotate(batch []index.Parsed) {
for i := range batch {
if batch[i].Art == nil {
continue
}
switch batch[i].Art.Type {
case artifact.Manifest:
changes, known := fl.changes(batch[i].Hash, batch[i].Art)
batch[i].Changes, batch[i].ChangesPending = changes, !known
if len(changes) > 0 && len(batch[i].Art.Parents) > 0 { // a root: see rootless
batch[i].Lines = fl.lineCounts(batch[i].Hash, changes)
}
case artifact.Attachment:
batch[i].Stub = fl.stub(batch[i].Art)
}
}
}
// annotateStubs is annotate's attachment half alone. The reindex path
// takes a check-in's file changes from zit's crosslink rather than
// from changes(), which is the expensive half, but it still has to
// learn which A cards name a stub — without this a rebuilt index folds
// every uploaded attachment as the 150-byte text file the stub is, and
// a picture that rendered yesterday becomes five lines of key-value
// text. That is exactly what a rebuild did to the live forge.
func (fl *fileLists) annotateStubs(batch []index.Parsed) {
for i := range batch {
if batch[i].Art != nil && batch[i].Art.Type == artifact.Attachment {
batch[i].Stub = fl.stub(batch[i].Art)
}
}
}
// stub reads the front of an A card's source blob and parses it when
// the first line says it is a stub (RFC 0016 §The stub). This is the
// one read of attachment bytes the ingest makes, and it is bounded: a
// real file costs its first kilobyte and a stub is never larger. A
// blob this store does not hold, or a malformed stub, folds as a file
// — the card then says what the head op finds, which is the honest
// answer when the stub cannot be trusted.
func (fl *fileLists) stub(a *artifact.Artifact) *attach.Stub {
at := a.Attachment
if fl.cli == nil || at == nil || at.Source == "" {
return nil
}
ctx, cancel := liveCtx()
defer cancel()
front, err := fl.cli.ArtifactHead(ctx, at.Source, attach.MaxStubBytes+1)
if err != nil || !attach.IsStub(front) {
return nil
}
s, err := attach.Parse(front)
if err != nil {
log.Printf("ingest: %s: attachment %s looks like a stub and is not: %v", at.Source, at.Name, err)
return nil
}
return &s
}
// lineCounts asks zit to count a check-in's changed lines, the way
// crosslink counts them on a rebuild, so a check-in folded live shows
// the same "+N −N" a rebuilt index would. One process per new check-in;
// a failure leaves the counts out, which reads as uncounted.
//
// numstat lists only files with lines to count, where crosslink counts
// every changed file: a mode change or an empty file added is "0 0"
// there and absent here. So a changed path numstat left out is counted
// as zero, once numstat has answered at all.
func (fl *fileLists) lineCounts(hash string, changes []string) map[string]index.LineCount {
return fl.lineCountsWithin(hash, changes, liveCallTimeout)
}
func (fl *fileLists) lineCountsWithin(hash string, changes []string, d time.Duration) map[string]index.LineCount {
ctx, cancel := callCtx(d)
defer cancel()
counts, err := fl.cli.NumStat(ctx, hash)
if err != nil {
log.Printf("ingest: line counts for %.12s: %v", hash, err)
return nil
}
out := make(map[string]index.LineCount, len(changes))
for path, lc := range counts {
out[path] = index.LineCount{Added: lc.Added, Removed: lc.Removed}
}
for _, path := range changes {
if _, ok := out[path]; !ok {
out[path] = index.LineCount{}
}
}
return out
}