package ingest
import (
"errors"
"fmt"
"os"
"os/exec"
"path/filepath"
"sync"
"testing"
"time"
)
// 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()
}
// Well inside the running rebuilds' three seconds: a queued rebuild
// that waited for a slot cannot come back in time.
bound := time.After(time.Second)
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 <-bound:
t.Fatal("FAIL: a stopped pipeline is still waiting in the rebuild queue")
}
}
t.Log("ok: stopped pipelines leave the rebuild queue at once")
running.Wait()
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")
}
}