package ingest
import (
"errors"
"fmt"
"os"
"os/exec"
"path/filepath"
"runtime"
"slices"
"testing"
"time"
"github.com/nats-io/nats.go"
"craftdesign.group/quarry/internal/index"
)
// outageKV is the cursor bucket during a NATS hiccup: the real bucket
// underneath, with the calls a test names failing the way a dropped
// connection makes them fail.
type outageKV struct {
nats.KeyValue
get, del error
}
func (k outageKV) Get(key string) (nats.KeyValueEntry, error) {
if k.get != nil {
return nil, k.get
}
return k.KeyValue.Get(key)
}
func (k outageKV) Delete(key string, opts ...nats.DeleteOpt) error {
if k.del != nil {
return k.del
}
return k.KeyValue.Delete(key, opts...)
}
// duringOutage routes the pipeline's cursor reads through an outageKV.
func duringOutage(p *Pipeline, get, del error) {
p.cursors = func() (nats.KeyValue, error) {
kv, err := p.ev.Cursors()
if err != nil {
return nil, err
}
return outageKV{KeyValue: kv, get: get, del: del}, nil
}
}
func checkins(t *testing.T, ix *index.Index) int {
t.Helper()
tl, _, err := ix.Timeline(index.TimelineFilter{Limit: 100, Kind: "ci"})
if err != nil {
t.Fatal(err)
}
return len(tl)
}
// A brief NATS outage used to read as "there is no cursor", which
// started a full rebuild; the rebuild emptied the index and then failed
// to delete the cursor, so later rounds carried on from the surviving
// cursor into an empty index. A failed read is not an absent cursor: it
// must change nothing and go to the round's backoff like any failure.
func TestACursorReadThatFailsChangesNothingAndBacksOff(t *testing.T) {
p, ix, _ := testPipeline(t)
if _, err := p.Sync(); err != nil {
t.Fatal(err)
}
before := checkins(t, ix)
if before == 0 {
t.Fatal("no baseline")
}
duringOutage(p, nats.ErrTimeout, nats.ErrTimeout)
_, err := p.Sync()
if !errors.Is(err, nats.ErrTimeout) {
t.Errorf("FAIL: a failed cursor read returned %v, want the NATS error", err)
} else {
t.Log("ok: the NATS error reaches the caller")
}
if got := checkins(t, ix); got != before {
t.Errorf("FAIL: the index went from %d check-ins to %d during a NATS hiccup", before, got)
} else {
t.Log("ok: the index is untouched")
}
if s := p.Stats(); s.Failures != 1 {
t.Errorf("FAIL: the failed round was not counted towards the backoff: %+v", s)
} else {
t.Log("ok: the failure is counted towards the backoff")
}
}
// If the cursor cannot be deleted, the rebuild must not have started:
// an emptied index beside a live cursor is the one state later rounds
// cannot recover from, because they resume from the cursor and never
// republish what the index lost.
func TestARebuildWhoseCursorDeleteFailsLeavesTheIndexAlone(t *testing.T) {
p, ix, _ := testPipeline(t)
if _, err := p.Sync(); err != nil {
t.Fatal(err)
}
before := checkins(t, ix)
duringOutage(p, nil, nats.ErrTimeout)
if _, err := p.Reindex(); !errors.Is(err, nats.ErrTimeout) {
t.Errorf("FAIL: a rebuild whose cursor delete failed returned %v", err)
} else {
t.Log("ok: the delete failure is returned")
}
if got := checkins(t, ix); got != before {
t.Errorf("FAIL: the index went from %d check-ins to %d although the cursor survived", before, got)
} else {
t.Log("ok: the index is untouched while the cursor survives")
}
}
// withFailingCrosslink puts a zit on PATH that fails `crosslink` and
// runs every other command with the real binary, so the rebuild fails
// part-way with its census reader already running.
func withFailingCrosslink(t *testing.T) {
t.Helper()
real, err := exec.LookPath("zit")
if err != nil {
t.Skip("zit not on PATH")
}
dir := t.TempDir()
script := fmt.Sprintf("#!/bin/sh\nif [ \"$1\" = crosslink ]; then echo 'crosslink: injected failure' >&2; exit 1; fi\nexec %q \"$@\"\n", real)
if err := os.WriteFile(filepath.Join(dir, "zit"), []byte(script), 0o755); err != nil {
t.Fatal(err)
}
t.Setenv("PATH", dir+string(os.PathListSeparator)+os.Getenv("PATH"))
}
// A rebuild that failed after starting its census reader left the
// reader blocked on its channel, with its `zit frames` child still
// running, and because round returned straight from the rebuild the
// failure never reached the stall accounting: it repeated on every
// two-second tick and leaked again each time.
func TestAFailedRebuildReapsItsCensusReaderAndBacksOff(t *testing.T) {
old := reindexBatch
reindexBatch = 1 // a handful of tags then overfills the channel
t.Cleanup(func() { reindexBatch = old })
p, _, repo := testPipeline(t)
for i := 0; i < 24; i++ {
run(t, repo, "tag", "add", fmt.Sprintf("t%02d", i), "trunk")
}
withFailingCrosslink(t)
base := runtime.NumGoroutine()
for i := 1; i <= stallAfter; i++ {
if _, err := p.Sync(); err == nil {
t.Fatal("a rebuild with a failing crosslink reported success")
}
if s := p.Stats(); s.Failures != i {
t.Errorf("FAIL: after %d failed rebuilds the backoff counted %d", i, s.Failures)
}
}
if s := p.Stats(); !s.Stalled {
t.Errorf("FAIL: %d failed rebuilds did not stall the pipeline: %+v", stallAfter, s)
} else {
t.Log("ok: failed rebuilds are counted and back off")
}
deadline := time.Now().Add(3 * time.Second)
for runtime.NumGoroutine() > base && time.Now().Before(deadline) {
time.Sleep(20 * time.Millisecond)
}
if n := runtime.NumGoroutine(); n > base {
buf := make([]byte, 1<<20)
t.Errorf("FAIL: %d goroutines outlived the failed rebuilds:\n%s", n-base, buf[:runtime.Stack(buf, true)])
} else {
t.Log("ok: no census reader outlives a failed rebuild")
}
}
// derivedSchema lists the index's triggers and named indexes: what a
// rebuild suspends for its fold and must put back.
func derivedSchema(t *testing.T, ix *index.Index) []string {
t.Helper()
rows, err := ix.RawDB().Query(`SELECT type || ' ' || name FROM sqlite_master
WHERE type IN ('trigger', 'index') AND name NOT LIKE 'sqlite_autoindex_%' ORDER BY 1`)
if err != nil {
t.Fatal(err)
}
defer rows.Close()
var out []string
for rows.Next() {
var s string
if err := rows.Scan(&s); err != nil {
t.Fatal(err)
}
out = append(out, s)
}
if err := rows.Err(); err != nil {
t.Fatal(err)
}
return out
}
// A rebuild drops the search triggers and the bulk indexes for its
// fold. One that failed after that returned without restoring them, and
// the store had no search and no indexes until a later rebuild
// succeeded — which the stall backoff can put many minutes off.
func TestAFailedRebuildRestoresSearchTriggersAndIndexes(t *testing.T) {
p, ix, _ := testPipeline(t)
if _, err := p.Sync(); err != nil {
t.Fatal(err)
}
want := derivedSchema(t, ix)
if !slices.Contains(want, "trigger search_doc_ai") {
t.Fatalf("no search trigger to begin with: %v", want)
}
withFailingCrosslink(t)
if _, err := p.Reindex(); err == nil {
t.Fatal("a rebuild with a failing crosslink reported success")
}
if got := derivedSchema(t, ix); !slices.Equal(got, want) {
t.Errorf("FAIL: after a failed rebuild the schema is\n%v\nwant\n%v", got, want)
} else {
t.Log("ok: a failed rebuild puts back every trigger and index it dropped")
}
}