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(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 ""
}