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()
if err := pollRound(lane, p); 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
}
// The reviewer's deadlock, in production's shape: two write-op
// rebuilds fill both rebuild slots; three stores' poll rounds queue for
// a slot; an operator reindex of each of those three stores takes a
// lane slot and waits for the store's lock, held by its queued round.
// When the write-op rebuilds finish, a queued round that had to get
// back into the lane never could, and ingest stopped fleet-wide.
func TestQueuedRebuildsAndOperatorReindexesAllFinish(t *testing.T) {
withSlowCrosslink(t, 2*time.Second)
s := watchSpans(t)
lane := work.NewLane("ingest", 3, 2)
// Every store first: making one takes longer under -race than a
// write-op rebuild, which could finish before anything queued.
var writers, queued []*Pipeline
for range maxRebuilds {
p, _, _ := testPipeline(t)
writers = append(writers, p)
}
for range 3 {
p, _, _ := testPipeline(t)
queued = append(queued, p)
}
finished := make(chan string, 16)
for i, p := range writers {
go func() {
_, err := p.Reindex() // a write op's rebuild: no lane
finished <- fmt.Sprintf("write-op rebuild %d: %v", i, err)
}()
}
for range maxRebuilds {
select {
case <-s.started:
case <-time.After(10 * time.Second):
t.Fatal("the write-op rebuilds never took their slots")
}
}
for _, p := range queued {
go func() {
err := pollRound(lane, p)
finished <- fmt.Sprintf("poll round %s: %v", p.code, err)
}()
}
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)
}
// Each parked store also gets an operator reindex and a second poll
// round, which used to wait for the store's lock holding a lane slot.
for _, p := range queued {
go func() {
err := operatorReindex(lane, p)
finished <- fmt.Sprintf("operator reindex %s: %v", p.code, err)
}()
go func() {
err := pollRound(lane, p)
finished <- fmt.Sprintf("second poll round %s: %v", p.code, err)
}()
}
units := maxRebuilds + 9
guard := time.After(20 * time.Second)
for i := range units {
select {
case msg := <-finished:
t.Log(msg)
case <-guard:
t.Fatalf("FAIL: %d of %d units finished in 20s: the lane is deadlocked", i, units)
}
}
t.Log("ok: every queued round and operator reindex finished")
}
// pollRound is the registrar's poll round: admitted through the lane.
func pollRound(lane *work.Lane, p *Pipeline) error {
_, err := p.SyncIn(func(body func(leave func()) error) error {
return lane.DoNamedLeavable(context.Background(), "ingest", p.code, body)
})
return err
}
// operatorReindex is the registrar's repo.reindex: a rebuild, which runs
// on a rebuild slot and never in the lane.
func operatorReindex(_ *work.Lane, p *Pipeline) error {
_, err := p.Reindex()
return err
}