package ingest
import (
"context"
"errors"
"fmt"
"os"
"os/exec"
"path/filepath"
"sync"
"testing"
"time"
"craftdesign.group/quarry/internal/work"
)
// withSlowCrosslink puts a zit on PATH whose crosslink — early in every
// rebuild — holds for d, so rebuilds overlap long enough to queue.
func withSlowCrosslink(t *testing.T, d time.Duration) {
t.Helper()
real, err := exec.LookPath("zit")
if err != nil {
t.Skip("zit not on PATH")
}
bin := t.TempDir()
script := fmt.Sprintf("#!/bin/sh\nif [ \"$1\" = crosslink ]; then sleep %g; fi\nexec %q \"$@\"\n", d.Seconds(), real)
if err := os.WriteFile(filepath.Join(bin, "zit"), []byte(script), 0o755); err != nil {
t.Fatal(err)
}
t.Setenv("PATH", bin+string(os.PathListSeparator)+os.Getenv("PATH"))
}
// spans records every rebuild's whole slot-holding span: how many ran at
// once at the peak, and how many ever started. started receives one
// value per start.
type spans struct {
mu sync.Mutex
now, peak, in int
started chan struct{}
}
func watchSpans(t *testing.T) *spans {
t.Helper()
s := &spans{started: make(chan struct{}, 64)}
old := rebuildSpan
rebuildSpan = func(delta int) {
s.mu.Lock()
defer s.mu.Unlock()
s.now += delta
s.peak = max(s.peak, s.now)
if delta > 0 {
s.in++
s.started <- struct{}{}
}
}
t.Cleanup(func() { rebuildSpan = old })
return s
}
func (s *spans) counts() (peak, in int) {
s.mu.Lock()
defer s.mu.Unlock()
return s.peak, s.in
}
// After downtime longer than the stream window every store on a node
// finds itself behind at once, and every one of them rebuilt at the
// same moment. Rebuilds across stores are bounded over their whole
// span, not just the step that happens to be slow; the rest queue.
func TestRebuildsAcrossStoresAreBounded(t *testing.T) {
withSlowCrosslink(t, time.Second)
s := watchSpans(t)
const stores = 4
var pipes []*Pipeline
for range stores {
p, _, _ := testPipeline(t)
pipes = append(pipes, p)
}
var wg sync.WaitGroup
errs := make([]error, stores)
for i, p := range pipes {
wg.Add(1)
go func() {
defer wg.Done()
_, errs[i] = p.Reindex()
}()
}
wg.Wait()
for i, err := range errs {
if err != nil {
t.Fatalf("store %d: %v", i, err)
}
}
peak, in := s.counts()
if in != stores {
t.Fatalf("saw %d rebuilds hold a slot, want %d", in, stores)
}
if peak > maxRebuilds {
t.Errorf("FAIL: %d rebuilds ran at once, want at most %d", peak, maxRebuilds)
} else {
t.Logf("ok: at most %d of %d rebuilds ran at once (limit %d)", peak, stores, maxRebuilds)
}
}
// A rebuild queued for a slot used to wait it out and then run in full
// before its store could stop, so after an outage shutdown took about
// half the fleet's rebuild time, and removing one store whose rebuild
// was queued blocked the same way. A stopped pipeline leaves the queue
// at once and never starts the rebuild.
func TestAStoppedPipelineLeavesTheRebuildQueue(t *testing.T) {
withSlowCrosslink(t, 3*time.Second)
s := watchSpans(t)
var pipes []*Pipeline
for range 4 {
p, _, _ := testPipeline(t)
pipes = append(pipes, p)
}
var running sync.WaitGroup
for _, p := range pipes[:maxRebuilds] {
running.Add(1)
go func() {
defer running.Done()
if _, err := p.Reindex(); err != nil {
t.Errorf("a running rebuild: %v", err)
}
}()
}
startBound := time.After(10 * time.Second)
for range maxRebuilds {
select {
case <-s.started:
case <-startBound:
t.Fatal("the first rebuilds never took their slots")
}
}
queued := pipes[maxRebuilds:]
returned := make(chan error, len(queued))
for _, p := range queued {
go func() {
_, err := p.Reindex()
returned <- err
}()
}
deadline := time.Now().Add(5 * time.Second)
for rebuildsQueued.Load() < int64(len(queued)) {
if time.Now().After(deadline) {
t.Fatalf("only %d rebuilds queued", rebuildsQueued.Load())
}
time.Sleep(10 * time.Millisecond)
}
for _, p := range queued {
p.Stop()
}
// Judged by order, not a clock: a queued rebuild that waited for a
// slot could only come back after a running one had freed it.
runningDone := make(chan struct{})
go func() {
running.Wait()
close(runningDone)
}()
for range queued {
select {
case err := <-returned:
if !errors.Is(err, ErrStopped) {
t.Errorf("FAIL: a stopped queued rebuild returned %v, want ErrStopped", err)
}
case <-runningDone:
t.Fatal("FAIL: a stopped pipeline waited in the rebuild queue until a slot came free")
}
}
t.Log("ok: stopped pipelines leave the rebuild queue at once")
<-runningDone
if _, in := s.counts(); in != maxRebuilds {
t.Errorf("FAIL: %d rebuilds started, want only the %d running before the stop", in, maxRebuilds)
} else {
t.Log("ok: no rebuild starts after its pipeline stopped")
}
}
// Five stores rebuilding after an outage used to fill the ingest lane:
// two held their lane slots rebuilding and a third held one waiting
// for a rebuild slot, the rest filled its queue, and every other
// store's ordinary round was shed. A round waiting for a rebuild slot
// now steps out of the lane while it waits.
func TestARoundQueuedForARebuildLeavesTheLaneToOthers(t *testing.T) {
plain, plainIx, plainRepo := testPipeline(t)
for range 2 { // the first round rebuilds; the second subscribes
if _, err := plain.Sync(); err != nil {
t.Fatal(err)
}
}
withSlowCrosslink(t, 3*time.Second)
s := watchSpans(t)
lane := work.NewLane("ingest", 3, 2) // production's shape
var rebuilding sync.WaitGroup
for range 5 {
p, _, _ := testPipeline(t) // no cursor yet: its round rebuilds
rebuilding.Add(1)
go func() {
defer rebuilding.Done()
err := lane.DoNamedAside(context.Background(), "ingest", p.code, func(aside func(func() error) error) error {
_, err := p.SyncAside(aside)
return err
})
if err != nil {
t.Errorf("a rebuilding round: %v", err)
}
}()
}
startBound := time.After(10 * time.Second)
for range maxRebuilds {
select {
case <-s.started:
case <-startBound:
t.Fatal("the first rebuilds never took their slots")
}
}
deadline := time.Now().Add(10 * time.Second)
for rebuildsQueued.Load() < 3 {
if time.Now().After(deadline) {
t.Fatalf("only %d rounds queued for a rebuild slot", rebuildsQueued.Load())
}
time.Sleep(10 * time.Millisecond)
}
before := checkins(t, plainIx)
commitFile(t, plainRepo, "c.txt", "gamma\n", "add c")
rebuilt := make(chan struct{})
go func() {
rebuilding.Wait()
close(rebuilt)
}()
ran := make(chan error, 1)
go func() {
ran <- lane.DoNamed(context.Background(), "ingest", plain.code, func() error {
_, err := plain.Sync()
return err
})
}()
select {
case err := <-ran:
if err != nil {
t.Fatalf("FAIL: a plain round while rebuilds queue: %v", err)
}
case <-rebuilt:
t.Fatal("FAIL: a plain round waited for the rebuilds to finish")
}
if got := checkins(t, plainIx); got != before+1 {
t.Errorf("FAIL: the plain round did not fold the new check-in: %d, want %d", got, before+1)
}
if st := lane.Stats(); st.Shed != 0 {
t.Errorf("FAIL: the lane shed %d units", st.Shed)
} else {
t.Log("ok: a plain round runs while five stores rebuild, and nothing is shed")
}
<-rebuilt
}