sift/internal/batch/runner_test.go

package batch

import (
	"context"
	"errors"
	"fmt"
	"strings"
	"sync"
	"testing"
	"time"

	"git.wntrmute.dev/kyle/sift/internal/auth"
	"git.wntrmute.dev/kyle/sift/internal/dcache"
)

// Targets run one at a time, in order, and each fetch is given the submitting caller.
func TestSerialOrderAndCaller(t *testing.T) {
	var mu sync.Mutex
	var calls, names []string
	inflight, peak := 0, 0
	fetch := func(_ context.Context, c *auth.Caller, tgt Target) (Result, error) {
		mu.Lock()
		inflight++
		if inflight > peak {
			peak = inflight
		}
		calls = append(calls, tgt.URL)
		names = append(names, c.Name)
		mu.Unlock()
		time.Sleep(time.Millisecond)
		mu.Lock()
		inflight--
		mu.Unlock()
		return Result{Content: "ok " + tgt.URL}, nil
	}
	r := newRunner(t, fetch, Config{})
	id, reused, err := r.Submit(context.Background(), alice, dcache.ClassFleet, urls("a", "b", "c"))
	if err != nil || reused {
		t.Fatalf("submit err=%v reused=%v", err, reused)
	}
	waitStatus(t, r, id, "done")
	mu.Lock()
	defer mu.Unlock()
	if strings.Join(calls, " ") != "https://a.example/ https://b.example/ https://c.example/" {
		t.Fatalf("order %v", calls)
	}
	if peak != 1 {
		t.Fatalf("peak inflight %d, want 1 (serial)", peak)
	}
	for _, n := range names {
		if n != "alice" {
			t.Fatalf("fetch saw caller %q, want alice", n)
		}
	}
	j, _, _ := r.Status(id)
	if j.Caller != "alice" || j.TargetsTotal != 3 || j.TargetsDone != 3 || j.CreatedAt == 0 {
		t.Fatalf("job %+v", j)
	}
}

// With one worker, two live jobs take turns target by target.
func TestRoundRobin(t *testing.T) {
	first := make(chan struct{})
	release := make(chan struct{})
	var once sync.Once
	var mu sync.Mutex
	var calls []string
	fetch := func(_ context.Context, _ *auth.Caller, tgt Target) (Result, error) {
		mu.Lock()
		calls = append(calls, strings.TrimSuffix(strings.TrimPrefix(tgt.URL, "https://"), ".example/"))
		mu.Unlock()
		once.Do(func() { close(first); <-release })
		return Result{Content: "ok"}, nil
	}
	r := newRunner(t, fetch, Config{Workers: 1})
	a, _, err := r.Submit(context.Background(), alice, dcache.ClassFleet, urls("a1", "a2", "a3"))
	if err != nil {
		t.Fatal(err)
	}
	<-first // a1 is in flight
	b, _, err := r.Submit(context.Background(), bob, dcache.ClassFleet, urls("b1", "b2", "b3"))
	if err != nil {
		t.Fatal(err)
	}
	close(release)
	waitStatus(t, r, a, "done")
	waitStatus(t, r, b, "done")
	mu.Lock()
	defer mu.Unlock()
	if got := strings.Join(calls, " "); got != "a1 b1 a2 b2 a3 b3" {
		t.Fatalf("order %q, want round-robin a1 b1 a2 b2 a3 b3", got)
	}
}

func TestFailureIsolation(t *testing.T) {
	fetch := func(_ context.Context, _ *auth.Caller, tgt Target) (Result, error) {
		if tgt.URL == "https://b.example/" {
			return Result{}, &TargetError{Msg: "upstream 500", Hint: "browser"}
		}
		return Result{Content: "ok " + tgt.URL}, nil
	}
	r := newRunner(t, fetch, Config{})
	id, _, err := r.Submit(context.Background(), alice, dcache.ClassFleet, urls("a", "b", "c"))
	if err != nil {
		t.Fatal(err)
	}
	waitStatus(t, r, id, "partial")
	j, ts, ok := r.Status(id)
	if !ok || j.TargetsDone != 3 {
		t.Fatalf("job %+v ok=%v", j, ok)
	}
	if ts[0].Status != "done" || ts[1].Status != "failed" || ts[2].Status != "done" {
		t.Fatalf("statuses %+v", ts)
	}
	if ts[1].Error != "upstream 500" || ts[1].Hint != "browser" {
		t.Fatalf("failed target error=%q hint=%q", ts[1].Error, ts[1].Hint)
	}
	if ts[0].Result == nil || ts[0].Result.Content != "ok https://a.example/" {
		t.Fatalf("result %+v", ts[0].Result)
	}
}

func TestAllFailedIsFailed(t *testing.T) {
	fetch := func(context.Context, *auth.Caller, Target) (Result, error) { return Result{}, errors.New("nope") }
	r := newRunner(t, fetch, Config{})
	id, _, err := r.Submit(context.Background(), alice, dcache.ClassFleet, urls("a", "b"))
	if err != nil {
		t.Fatal(err)
	}
	waitStatus(t, r, id, "failed")
	_, ts, _ := r.Status(id)
	if ts[0].Error != "nope" || ts[0].Hint != "" {
		t.Fatalf("plain error stored as error=%q hint=%q", ts[0].Error, ts[0].Hint)
	}
}

func TestIdempotentResubmit(t *testing.T) {
	var mu sync.Mutex
	n := 0
	fetch := func(context.Context, *auth.Caller, Target) (Result, error) {
		mu.Lock()
		n++
		mu.Unlock()
		return Result{Content: "ok"}, nil
	}
	r := newRunner(t, fetch, Config{})
	id1, reused1, err := r.Submit(context.Background(), alice, dcache.ClassFleet, urls("a", "b"))
	if err != nil || reused1 {
		t.Fatalf("first submit err=%v reused=%v", err, reused1)
	}
	id2, reused2, err := r.Submit(context.Background(), alice, dcache.ClassFleet, urls("a", "b"))
	if err != nil || !reused2 || id2 != id1 {
		t.Fatalf("resubmit id=%q reused=%v err=%v", id2, reused2, err)
	}
	waitStatus(t, r, id1, "done")
	// Still reused once done (within the lease), and never re-fetched.
	if id3, reused3, err := r.Submit(context.Background(), alice, dcache.ClassFleet, urls("a", "b")); err != nil || !reused3 || id3 != id1 {
		t.Fatalf("resubmit after done id=%q reused=%v err=%v", id3, reused3, err)
	}
	// Another caller with the same list gets its own job.
	if id4, reused4, err := r.Submit(context.Background(), bob, dcache.ClassFleet, urls("a", "b")); err != nil || reused4 || id4 == id1 {
		t.Fatalf("other caller id=%q reused=%v err=%v", id4, reused4, err)
	}
	time.Sleep(50 * time.Millisecond)
	mu.Lock()
	defer mu.Unlock()
	if n != 4 {
		t.Fatalf("fetched %d times, want 4 (2 for alice's one job, 2 for bob's)", n)
	}
}

// A cancelled job is not handed back to a resubmit: the same list starts a new job.
func TestResubmitAfterCancelIsNew(t *testing.T) {
	release := make(chan struct{})
	fetch := func(context.Context, *auth.Caller, Target) (Result, error) {
		<-release
		return Result{Content: "ok"}, nil
	}
	r := newRunner(t, fetch, Config{})
	id1, _, err := r.Submit(context.Background(), alice, dcache.ClassFleet, urls("a", "b"))
	if err != nil {
		t.Fatal(err)
	}
	waitStatus(t, r, id1, "running")
	if !r.Cancel(id1) {
		t.Fatal("cancel returned false")
	}
	close(release)
	waitStatus(t, r, id1, "cancelled")
	id2, reused, err := r.Submit(context.Background(), alice, dcache.ClassFleet, urls("a", "b"))
	if err != nil || reused || id2 == id1 {
		t.Fatalf("resubmit after cancel id=%q reused=%v err=%v", id2, reused, err)
	}
}

// Submit returns with the job stored "queued"; it becomes "running" when a worker takes it.
func TestQueuedThenRunning(t *testing.T) {
	release := make(chan struct{})
	fetch := func(context.Context, *auth.Caller, Target) (Result, error) {
		<-release
		return Result{Content: "ok"}, nil
	}
	r := newRunner(t, fetch, Config{Workers: 1})
	defer close(release)
	a, _, _ := r.Submit(context.Background(), alice, dcache.ClassFleet, urls("a"))
	waitStatus(t, r, a, "running")
	b, _, err := r.Submit(context.Background(), bob, dcache.ClassFleet, urls("b"))
	if err != nil {
		t.Fatal(err)
	}
	// The only worker is busy with alice's target, so bob's job waits in the queue.
	time.Sleep(20 * time.Millisecond)
	if j, _, _ := r.Status(b); j.Status != "queued" {
		t.Fatalf("bob's job is %q while the only worker is busy, want queued", j.Status)
	}
}

func TestCancel(t *testing.T) {
	started := make(chan struct{}, 1)
	release := make(chan struct{})
	var mu sync.Mutex
	var calls []string
	fetch := func(_ context.Context, _ *auth.Caller, tgt Target) (Result, error) {
		mu.Lock()
		calls = append(calls, tgt.URL)
		mu.Unlock()
		select {
		case started <- struct{}{}:
		default:
		}
		<-release
		return Result{Content: "ok"}, nil
	}
	r := newRunner(t, fetch, Config{})
	id, _, err := r.Submit(context.Background(), alice, dcache.ClassFleet, urls("a", "b", "c"))
	if err != nil {
		t.Fatal(err)
	}
	<-started // the first target is in flight
	if !r.Cancel(id) {
		t.Fatal("cancel returned false")
	}
	close(release) // let the in-flight target finish
	waitStatus(t, r, id, "cancelled")
	time.Sleep(20 * time.Millisecond)
	mu.Lock()
	n := len(calls)
	mu.Unlock()
	if n != 1 {
		t.Fatalf("cancelled after %d fetches, want 1", n)
	}
	j, ts, ok := r.Status(id)
	if !ok || j.Status != "cancelled" {
		t.Fatalf("job %+v: the in-flight target's write must not overwrite cancelled", j)
	}
	if ts[0].Status != "done" {
		t.Fatalf("in-flight target should be stored done: %+v", ts[0])
	}
	if ts[1].Status != "pending" || ts[2].Status != "pending" {
		t.Fatalf("later targets must stay pending: %+v", ts[1:])
	}
	if r.Cancel(id) {
		t.Fatal("cancelling a terminal job must return false")
	}
	if r.Cancel("no-such-job") {
		t.Fatal("cancelling an unknown job must return false")
	}
}

// A target that reports ErrRetryLater (the fetch gate was busy) is retried, not failed.
func TestBusyIsRetried(t *testing.T) {
	var mu sync.Mutex
	n := 0
	fetch := func(context.Context, *auth.Caller, Target) (Result, error) {
		mu.Lock()
		defer mu.Unlock()
		n++
		if n <= 2 {
			return Result{}, fmt.Errorf("busy: %w", ErrRetryLater)
		}
		return Result{Content: "ok"}, nil
	}
	r := newRunner(t, fetch, Config{RetryWait: time.Millisecond})
	id, _, err := r.Submit(context.Background(), alice, dcache.ClassFleet, urls("a"))
	if err != nil {
		t.Fatal(err)
	}
	waitStatus(t, r, id, "done")
	mu.Lock()
	defer mu.Unlock()
	if n != 3 {
		t.Fatalf("fetched %d times, want 3 (two busy, then ok)", n)
	}
}

// Retrying is bounded: a target busy for longer than RetryFor fails with a busy error.
func TestBusyGivesUp(t *testing.T) {
	fetch := func(context.Context, *auth.Caller, Target) (Result, error) {
		return Result{}, fmt.Errorf("busy: %w", ErrRetryLater)
	}
	r := newRunner(t, fetch, Config{RetryWait: time.Millisecond, RetryFor: 30 * time.Millisecond})
	id, _, err := r.Submit(context.Background(), alice, dcache.ClassFleet, urls("a"))
	if err != nil {
		t.Fatal(err)
	}
	waitStatus(t, r, id, "failed")
	_, ts, _ := r.Status(id)
	if ts[0].Status != "failed" || !strings.Contains(ts[0].Error, "busy") {
		t.Fatalf("target %+v, want failed with a busy error", ts[0])
	}
}

func TestPanicRecovered(t *testing.T) {
	fetch := func(_ context.Context, _ *auth.Caller, tgt Target) (Result, error) {
		if tgt.URL == "https://b.example/" {
			panic("boom")
		}
		return Result{Content: "ok"}, nil
	}
	r := newRunner(t, fetch, Config{})
	id, _, err := r.Submit(context.Background(), alice, dcache.ClassFleet, urls("a", "b", "c"))
	if err != nil {
		t.Fatal(err)
	}
	waitStatus(t, r, id, "partial")
	_, ts, _ := r.Status(id)
	if ts[0].Status != "done" || ts[1].Status != "failed" || ts[2].Status != "done" {
		t.Fatalf("statuses %+v", ts)
	}
	if ts[1].Error == "" || strings.Contains(ts[1].Error, "boom") {
		t.Fatalf("panic target error %q: must be set, and must not echo the panic value", ts[1].Error)
	}
}

// The hooks see every job outcome once and every stored target once.
func TestHooks(t *testing.T) {
	var mu sync.Mutex
	jobs := map[string]int{}
	var targets, jobSecs int
	var bytes int
	h := Hooks{
		Job: func(o string) { mu.Lock(); jobs[o]++; mu.Unlock() },
		Target: func(err error, _ time.Duration, n int) {
			mu.Lock()
			targets++
			bytes += n
			mu.Unlock()
		},
		JobSeconds: func(time.Duration) { mu.Lock(); jobSecs++; mu.Unlock() },
	}
	fetch := func(_ context.Context, _ *auth.Caller, tgt Target) (Result, error) {
		if tgt.URL == "https://b.example/" {
			return Result{}, errors.New("nope")
		}
		return Result{Content: "ok"}, nil
	}
	r := newRunner(t, fetch, Config{Hooks: h})
	id, _, _ := r.Submit(context.Background(), alice, dcache.ClassFleet, urls("a", "b"))
	waitStatus(t, r, id, "partial")
	_, _, _ = r.Submit(context.Background(), alice, dcache.ClassFleet, urls("a", "b")) // reused
	time.Sleep(20 * time.Millisecond)
	mu.Lock()
	defer mu.Unlock()
	if jobs["created"] != 1 || jobs["partial"] != 1 || jobs["reused"] != 1 || len(jobs) != 3 {
		t.Fatalf("job outcomes %v", jobs)
	}
	if targets != 2 || jobSecs != 1 {
		t.Fatalf("targets=%d jobSeconds=%d, want 2 and 1", targets, jobSecs)
	}
	if bytes <= 0 {
		t.Fatalf("stored result bytes %d, want > 0 for the done target", bytes)
	}
}