Skip to content
quarry Artifact [bcf1bccf55]
Login

Artifact [bcf1bccf55]

Check-ins Using Download Hex

Artifact bcf1bccf555f58f501f35fa805258840e3a663e5a54193361018b5414f4fea6f:

package ingest

import (
	"craftdesign.group/quarry/internal/artifact"
	"errors"
	"os/exec"
	"path/filepath"
	"strings"
	"testing"
	"time"

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

// THE invariant (ticket 0c51d50626): a failed fold must never advance
// the publisher cursor. The stream is a bounded live tail; messages a
// failing fold leaves unacked age out of it, and once the cursor has
// moved past them nothing ever republishes those artifacts — the index
// is silently, permanently short, and every surface renders the hole as
// truth. The 2026-08-19 schema incident emptied ticket_change on every
// store exactly this way.
//
// With the cursor held back, a fold failure costs only repetition: the
// next round re-reads the same tail from the store and republishes,
// which idempotent folds absorb.
func TestFailedFoldDoesNotAdvanceCursor(t *testing.T) {
	p, ix, repo := testPipeline(t)
	if _, err := p.Sync(); err != nil {
		t.Fatal(err)
	}
	kv, err := p.ev.Cursors()
	if err != nil {
		t.Fatal(err)
	}
	before := loadCursor(kv, "test")

	commitFile(t, repo, "c.txt", "gamma\n", "add c")
	ix.Close() // every fold now fails, the way a schema bug made them fail

	if _, err := p.Sync(); err == nil {
		t.Fatal("a round with a failing fold reported success")
	}
	if after := loadCursor(kv, "test"); after != before {
		t.Fatalf("the cursor advanced past artifacts the fold never indexed:\n before %+v\n after  %+v", before, after)
	}
}

// After repeated fold failures the pipeline must stop grinding and say
// so: each retry republishes the store's tail, and a wedged store
// retried every two seconds is an expensive way to stay broken. The
// pipeline backs off with an alarm, KEEPS backing off, and heals by
// itself on the attempt after the cause is fixed — the store was always
// correct, so no operator ceremony is owed beyond fixing the cause.
func TestStallBacksOffAlarmsAndRecovers(t *testing.T) {
	// Compressed BEFORE the first Sync: the durable consumer is created
	// on the first fold with whatever ackWait holds, and an existing
	// consumer keeps its config.
	old, oldAck := stallBase, ackWait
	stallBase, ackWait = time.Millisecond, 200*time.Millisecond
	t.Cleanup(func() { stallBase, ackWait = old, oldAck })

	p, ix, repo := testPipeline(t)
	if _, err := p.Sync(); err != nil {
		t.Fatal(err)
	}
	commitFile(t, repo, "c.txt", "gamma\n", "add c")
	ix.Close()

	for i := 0; i < stallAfter; i++ {
		_, err := p.Sync()
		if err == nil {
			t.Fatal("failing round reported success")
		}
		if errors.Is(err, ErrStalled) {
			t.Fatalf("stalled after %d failures; the threshold is %d", i+1, stallAfter)
		}
	}
	s := p.Stats()
	if !s.Stalled || s.RetryAt.IsZero() {
		t.Fatalf("no stall after %d consecutive failures: %+v", stallAfter, s)
	}

	// While stalled, Sync refuses with the sentinel rather than running
	// a round — the poller can tell "backing off" from a fresh failure.
	if _, err := p.Sync(); !errors.Is(err, ErrStalled) {
		t.Fatalf("stalled Sync returned %v, want ErrStalled", err)
	}

	// The cooldown elapses, the cause is fixed, and the next attempt
	// heals: the held-back cursor means the unfolded tail is re-read
	// from the store and this time lands.
	fixed, err := index.Open(filepath.Join(t.TempDir(), "fresh.db"))
	if err != nil {
		t.Fatal(err)
	}
	t.Cleanup(func() { fixed.Close() })
	p.ix = fixed
	time.Sleep(20 * time.Millisecond)

	if _, err := p.Sync(); err != nil {
		t.Fatalf("recovery round failed: %v", err)
	}
	if s := p.Stats(); s.Stalled || s.Failures != 0 {
		t.Fatalf("stall did not clear on success: %+v", s)
	}
	tl, _, err := fixed.Timeline(index.TimelineFilter{Limit: 10})
	if err != nil {
		t.Fatal(err)
	}
	var found bool
	for _, e := range tl {
		if e.Comment == "add c" {
			found = true
		}
	}
	if !found {
		t.Fatalf("the artifact the failing folds never indexed did not arrive after recovery: %+v", tl)
	}
}

// A dangling parent only counts as loss if the store holds that parent
// AS A CHECK-IN. tk's history carries a P card naming a file blob —
// tkMacOSXEvent.c, CVS-conversion residue from 2011 — and fossil's own
// database records the same edge and draws the same orphan. Probing
// with "does the store hold this hash" called that damage, so tk
// rebuilt 69,182 artifacts, failed to conjure a check-in from a C
// file, and reported itself permanently broken. The probe must parse
// what it finds: only a manifest is repairable loss.
func TestBlobParentIsNotCountedAsLoss(t *testing.T) {
	p, ix, repo := testPipeline(t)
	if _, err := p.Sync(); err != nil {
		t.Fatal(err)
	}
	// A real blob hash from the store: a.txt's content artifact.
	out, err := exec.Command("zit", "artifact", "-R", repo, "--raw",
		blobHash(t, repo, "a.txt")).Output()
	if err != nil || len(out) == 0 {
		t.Fatalf("store blob unreadable: %v", err)
	}

	// Fold a check-in whose parent is that blob — tk's exact shape.
	if err := ix.FoldNow([]index.Parsed{{Hash: strings.Repeat("d1", 32),
		Art: &artifact.Artifact{Type: artifact.Manifest,
			Date: "2026-01-01T00:00:00", User: "dan", Comment: "child of a blob",
			Parents: []string{blobHash(t, repo, "a.txt")}}}}); err != nil {
		t.Fatal(err)
	}
	lost, err := p.lostArtifacts()
	if err != nil {
		t.Fatal(err)
	}
	if lost != 0 {
		t.Fatalf("a blob parent counted as %d lost artifacts; no rebuild can make it a check-in", lost)
	}
}

// blobHash resolves the content hash of a committed file at trunk.
func blobHash(t *testing.T, repo, path string) string {
	t.Helper()
	out, err := exec.Command("zit", "ls", "trunk", "--long", "-R", repo).Output()
	if err != nil {
		t.Fatal(err)
	}
	for _, line := range strings.Split(string(out), "\n") {
		if strings.Contains(line, path) {
			for _, f := range strings.Fields(line) {
				if len(f) >= 40 && !strings.Contains(f, ".") {
					return f
				}
			}
		}
	}
	t.Fatalf("no blob hash for %s in:\n%s", path, out)
	return ""
}