Skip to content
quarry Artifact [6d9fdcbda6]
Login

Artifact [6d9fdcbda6]

Check-ins Using Download Hex

Artifact 6d9fdcbda616403b46d7525ab3cd1b540cc43c2a5f701c2b1d5ca560e78d19fc:

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(10, "", "", "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(10, "", "", "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(10, "", "", "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(10, "", "", "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(10, "", "", "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(10, "", "", "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(10, "", "", "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(commits+50, "", "", "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)
	}
}