package ingest
import (
"bytes"
"compress/zlib"
"crypto/md5"
"crypto/sha3"
"encoding/binary"
"encoding/hex"
"fmt"
"io"
"net"
"net/http"
"os"
"os/exec"
"path/filepath"
"strings"
"testing"
"time"
"github.com/nats-io/nats.go"
"craftdesign.group/quarry/internal/xfer"
)
// sealed closes cards with their Z card and names the artifact.
func sealed(cards string) ([]byte, string) {
sum := md5.Sum([]byte(cards))
raw := []byte(cards + "Z " + hex.EncodeToString(sum[:]) + "\n")
h := sha3.Sum256(raw)
return raw, hex.EncodeToString(h[:])
}
// pushRaw pushes artifacts into the repository as they are, through
// `zit serve`, the way a peer's sync round delivers them: no client
// checks what they say.
func (r *treeRig) pushRaw(t *testing.T, arts ...[]byte) {
t.Helper()
code, err := xfer.ProjectCode(r.repo)
if err != nil {
t.Fatal(err)
}
l, err := net.Listen("tcp", "127.0.0.1:0")
if err != nil {
t.Fatal(err)
}
port := l.Addr().(*net.TCPAddr).Port
l.Close()
srv := exec.Command("zit", "serve", "--port", fmt.Sprint(port))
srv.Dir, srv.Env = r.repo, r.env
if err := srv.Start(); err != nil {
t.Fatal(err)
}
defer func() { srv.Process.Kill(); srv.Wait() }()
body := "push " + code + " " + code + "\n"
for _, a := range arts {
h := sha3.Sum256(a)
body += fmt.Sprintf("file %s %d\n%s", hex.EncodeToString(h[:]), len(a), a)
}
var buf bytes.Buffer
var n [4]byte
binary.BigEndian.PutUint32(n[:], uint32(len(body)))
buf.Write(n[:])
zw := zlib.NewWriter(&buf)
zw.Write([]byte(body))
zw.Close()
url := fmt.Sprintf("http://127.0.0.1:%d/xfer", port)
for range 50 {
resp, err := http.Post(url, "application/x-fossil", bytes.NewReader(buf.Bytes()))
if err != nil {
time.Sleep(100 * time.Millisecond)
continue
}
io.Copy(io.Discard, resp.Body)
resp.Body.Close()
if resp.StatusCode != http.StatusOK {
t.Fatalf("push: %s", resp.Status)
}
for _, a := range arts {
h := sha3.Sum256(a)
if _, err := r.cli.ArtifactRaw(t.Context(), hex.EncodeToString(h[:]), 1<<20); err != nil {
t.Fatalf("rig: %.12s was not stored: %v", hex.EncodeToString(h[:]), err)
}
}
return
}
t.Fatalf("push: zit serve never answered")
}
// indexedTree is a repository whose index holds an imported tree
// history, and the tip and one of its file blobs to build on.
func indexedTree(t *testing.T) (r *treeRig, tip, blob string) {
t.Helper()
r = newTreeRig(t)
r.grow(t, 8)
r.push(t, true)
if _, err := r.p.Sync(); err != nil {
t.Fatal(err)
}
tip, _, err := r.cli.Info(t.Context(), "trunk")
if err != nil {
t.Fatal(err)
}
files, err := r.cli.Ls(t.Context(), tip)
if err != nil || len(files) == 0 {
t.Fatalf("rig: no files at the tip (%v)", err)
}
if pending, err := r.ix.ChangesPending(); err != nil || pending {
t.Fatalf("rig: changes pending before anything was crafted (%v)", err)
}
return r, tip, files[0].Blob
}
// agreesWithZit checks the index's change rows for a check-in against
// zit's own answer: what zit names, the index records; when zit cannot
// name them, the index records none and says it is behind.
func (r *treeRig) agreesWithZit(t *testing.T, hash string) (zitRefused bool) {
t.Helper()
if _, _, ok, err := r.ix.CheckinByPrefix(hash); err != nil || !ok {
t.Errorf("FAIL: check-in %.12s is not indexed (%v)", hash, err)
}
stats, err := r.ix.ChangeStats([]string{hash}, "")
if err != nil {
t.Fatal(err)
}
got, recorded := stats[hash]
pending, err := r.ix.ChangesPending()
if err != nil {
t.Fatal(err)
}
want, zerr := r.cli.ChangedFiles(t.Context(), hash)
if zerr != nil {
if recorded {
t.Errorf("FAIL: zit names no changes for %.12s (%v), the index records %+v", hash, zerr, got)
}
if !pending {
t.Errorf("FAIL: %.12s is folded without changes and the index does not say it is behind", hash)
}
return true
}
if !recorded || got.Files != len(want) || !got.Counted {
t.Errorf("FAIL: zit names %d changed files for %.12s, the index records %+v", len(want), hash, got)
}
if pending {
t.Errorf("FAIL: %.12s is resolved and the index still says it is behind", hash)
}
return false
}
// A tree check-in can reach a round before its sub-manifests: a push
// split across sync rounds, a mirror converging over several pulls. It
// is indexed at once, without changes and as pending; the round that
// brings its nodes completes it, with no reindex.
func TestATreeCheckinWhoseNodesArriveLaterIsCompleted(t *testing.T) {
r, tip, blob := indexedTree(t)
leaf, leafHash := sealed("F new/late.txt " + blob + "\n")
ci, ciHash := sealed("C nodes\\slater\nD 2026-11-14T00:00:01\nP " + tip + "\nS new/late.txt " + leafHash + "\nU ana\n")
r.pushRaw(t, ci)
if _, err := r.p.Sync(); err != nil {
t.Fatalf("FAIL: the round with the check-in alone: %v", err)
}
if !r.agreesWithZit(t, ciHash) {
t.Fatalf("rig: zit read %.12s before its node arrived", ciHash)
}
r.pushRaw(t, leaf)
if _, err := r.p.Sync(); err != nil {
t.Fatalf("FAIL: the round with the node: %v", err)
}
// Not due yet: a retry waits out its backoff, not every round.
if pending, err := r.ix.ChangesPending(); err != nil || !pending {
t.Errorf("FAIL: retried before its backoff (pending %v, %v)", pending, err)
}
r.later(pendingDelay(0))
if _, err := r.p.Sync(); err != nil {
t.Fatalf("FAIL: the round after the backoff: %v", err)
}
if r.agreesWithZit(t, ciHash) {
t.Errorf("FAIL: zit still cannot read %.12s with its node present", ciHash)
}
if !t.Failed() {
t.Log("ok")
}
}
// A tree whose S cards name what is not a sub-manifest — a file blob,
// another check-in's manifest — or whose leaves overlap and run out of
// order never gets changes zit would not name: never a vanished
// directory, never another check-in's files spliced in.
func TestACraftedTreeGetsNoChangesZitWouldNotName(t *testing.T) {
r, tip, blob := indexedTree(t)
leafA, leafAHash := sealed("F d1/x.txt " + blob + "\nF d3/z.txt " + blob + "\n")
leafB, leafBHash := sealed("F d2/y.txt " + blob + "\n")
checkin := func(n int, scards string) ([]byte, string) {
return sealed(fmt.Sprintf("C crafted\\s%d\nD 2026-11-14T00:00:0%d\nP %s\n%sU ana\n", n, n, tip, scards))
}
toBlob, toBlobHash := checkin(1, "S a.txt "+blob+"\n")
toCheckin, toCheckinHash := checkin(2, "S a.txt "+tip+"\n")
overlap, overlapHash := checkin(3, "S d1/x.txt "+leafAHash+"\nS d2/y.txt "+leafBHash+"\n")
r.pushRaw(t, leafA, leafB, toBlob, toCheckin, overlap)
if _, err := r.p.Sync(); err != nil {
t.Fatalf("FAIL: the round: %v", err)
}
for name, h := range map[string]string{"an S card naming a file": toBlobHash,
"an S card naming a check-in": toCheckinHash, "overlapping leaves": overlapHash} {
if !r.agreesWithZit(t, h) {
t.Logf("zit reads %s (%.12s); the index agrees with it", name, h)
}
}
crafted := []string{toBlobHash, toCheckinHash, overlapHash}
// Each is retried once per backoff, pendingAttempts times, then
// given up: it keeps no changes, and the index stops saying it is
// behind. No reindex.
attempts := func() map[string]int {
rows, err := r.ix.PendingChanges(1000, r.now.Add(1000*time.Hour))
if err != nil {
t.Fatal(err)
}
out := map[string]int{}
for _, c := range rows {
out[c.Hash] = c.Attempts
}
return out
}
for n := 1; n <= pendingAttempts; n++ {
r.later(pendingDelay(n - 1))
if _, err := r.p.Sync(); err != nil {
t.Fatal(err)
}
got := attempts()
for _, h := range crafted {
if got[h] != n {
t.Errorf("FAIL: %.12s tried %d times after %d backoffs (retry not running?)", h, got[h], n)
}
}
}
r.later(pendingDelay(pendingAttempts))
if _, err := r.p.Sync(); err != nil {
t.Fatal(err)
}
if pending, err := r.ix.ChangesPending(); err != nil || pending {
t.Errorf("FAIL: still pending after %d attempts (%v): every page would say the index is rebuilding", pendingAttempts, err)
}
stats, err := r.ix.ChangeStats(crafted, "")
if err != nil {
t.Fatal(err)
}
if len(stats) != 0 {
t.Errorf("FAIL: a given-up check-in shows changes: %+v", stats)
}
for _, h := range crafted {
if _, _, ok, err := r.ix.CheckinByPrefix(h); err != nil || !ok {
t.Errorf("FAIL: given-up check-in %.12s left the timeline (%v)", h, err)
}
}
// The age bound ends a row however few times it was tried.
late, lateHash := checkin(4, "S b.txt "+blob+"\n")
r.pushRaw(t, late)
if _, err := r.p.Sync(); err != nil {
t.Fatal(err)
}
if got := attempts(); len(got) != 1 || got[lateHash] != 0 {
t.Fatalf("rig: pending %v, want only %.12s untried", got, lateHash)
}
r.later(pendingMaxAge)
if _, err := r.p.Sync(); err != nil {
t.Fatal(err)
}
if pending, err := r.ix.ChangesPending(); err != nil || pending {
t.Errorf("FAIL: still pending %s after its first fold (%v)", pendingMaxAge, err)
}
if !t.Failed() {
t.Log("ok")
}
}
// A batch whose annotation outlasts ackWait is not redelivered while it
// is still being worked on.
func TestABatchStillBeingAnnotatedIsNotRedelivered(t *testing.T) {
p, _, _ := testPipeline(t)
old := ackWait
ackWait = 300 * time.Millisecond
t.Cleanup(func() { ackWait = old })
if _, err := p.ev.JS().Publish(p.subject("artifact"), []byte("{}")); err != nil {
t.Fatal(err)
}
sub, err := p.ev.JS().PullSubscribe(p.subject("*"), "keepworking", nats.AckWait(ackWait))
if err != nil {
t.Fatal(err)
}
msgs, err := sub.Fetch(1, nats.MaxWait(time.Second))
if err != nil || len(msgs) != 1 {
t.Fatalf("rig: fetched %d (%v)", len(msgs), err)
}
stop := keepWorking(msgs)
time.Sleep(4 * ackWait)
again, _ := sub.Fetch(1, nats.MaxWait(100*time.Millisecond))
stop()
if len(again) != 0 {
t.Errorf("FAIL: a message still being worked on was redelivered")
}
if err := msgs[0].Ack(); err != nil {
t.Fatal(err)
}
if !t.Failed() {
t.Log("ok")
}
}
// A zit call that runs out of time leaves its check-in folded and
// pending, never with a guess; a later round, after the backoff,
// completes it.
func TestATimedOutCallLeavesTheCheckinPendingUntilALaterRound(t *testing.T) {
r := newTreeRig(t)
r.grow(t, 8)
r.push(t, true)
old := liveCallTimeout
liveCallTimeout = time.Nanosecond
_, err := r.p.Sync()
liveCallTimeout = old
if err != nil {
t.Fatalf("FAIL: a round whose zit calls ran out of time failed: %v", err)
}
tip, _, err := r.cli.Info(t.Context(), "trunk")
if err != nil {
t.Fatal(err)
}
if _, _, ok, err := r.ix.CheckinByPrefix(tip); err != nil || !ok {
t.Errorf("FAIL: the tip %.12s was not folded (%v)", tip, err)
}
if stats, _ := r.ix.ChangeStats([]string{tip}, ""); len(stats) != 0 {
t.Errorf("FAIL: changes recorded without zit's answer: %+v", stats)
}
if pending, err := r.ix.ChangesPending(); err != nil || !pending {
t.Errorf("FAIL: nothing pending after the calls ran out of time (%v)", err)
}
// Later rounds, each trying a bounded number again once due.
for n := range 4 {
r.later(pendingDelay(n))
if _, err := r.p.Sync(); err != nil {
t.Fatal(err)
}
}
if pending, err := r.ix.ChangesPending(); err != nil || pending {
t.Errorf("FAIL: changes still pending after later rounds (%v)", err)
}
r.followed(t) // and what they resolved equals a rebuild
if !t.Failed() {
t.Log("ok")
}
}
// Many check-ins zit refuses never hold more than the row cap, and a
// round's retries end at the budget however slow zit is.
func TestPendingRowsAndRetryWorkAreBounded(t *testing.T) {
countZit(t)
r, tip, blob := indexedTree(t)
oldCap, oldBudget := pendingRowCap, retryBudget
pendingRowCap, retryBudget = 5, 1500*time.Millisecond
t.Cleanup(func() { pendingRowCap, retryBudget = oldCap, oldBudget })
var flood [][]byte
for i := range 12 {
raw, _ := sealed(fmt.Sprintf("C flood\\s%d\nD 2026-11-14T01:00:%02d\nP %s\nS f%02d.txt %s\nU ana\n", i, i, tip, i, blob))
flood = append(flood, raw)
}
r.pushRaw(t, flood...)
if _, err := r.p.Sync(); err != nil {
t.Fatal(err)
}
r.later(time.Second) // the cap is applied as the next round starts
if _, err := r.p.Sync(); err != nil {
t.Fatal(err)
}
rows, err := r.ix.PendingChanges(1000, r.now.Add(1000*time.Hour))
if err != nil {
t.Fatal(err)
}
if len(rows) > pendingRowCap {
t.Errorf("FAIL: %d check-ins pending, over the cap of %d", len(rows), pendingRowCap)
}
// Every diff now takes a second; five rows are due.
log := filepath.Join(t.TempDir(), "calls")
t.Setenv("ZIT_CALL_LOG", log)
t.Setenv("ZIT_SLOW", "1")
r.later(pendingDelay(0))
start := time.Now()
if _, err := r.p.Sync(); err != nil {
t.Fatal(err)
}
took := time.Since(start)
t.Setenv("ZIT_SLOW", "")
raw, _ := os.ReadFile(log)
diffs := strings.Count(string(raw), "diff\n")
if diffs > 2 {
t.Errorf("FAIL: %d slow retries in a round with a %s budget", diffs, retryBudget)
}
// The budget, plus the round's own work and one killed call's wait.
if limit := retryBudget + 5*time.Second; took > limit {
t.Errorf("FAIL: the round took %s, past %s", took.Round(time.Millisecond), limit)
}
t.Logf("slow round: %d diff calls, %s", diffs, took.Round(time.Millisecond))
if !t.Failed() {
t.Log("ok")
}
}