package ingest
import (
"path/filepath"
"testing"
"time"
"github.com/nats-io/nats.go"
"craftdesign.group/quarry/internal/events"
"craftdesign.group/quarry/internal/index"
)
// shrinkWindow cuts the event stream's retention to d and waits until
// everything older has aged out, which is what an hour of downtime does
// to a node on the real one-hour window.
func shrinkWindow(t *testing.T, p *Pipeline, d time.Duration) {
t.Helper()
js := p.ev.JS()
info, err := js.StreamInfo(events.StreamName)
if err != nil {
t.Fatal(err)
}
cfg := info.Config
cfg.MaxAge = d
cfg.Duplicates = d // may not exceed MaxAge
if _, err := js.UpdateStream(&cfg); err != nil {
t.Fatal(err)
}
deadline := time.Now().Add(5 * time.Second)
for {
info, err := js.StreamInfo(events.StreamName)
if err != nil {
t.Fatal(err)
}
if info.State.Msgs == 0 {
return
}
if time.Now().After(deadline) {
t.Fatalf("the stream still holds %d messages after its window passed", info.State.Msgs)
}
time.Sleep(50 * time.Millisecond)
}
}
func hasComment(t *testing.T, ix *index.Index, comment string) bool {
t.Helper()
tl, _, err := ix.Timeline(index.TimelineFilter{Limit: 100, Kind: "ci"})
if err != nil {
t.Fatal(err)
}
for _, e := range tl {
if e.Comment == comment {
return true
}
}
return false
}
// A node away for longer than the stream's window resumes from a
// position the stream no longer holds. The fetch after the gap simply
// returns newer messages (events/gap_test.go measured it), so a fold
// that only watches NumPending carries on with a hole. The cursor is
// shared, so nothing republishes what the other node already published:
// the hole is permanent unless the gap is noticed and the index rebuilt.
func TestANodeThatFallsBehindTheWindowRebuilds(t *testing.T) {
away, awayIx, repo := testPipeline(t)
for i := 0; i < 2; i++ { // bootstrap, then a drained fold that creates the consumer
if _, err := away.Sync(); err != nil {
t.Fatal(err)
}
}
otherIx, err := index.Open(filepath.Join(t.TempDir(), "other.db"))
if err != nil {
t.Fatal(err)
}
t.Cleanup(func() { otherIx.Close() })
other := NewPipeline(away.cli, away.ev, otherIx, "test", "n2")
if _, err := other.Sync(); err != nil {
t.Fatal(err)
}
const late = "published while n1 was away"
commitFile(t, repo, "late.txt", "late\n", late)
if _, err := other.Sync(); err != nil {
t.Fatal(err)
}
if !hasComment(t, otherIx, late) {
t.Fatal("the publishing node never indexed the check-in")
}
shrinkWindow(t, away, 300*time.Millisecond)
if _, err := away.Sync(); err != nil {
t.Fatalf("the round after the gap failed: %v", err)
}
if s := away.Stats(); s.WindowGaps != 1 {
t.Errorf("FAIL: the gap was not counted: %+v", s)
} else {
t.Log("ok: the gap behind the stream's window is counted")
}
if !hasComment(t, awayIx, late) {
t.Error("FAIL: the node behind the window never learned the check-in it missed")
} else {
t.Log("ok: the node behind the window rebuilt and holds what it missed")
}
}
// The stream is shared by every repository, and a durable consumer's
// ack floor only moves when ITS messages are acked. A repository with
// nothing new for an hour therefore sits behind the stream's first
// sequence while the rest of the fleet publishes, although it has missed
// nothing. Reading that as a gap would rebuild every idle store every
// hour, and again on every restart.
func TestAnIdleRepositoryOnABusyStreamIsNotRebuilt(t *testing.T) {
p, ix, _ := testPipeline(t)
for i := 0; i < 2; i++ {
if _, err := p.Sync(); err != nil {
t.Fatal(err)
}
}
for i := 0; i < 5; i++ {
if _, err := p.ev.JS().Publish("zit.evt.elsewhere.artifact", []byte(`{}`)); err != nil {
t.Fatal(err)
}
}
if _, err := p.Sync(); err != nil { // a live round while the fleet is busy
t.Fatal(err)
}
shrinkWindow(t, p, 300*time.Millisecond)
if _, err := p.Sync(); err != nil {
t.Fatal(err)
}
if s := p.Stats(); s.WindowGaps != 0 {
t.Errorf("FAIL: an idle repository was read as behind the window: %+v", s)
} else {
t.Log("ok: a live idle repository is not rebuilt")
}
restarted := NewPipeline(p.cli, p.ev, ix, "test", "n1")
if _, err := restarted.Sync(); err != nil {
t.Fatal(err)
}
if s := restarted.Stats(); s.WindowGaps != 0 {
t.Errorf("FAIL: a restarted idle repository was read as behind the window: %+v", s)
} else {
t.Log("ok: a restarted idle repository is not rebuilt")
}
}
// The first run after the window check shipped finds no mark anywhere,
// and on a busy shared stream every idle store's ack floor is behind the
// stream's first sequence. Treating that as a gap would rebuild the whole
// fleet at once on deploy. A missing mark is seeded from the stream as it
// stands; only a node that falls behind a mark it HAS is rebuilt.
func TestAMissingMarkIsSeededAndOnlyAnExistingMarkCanReportAGap(t *testing.T) {
p, ix, repo := testPipeline(t)
for i := 0; i < 2; i++ {
if _, err := p.Sync(); err != nil {
t.Fatal(err)
}
}
kv, err := p.ev.Cursors()
if err != nil {
t.Fatal(err)
}
// The state a node was in before this code shipped: an old floor on a
// busy stream, and no mark.
if err := kv.Delete(p.markKey()); err != nil {
t.Fatal(err)
}
for i := 0; i < 5; i++ {
if _, err := p.ev.JS().Publish("zit.evt.elsewhere.artifact", []byte(`{}`)); err != nil {
t.Fatal(err)
}
}
shrinkWindow(t, p, 300*time.Millisecond)
upgraded := NewPipeline(p.cli, p.ev, ix, "test", "n1")
if _, err := upgraded.Sync(); err != nil {
t.Fatal(err)
}
if s := upgraded.Stats(); s.WindowGaps != 0 {
t.Errorf("FAIL: a missing mark was read as a gap: %+v", s)
} else {
t.Log("ok: a missing mark does not rebuild")
}
if _, err := kv.Get(upgraded.markKey()); err != nil {
t.Errorf("FAIL: the missing mark was not seeded: %v", err)
} else {
t.Log("ok: the missing mark is seeded")
}
// With the mark in place, falling behind it is a real gap.
otherIx, err := index.Open(filepath.Join(t.TempDir(), "other.db"))
if err != nil {
t.Fatal(err)
}
t.Cleanup(func() { otherIx.Close() })
other := NewPipeline(p.cli, p.ev, otherIx, "test", "n2")
if _, err := other.Sync(); err != nil {
t.Fatal(err)
}
const late = "published behind the seeded mark"
commitFile(t, repo, "late.txt", "late\n", late)
if _, err := other.Sync(); err != nil {
t.Fatal(err)
}
deadline := time.Now().Add(5 * time.Second)
for {
info, err := p.ev.JS().StreamInfo(events.StreamName)
if err != nil {
t.Fatal(err)
}
if info.State.Msgs == 0 {
break
}
if time.Now().After(deadline) {
t.Fatal("the late check-in never aged out")
}
time.Sleep(50 * time.Millisecond)
}
if _, err := upgraded.Sync(); err != nil {
t.Fatal(err)
}
if s := upgraded.Stats(); s.WindowGaps != 1 || !hasComment(t, ix, late) {
t.Errorf("FAIL: falling behind an existing mark did not rebuild: %+v", s)
} else {
t.Log("ok: falling behind an existing mark rebuilds")
}
}
// recreateStream deletes the event stream and the cursor bucket and
// makes them again, as events.Start does when it migrates an older
// stream on a node that shares this NATS.
func recreateStream(t *testing.T, p *Pipeline) {
t.Helper()
js := p.ev.JS()
info, err := js.StreamInfo(events.StreamName)
if err != nil {
t.Fatal(err)
}
if err := js.DeleteStream(events.StreamName); err != nil {
t.Fatal(err)
}
if err := js.DeleteKeyValue(events.CursorBucket); err != nil {
t.Fatal(err)
}
if _, err := js.AddStream(&info.Config); err != nil {
t.Fatal(err)
}
if _, err := js.CreateKeyValue(&nats.KeyValueConfig{Bucket: events.CursorBucket}); err != nil {
t.Fatal(err)
}
}
// Recreating the stream deletes its consumers. A running node kept its
// handle to the deleted one, and every fold failed on it until the
// process restarted.
func TestARecreatedStreamIsFoldedWithoutARestart(t *testing.T) {
p, ix, repo := testPipeline(t)
for range 2 { // the first round rebuilds; the second subscribes
if _, err := p.Sync(); err != nil {
t.Fatal(err)
}
}
if p.sub == nil {
t.Fatal("no consumer after a folding round")
}
recreateStream(t, p)
for i := 1; i <= 2; i++ {
if _, err := p.Sync(); err != nil {
t.Fatalf("FAIL: round %d after the stream was recreated: %v", i, err)
}
}
t.Log("ok: rounds after the recreation succeed")
before := checkins(t, ix)
commitFile(t, repo, "c.txt", "gamma\n", "add c")
if _, err := p.Sync(); err != nil {
t.Fatalf("FAIL: folding a new check-in from the recreated stream: %v", err)
}
if got := checkins(t, ix); got != before+1 {
t.Errorf("FAIL: the new check-in was not folded: %d check-ins, want %d", got, before+1)
} else {
t.Log("ok: a new check-in is folded from the recreated stream")
}
}