package ingest
import (
"fmt"
"os"
"os/exec"
"path/filepath"
"strconv"
"strings"
"sync"
"testing"
)
// 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; the rest queue.
func TestRebuildsAcrossStoresAreBounded(t *testing.T) {
real, err := exec.LookPath("zit")
if err != nil {
t.Skip("zit not on PATH")
}
// A zit whose crosslink — the start of every rebuild — notes how
// many are running, and holds long enough for the rest to pile in.
bin, running := t.TempDir(), t.TempDir()
seen := filepath.Join(t.TempDir(), "seen")
script := fmt.Sprintf("#!/bin/sh\nif [ \"$1\" = crosslink ]; then : > %q/$$; ls %q | wc -l >> %q; sleep 1; rm -f %q/$$; fi\nexec %q \"$@\"\n",
running, running, seen, running, 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"))
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)
}
}
raw, err := os.ReadFile(seen)
if err != nil {
t.Fatal(err)
}
counts := strings.Fields(string(raw))
if len(counts) != stores {
t.Fatalf("saw %d rebuilds start, want %d", len(counts), stores)
}
peak := 0
for _, c := range counts {
n, err := strconv.Atoi(c)
if err != nil {
t.Fatal(err)
}
peak = max(peak, n)
}
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)
}
}