package ingest
import (
"encoding/json"
"os/exec"
"path/filepath"
"testing"
"time"
"craftdesign.group/quarry/internal/events"
"craftdesign.group/quarry/internal/index"
"craftdesign.group/quarry/internal/zitcli"
)
// storeOnStream is one repository's pipeline on a shared event plane.
func storeOnStream(t *testing.T, ev *events.Server, code string) (*Pipeline, string, *index.Index) {
t.Helper()
repo := t.TempDir()
run(t, repo, "init")
commitFile(t, repo, "a.txt", "alpha\n", "add a in "+code)
ix, err := index.Open(filepath.Join(t.TempDir(), code+".db"))
if err != nil {
t.Fatal(err)
}
t.Cleanup(func() { ix.Close() })
return NewPipeline(zitcli.New(repo), ev, ix, code, "n1"), repo, ix
}
func readMark(t *testing.T, p *Pipeline) windowMark {
t.Helper()
kv, err := p.cursorKV()
if err != nil {
t.Fatal(err)
}
entry, err := kv.Get(p.markKey())
if err != nil {
t.Fatal(err)
}
var m windowMark
if err := json.Unmarshal(entry.Value(), &m); err != nil {
t.Fatal(err)
}
return m
}
// quietAcrossARestart folds a quiet store and a busy one, lets the busy
// one's events age off the front of the shared stream while the quiet
// one sits still, and restarts the event plane. It returns the quiet
// store's pipeline as the restarted process makes it.
func quietAcrossARestart(t *testing.T, beforeAging, beforeRestart func(quiet *Pipeline)) (*Pipeline, uint64) {
t.Helper()
if _, err := exec.LookPath("zit"); err != nil {
t.Skip("zit not on PATH")
}
natsDir := filepath.Join(t.TempDir(), "nats")
ev, err := events.Start(natsDir)
if err != nil {
t.Fatal(err)
}
quiet, quietRepo, quietIx := storeOnStream(t, ev, "quiet")
busy, busyRepo, _ := storeOnStream(t, ev, "busy")
for _, p := range []*Pipeline{quiet, busy} {
for i := 0; i < 2; i++ { // bootstrap, then a drained fold that sets the mark
if _, err := p.Sync(); err != nil {
t.Fatal(err)
}
}
}
for i, msg := range []string{"busy one", "busy two", "busy three"} {
commitFile(t, busyRepo, "b.txt", msg+"\n", msg)
if _, err := busy.Sync(); err != nil {
t.Fatalf("busy round %d: %v", i, err)
}
}
if beforeAging != nil {
beforeAging(quiet)
}
shrinkWindow(t, busy, 300*time.Millisecond) // every event so far ages out
if beforeRestart != nil {
beforeRestart(quiet)
}
info, err := ev.JS().StreamInfo(events.StreamName)
if err != nil {
t.Fatal(err)
}
first := info.State.FirstSeq
ev.Close()
ev, err = events.Start(natsDir)
if err != nil {
t.Fatal(err)
}
t.Cleanup(ev.Close)
return NewPipeline(zitcli.New(quietRepo), ev, quietIx, "quiet", "n1"), first
}
// Every store shares one stream, and a quiet store's folded position
// stays where its last round left it. The stream dropping OTHER stores'
// events past that position loses nothing of the quiet store's, yet the
// window-gap rule read it as "behind the window" and rebuilt every quiet
// store in full at each restart more than an hour after its last event.
func TestAQuietStoreResumesAfterOtherStoresEventsAgeOut(t *testing.T) {
quiet, first := quietAcrossARestart(t, nil, nil)
if _, err := quiet.Sync(); err != nil {
t.Fatal(err)
}
if s := quiet.Stats(); s.WindowGaps != 0 {
t.Errorf("FAIL: a quiet store was rebuilt for events that were not its own: %+v", s)
} else {
t.Log("ok: the quiet store resumed without a rebuild")
}
if m := readMark(t, quiet); m.Seq+1 < first {
t.Errorf("FAIL: the quiet store's mark stayed behind the window: %d, stream starts at %d", m.Seq, first)
} else {
t.Log("ok: the quiet store's mark moved up to the window")
}
}
// Without a record of the store's last publish, nothing says whether its
// own events were among those dropped, so the rebuild stays: a wrong
// "nothing lost" is a silent hole in the index, a wrong rebuild is only
// slow.
func TestAStoreWithNoPublishRecordStillRebuilds(t *testing.T) {
quiet, _ := quietAcrossARestart(t, nil, func(quiet *Pipeline) {
kv, err := quiet.cursorKV()
if err != nil {
t.Fatal(err)
}
if err := kv.Delete(quiet.pubKey()); err != nil {
t.Fatal(err)
}
})
if _, err := quiet.Sync(); err != nil {
t.Fatal(err)
}
if s := quiet.Stats(); s.WindowGaps != 1 {
t.Errorf("FAIL: an undeterminable gap was not rebuilt: %+v", s)
} else {
t.Log("ok: an undeterminable gap still rebuilds")
}
kv, err := quiet.cursorKV()
if err != nil {
t.Fatal(err)
}
if _, err := kv.Get(quiet.pubKey()); err != nil {
t.Errorf("FAIL: the rebuild did not seed the publish record: %v", err)
}
}
// A store idle since before the publish record existed has none, and a
// store that is never rebuilt never gets one from a rebuild. Its first
// drained round inside the window seeds the record, so the restart after
// that one is judged exactly instead of rebuilding.
func TestADrainedRoundSeedsAMissingPublishRecord(t *testing.T) {
quiet, _ := quietAcrossARestart(t, func(quiet *Pipeline) {
kv, err := quiet.cursorKV()
if err != nil {
t.Fatal(err)
}
if err := kv.Delete(quiet.pubKey()); err != nil {
t.Fatal(err)
}
// The boot round of the release that ships the record, inside the
// window: a drained fold, not a rebuild.
if _, err := quiet.Sync(); err != nil {
t.Fatal(err)
}
if s := quiet.Stats(); s.WindowGaps != 0 {
t.Fatalf("a store inside the window was rebuilt: %+v", s)
}
for i := 0; i < 3; i++ { // the fleet carries on after the quiet store's round
if _, err := quiet.ev.JS().Publish("zit.evt.elsewhere.artifact", []byte(`{}`)); err != nil {
t.Fatal(err)
}
}
}, nil)
if _, err := quiet.Sync(); err != nil {
t.Fatal(err)
}
if s := quiet.Stats(); s.WindowGaps != 0 {
t.Errorf("FAIL: the restart after the seeding round rebuilt the store: %+v", s)
} else {
t.Log("ok: the seeded record let the next restart resume without a rebuild")
}
}
// A record from an earlier incarnation of the stream numbers a different
// sequence, so however low it reads it vouches for nothing on this one.
func TestAPublishRecordFromAnEarlierStreamStillRebuilds(t *testing.T) {
quiet, _ := quietAcrossARestart(t, nil, func(quiet *Pipeline) {
kv, err := quiet.cursorKV()
if err != nil {
t.Fatal(err)
}
// Sequence 0 sits at or below any folded position.
raw, err := json.Marshal(published{Seq: 0, Stream: time.Unix(1, 0).UTC()})
if err != nil {
t.Fatal(err)
}
if _, err := kv.Put(quiet.pubKey(), raw); err != nil {
t.Fatal(err)
}
})
if _, err := quiet.Sync(); err != nil {
t.Fatal(err)
}
if s := quiet.Stats(); s.WindowGaps != 1 {
t.Errorf("FAIL: an earlier stream's record vouched for this one: %+v", s)
} else {
t.Log("ok: an earlier stream's record does not skip the rebuild")
}
}