Artifact [1e51005e84]

Artifact 1e51005e840f168ed06704957b0b03082ed3761819b3dae4bad0cea770cc68c8:

• File internal/content/sparse_corpus_test.go — part of check-in [b28c06506b] at 2026-10-22 02:57:50 on branch feat/horizon-cones — fix(content): classify sparse stores from crosslink tables, not the whole store

  Sparse classification read, inflated and delta-expanded every held blob, file content included, under a whole-store byte budget. On a clone of the Fossil SCM repository (67,894 blobs, 8.4 GB expanded) LoadAbsences failed at once with "byte budget exceeded", so no path-cone or content-horizon store of a real repository could classify its phantoms, and pull, verify and compaction failed with it.

  Reference facts now come from the tables wherever they provably stand in for raw cards. A blob the crosslinker recorded as no artifact and mlink names as a file version is never read. A completely linked check-in (event row, no orphan row, a primary parent edge) whose primary parent and baseline trees are themselves reproduced contributes its mlink rows as F cards and its plink, cherrypick and tagxref rows as references; by induction over the primary chain those rows give the same (name, file) pairs as its raw cards. Everything else is still read raw: unlinked and orphaned check-ins, roots, check-ins whose parent never arrived, other artifacts, and every check-in once the store has seen a sub-manifest, since trees are not crosslinked. Budgets are charged per round for the selected blobs and their delta ancestors only, and a base's bytes are released once its last delta is applied.

  A content horizon now reads only its held set's manifests and the baselines and nodes they reach, on demand, with one visited set and a byte budget scaled to the held set.

  On the Fossil SCM clone, cone {src} classifies in about 2 s at a 65 MiB peak heap, reading 8,655 of 67,894 blobs; the inside/outside sets equal the all-raw result, and Outside/Required agree on every artifact. (user: [email protected], size: 4426) [annotate] [blame] [check-ins using] [more...]

package content_test

import (
	"context"
	"io"
	"os"
	"path/filepath"
	"runtime"
	"sync"
	"testing"
	"time"

	"github.com/danmestas/go-libfossil/internal/cone"
	"github.com/danmestas/go-libfossil/internal/content"
	"github.com/danmestas/go-libfossil/internal/repo"
)

// TestCorpusSparseClassification classifies a real repository's phantoms
// under the store cone {src}, then under a content horizon at a recent trunk
// check-in, and reports wall time, reads and peak heap for each. It is
// skipped unless LIBFOSSIL_CORPUS names a repository; it works on a copy.
func TestCorpusSparseClassification(t *testing.T) {
	corpus := os.Getenv("LIBFOSSIL_CORPUS")
	if corpus == "" {
		t.Skip("set LIBFOSSIL_CORPUS to a .fossil repository to run")
	}
	work := filepath.Join(t.TempDir(), "corpus.fossil")
	if err := corpusCopy(work, corpus); err != nil {
		t.Fatalf("copy corpus: %v", err)
	}
	r, err := repo.Open(work)
	if err != nil {
		t.Fatalf("open corpus: %v", err)
	}
	t.Cleanup(func() { r.Close() })
	// Elide a few old file versions: some named only outside src/, and some
	// src/ versions no recent check-in names, for the horizon.
	for _, query := range []string{
		`SELECT b.uuid FROM mlink m JOIN filename f ON f.fnid=m.fnid JOIN blob b ON b.rid=m.fid
		 WHERE b.size>=0 AND f.name NOT LIKE 'src/%' ORDER BY m.mid LIMIT 5`,
		`SELECT b.uuid FROM mlink m JOIN filename f ON f.fnid=m.fnid JOIN blob b ON b.rid=m.fid
		 WHERE b.size>=0 AND f.name LIKE 'src/%' ORDER BY m.mid LIMIT 5`,
	} {
		var uuids []string
		rows, err := r.DB().Query(query)
		if err != nil {
			t.Fatal(err)
		}
		for rows.Next() {
			var uuid string
			if err := rows.Scan(&uuid); err != nil {
				t.Fatal(err)
			}
			uuids = append(uuids, uuid)
		}
		if err := rows.Close(); err != nil {
			t.Fatal(err)
		}
		for _, uuid := range uuids {
			hzPhantomize(t, r, uuid)
		}
	}
	path, err := cone.New([]string{"src"}, nil)
	if err != nil {
		t.Fatal(err)
	}
	if a := corpusClassify(t, r, "cone {src}", path); a.Count(content.Elided) == 0 {
		t.Error("cone {src} elided nothing")
	}
	var c string
	if err := r.DB().QueryRow(`SELECT b.uuid FROM event e JOIN blob b ON b.rid=e.objid
		JOIN tagxref x ON x.rid=e.objid AND x.tagtype>0 AND x.value='trunk'
		JOIN tag g ON g.tagid=x.tagid AND g.tagname='branch'
		WHERE e.type='ci' ORDER BY e.mtime DESC LIMIT 1 OFFSET 20`).Scan(&c); err != nil {
		t.Fatalf("recent trunk check-in: %v", err)
	}
	h, err := cone.NewHorizon(c, cone.RuleDescendants, cone.DepthContent)
	if err != nil {
		t.Fatal(err)
	}
	both, err := cone.NewWithHorizon([]string{"src"}, nil, h)
	if err != nil {
		t.Fatal(err)
	}
	if a := corpusClassify(t, r, "cone {src} + horizon "+c[:10], both); a.Count(content.Elided) <= 5 {
		t.Error("the horizon elided no old src/ version")
	}
}

func corpusClassify(t *testing.T, r *repo.Repo, label string, c *cone.Cone) *content.Absences {
	t.Helper()
	if err := content.SetStoreCone(r.DB(), c); err != nil {
		t.Fatal(err)
	}
	runtime.GC()
	stop := corpusHeapPeak()
	start := time.Now()
	a, work, err := content.LoadAbsencesWork(context.Background(), r.DB())
	elapsed := time.Since(start)
	peak := stop()
	if err != nil {
		t.Fatalf("%s: %v (work %+v)", label, err, work)
	}
	t.Logf("%s: %v, peak heap %d MiB, elided %d transit %d shunned %d", label, elapsed,
		peak>>20, a.Count(content.Elided), a.Count(content.Transit), a.Count(content.Shunned))
	t.Logf("%s: work %+v", label, work)
	return a
}

// corpusHeapPeak samples the live heap until stopped, which returns the
// peak, a last reading included.
func corpusHeapPeak() func() uint64 {
	var mu sync.Mutex
	var max uint64
	done := make(chan struct{})
	var wg sync.WaitGroup
	sample := func() {
		var m runtime.MemStats
		runtime.ReadMemStats(&m)
		mu.Lock()
		if m.HeapAlloc > max {
			max = m.HeapAlloc
		}
		mu.Unlock()
	}
	wg.Add(1)
	go func() {
		defer wg.Done()
		tick := time.NewTicker(5 * time.Millisecond)
		defer tick.Stop()
		for {
			select {
			case <-done:
				return
			case <-tick.C:
				sample()
			}
		}
	}()
	stop := func() uint64 {
		sample()
		close(done)
		wg.Wait()
		return max
	}
	return stop
}

func corpusCopy(dst, src string) error {
	in, err := os.Open(src)
	if err != nil {
		return err
	}
	defer in.Close()
	out, err := os.OpenFile(dst, os.O_WRONLY|os.O_CREATE|os.O_EXCL, 0o600)
	if err != nil {
		return err
	}
	if _, err := io.Copy(out, in); err != nil {
		out.Close()
		return err
	}
	return out.Close()
}