Artifact [3437efe39d]

Artifact 3437efe39dc70256d35ccecf1bb4461a38f1ca72f067828d706c43decff65497:

• File internal/ingest/ingest_test.go — part of check-in [ff7f5d3d69] at 2026-08-23 22:03:24 on branch trunk — check-in page: tag history, branch diff, manifest view

  Fossil lets an author correct a check-in comment after the fact; quarry showed the superseded C card as if it were current, on 697 fossil check-ins. The correction is now applied at read, batched in AttachCI, with the original kept beside it — the stored card is an artifact fact and stays untouched. hidden and closed surface too.

  Branch diff is the branch against its fork point, which is not the landing page's base (the target's current tip). Offered only where a fork point exists.

  The manifest view renders the eight cards that are not the file list and counts the other 980, linking the tree.

  Also: unified/side-by-side now swaps in place via datastar, keeping the window and correcting the URL; the wiki note panel lets blocks use the full width while prose keeps its measure. (user: dmestas, size: 10330) [annotate] [blame] [check-ins using]

package ingest

import (
	"errors"
	"fmt"
	"os"
	"os/exec"
	"path/filepath"
	"strings"
	"testing"

	"github.com/nats-io/nats.go"

	"craftdesign.group/quarry/internal/events"
	"craftdesign.group/quarry/internal/index"
	"craftdesign.group/quarry/internal/zitcli"
)

func run(t *testing.T, dir string, args ...string) {
	t.Helper()
	cmd := exec.Command("zit", args...)
	cmd.Dir = dir
	if out, err := cmd.CombinedOutput(); err != nil {
		t.Fatalf("zit %v: %v\n%s", args, err, out)
	}
}

func commitFile(t *testing.T, repo, name, body, msg string) {
	t.Helper()
	if err := os.WriteFile(filepath.Join(repo, name), []byte(body), 0o644); err != nil {
		t.Fatal(err)
	}
	run(t, repo, "add", name)
	run(t, repo, "commit", "-m", msg)
}

func testPipeline(t *testing.T) (*Pipeline, *index.Index, string) {
	t.Helper()
	if _, err := exec.LookPath("zit"); err != nil {
		t.Skip("zit not on PATH")
	}
	repo := t.TempDir()
	run(t, repo, "init")
	commitFile(t, repo, "a.txt", "alpha\n", "add a")
	commitFile(t, repo, "b.txt", "beta\n", "add b")

	ev, err := events.Start(filepath.Join(t.TempDir(), "nats"))
	if err != nil {
		t.Fatal(err)
	}
	t.Cleanup(ev.Close)
	ix, err := index.Open(filepath.Join(t.TempDir(), "index.db"))
	if err != nil {
		t.Fatal(err)
	}
	t.Cleanup(func() { ix.Close() })
	return NewPipeline(zitcli.New(repo), ev, ix, "test", "n1"), ix, repo
}

func TestPipelineBootstrapAndIncremental(t *testing.T) {
	pipe, ix, repo := testPipeline(t)

	n, err := pipe.Sync()
	if err != nil {
		t.Fatal(err)
	}
	if n == 0 {
		t.Fatal("bootstrap published nothing")
	}
	tl, _, err := ix.Timeline(index.TimelineFilter{Limit: 10, Kind: "ci"})
	if err != nil {
		t.Fatal(err)
	}
	if len(tl) != 2 {
		t.Fatalf("timeline has %d entries: %+v", len(tl), tl)
	}
	comments := map[string]bool{tl[0].Comment: true, tl[1].Comment: true}
	if !comments["add a"] || !comments["add b"] {
		t.Fatalf("timeline contents: %+v", tl)
	}

	// Current cursor: nothing published, nothing re-folded.
	if n, err = pipe.Sync(); err != nil || n != 0 {
		t.Fatalf("idle sync: n=%d err=%v", n, err)
	}

	// Append flows store → events → index.
	commitFile(t, repo, "c.txt", "gamma\n", "add c")
	if n, err = pipe.Sync(); err != nil || n == 0 {
		t.Fatalf("incremental sync: n=%d err=%v", n, err)
	}
	tl, _, _ = ix.Timeline(index.TimelineFilter{Limit: 10, Kind: "ci"})
	if len(tl) != 3 {
		t.Fatalf("timeline after append: %+v", tl)
	}
}

func TestPipelineRecoversFromStoreRewrite(t *testing.T) {
	pipe, ix, repo := testPipeline(t)
	if _, err := pipe.Sync(); err != nil {
		t.Fatal(err)
	}
	before, _, _ := ix.Timeline(index.TimelineFilter{Limit: 10, Kind: "ci"})

	// Rewrite the store under the cursor: shun a blob and reclaim.
	head, err := exec.Command("zit", "ls", "-R", repo, "tip").Output()
	if err != nil {
		t.Fatal(err)
	}
	blob := strings.Fields(string(head))[0]
	run(t, repo, "shun", blob, "--force")

	// The reset event and re-census flow through the stream; the index
	// converges to the same timeline.
	if _, err := pipe.Sync(); err != nil {
		t.Fatalf("sync after rewrite should reset and rebuild, got %v", err)
	}
	after, _, _ := ix.Timeline(index.TimelineFilter{Limit: 10, Kind: "ci"})
	if len(after) != len(before) {
		t.Fatalf("rebuild diverged: %d vs %d check-ins", len(after), len(before))
	}
}

func TestPipelineSurvivesIndexerRestart(t *testing.T) {
	// The durable consumer's position lives in the stream, not the
	// process: a fresh Pipeline over the same events+index must not
	// re-fold or lose anything.
	pipe, ix, repo := testPipeline(t)
	if _, err := pipe.Sync(); err != nil {
		t.Fatal(err)
	}
	commitFile(t, repo, "d.txt", "delta\n", "add d")
	fresh := NewPipeline(pipe.cli, pipe.ev, pipe.ix, "test", "n1")
	if _, err := fresh.Sync(); err != nil {
		t.Fatal(err)
	}
	tl, _, _ := ix.Timeline(index.TimelineFilter{Limit: 10, Kind: "ci"})
	if len(tl) != 3 {
		t.Fatalf("timeline after restart: %+v", tl)
	}
}

func TestReindexRebuildsFromZero(t *testing.T) {
	pipe, ix, _ := testPipeline(t)
	if _, err := pipe.Sync(); err != nil {
		t.Fatal(err)
	}
	before, _, _ := ix.Timeline(index.TimelineFilter{Limit: 10, Kind: "ci"})
	if len(before) == 0 {
		t.Fatal("no baseline")
	}
	n, err := pipe.Reindex()
	if err != nil {
		t.Fatal(err)
	}
	if n == 0 {
		t.Fatal("reindex republished nothing")
	}
	after, _, _ := ix.Timeline(index.TimelineFilter{Limit: 10, Kind: "ci"})
	if len(after) != len(before) {
		t.Fatalf("reindex diverged: %d vs %d", len(after), len(before))
	}
	// A reindexed index answers idle syncs normally afterwards.
	if n, err := pipe.Sync(); err != nil || n != 0 {
		t.Fatalf("post-reindex idle sync: n=%d err=%v", n, err)
	}
}

// ZITEVT is capped and discards OLD messages, so a census larger than
// the cap once deleted its own head while the indexer was still folding
// it: the index came out silently partial and then reported itself
// done. sqlite indexed 41,478 of 131,558 artifacts that way, losing
// every ticket, wiki page and forum post. Publishing must yield to
// folding before the stream can overrun.
func TestLargeCensusSurvivesACappedStream(t *testing.T) {
	p, ix, repo := testPipeline(t)

	// A stream far smaller than the census, and a flush threshold small
	// enough that a test-sized corpus crosses it.
	if _, err := p.ev.JS().UpdateStream(&nats.StreamConfig{
		Name:     events.StreamName,
		Subjects: []string{"zit.evt.>"},
		Storage:  nats.FileStorage,
		Discard:  nats.DiscardOld,
		MaxBytes: 128 << 10,
	}); err != nil {
		t.Fatal(err)
	}
	old := flushBytes
	flushBytes = 8 << 10
	t.Cleanup(func() { flushBytes = old })

	const commits = 60
	body := strings.Repeat("padding to make each manifest substantial\n", 40)
	for i := 0; i < commits; i++ {
		commitFile(t, repo, fmt.Sprintf("f%02d.txt", i), body+fmt.Sprint(i), fmt.Sprintf("commit %d", i))
	}
	if _, err := p.Sync(); err != nil {
		t.Fatal(err)
	}
	tl, _, err := ix.Timeline(index.TimelineFilter{Limit: commits + 50, Kind: "ci"})
	if err != nil {
		t.Fatal(err)
	}
	if want := commits + 2; len(tl) != want { // +2 from testPipeline's fixture
		t.Fatalf("indexed %d check-ins, want %d — the stream dropped events the indexer never folded", len(tl), want)
	}
}

// The census must be handed over AS IT IS READ, never collected first:
// collecting it holds the whole stream in memory, 13.2GB for sqlite,
// which is the standing barrier to a store of real size.
//
// What this test can prove cheaply is the half that a unit test can
// see — that a caller may STOP mid-census and the read stops with it,
// which a buffering reader cannot offer at any price. The memory claim
// itself is measured against a real corpus, not asserted here.
func TestCensusReadCanBeAbandonedMidStream(t *testing.T) {
	p, _, repo := testPipeline(t)

	old := flushBytes
	flushBytes = 1 // flush after every artifact, so each one is observable
	t.Cleanup(func() { flushBytes = old })

	for i := 0; i < 12; i++ {
		commitFile(t, repo, fmt.Sprintf("s%02d.txt", i), strings.Repeat("x", 200), fmt.Sprintf("c%d", i))
	}

	// The whole census still arrives when nobody interrupts.
	seen := 0
	_, err := p.cli.FramesEach(zitcli.Cursor{}, func(a zitcli.Artifact) error {
		seen++
		return nil
	})
	if err != nil {
		t.Fatal(err)
	}
	if seen < 10 {
		t.Fatalf("only %d artifacts streamed, expected the whole census", seen)
	}

	// And an error mid-stream must stop the read rather than run to the
	// end, so a caller can abandon a 13GB census cheaply.
	stopped := 0
	_, err = p.cli.FramesEach(zitcli.Cursor{}, func(a zitcli.Artifact) error {
		stopped++
		if stopped == 2 {
			return errors.New("caller gave up")
		}
		return nil
	})
	if err == nil {
		t.Fatal("an error from the callback did not stop the read")
	}
	if stopped != 2 {
		t.Fatalf("read continued past the callback error: %d artifacts", stopped)
	}
}

// The event stream is a bounded live tail, so a sustained fold failure
// can let unacked messages age out; the publisher's cursor then moves
// past them and nothing republishes. Nothing errors at the moment of
// loss, which is why a round that SUCCEEDS must still prove the index
// is whole — and repair it from the store when it is not.
func TestSyncRepairsAGapItFinds(t *testing.T) {
	p, ix, _ := testPipeline(t)

	if _, err := p.Sync(); err != nil {
		t.Fatal(err)
	}
	before, err := ix.DanglingParents(50)
	if err != nil {
		t.Fatal(err)
	}
	if len(before) != 0 {
		t.Fatalf("the fixture starts with %d dangling parents", len(before))
	}

	// Punch out a check-in the way a lost artifact does: the row goes,
	// the edge that referenced it stays.
	var victim string
	if err := ix.RawDB().QueryRow(
		`SELECT parent FROM checkin_parent LIMIT 1`).Scan(&victim); err != nil {
		t.Skipf("fixture has no parent edges: %v", err)
	}
	if _, err := ix.RawDB().Exec(`DELETE FROM checkin WHERE hash = ?`, victim); err != nil {
		t.Fatal(err)
	}
	if d, _ := ix.DanglingParents(50); len(d) == 0 {
		t.Fatal("could not create a gap to repair")
	}

	// A successful round must notice and rebuild from the store.
	if _, err := p.Sync(); err != nil {
		t.Fatal(err)
	}
	after, err := ix.DanglingParents(50)
	if err != nil {
		t.Fatal(err)
	}
	if len(after) != 0 {
		t.Errorf("after a round the index still has %d dangling edges; the gap was not repaired", len(after))
	}
}

// A rebuild that cannot close a gap must be reported once, not retried
// forever: replaying a large store costs minutes, and doing it every
// poll would make an unrepairable index consume the machine instead of
// telling anyone. This pins the "try once, then report" contract.
func TestSyncDoesNotRebuildForeverOnAnUnrepairableGap(t *testing.T) {
	p, ix, _ := testPipeline(t)
	if _, err := p.Sync(); err != nil {
		t.Fatal(err)
	}

	// A dangling edge whose parent the store does NOT hold — the shape
	// fossil-born histories legitimately have. It must never trigger a
	// rebuild at all.
	if _, err := ix.RawDB().Exec(
		`INSERT INTO checkin_parent (child, parent, ord)
		 SELECT hash, ?, 9 FROM checkin LIMIT 1`, "ff"+strings.Repeat("00", 19)); err != nil {
		t.Fatal(err)
	}
	before := p.Stats()
	if _, err := p.Sync(); err != nil {
		t.Fatal(err)
	}
	after := p.Stats()
	if after.Gaps != 0 {
		t.Errorf("an ancestor the store never held is counted as loss (gaps=%d); "+
			"every fossil-born repository would rebuild on every poll", after.Gaps)
	}
	_ = before
}