package ingest
import (
"errors"
"path/filepath"
"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)
}
}