sift/internal/batch/review_test.go

package batch

// P6 owner review: regression tests for the runner (findings from the independent review).

import (
	"context"
	"fmt"
	"sync"
	"sync/atomic"
	"testing"
	"time"

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

// A job cancelled while its in-flight target comes back busy leaves the rotation.
func TestCancelWhileBusyLeavesRotation(t *testing.T) {
	started := make(chan struct{}, 1)
	release := make(chan struct{})
	var depth atomic.Int64
	fetch := func(context.Context, *auth.Caller, Target) (Result, error) {
		select {
		case started <- struct{}{}:
		default:
		}
		<-release
		return Result{}, fmt.Errorf("busy: %w", ErrRetryLater)
	}
	r := newRunner(t, fetch, Config{RetryWait: time.Millisecond, Hooks: Hooks{Depth: func(n int) { depth.Store(int64(n)) }}})
	id, _, err := r.Submit(context.Background(), alice, dcache.ClassFleet, urls("a", "b"))
	if err != nil {
		t.Fatal(err)
	}
	<-started
	if !r.Cancel(id) {
		t.Fatal("cancel returned false")
	}
	close(release)
	deadline := time.Now().Add(2 * time.Second)
	for time.Now().Before(deadline) {
		r.mu.Lock()
		n := len(r.live)
		r.mu.Unlock()
		if n == 0 && depth.Load() == 0 {
			return
		}
		time.Sleep(5 * time.Millisecond)
	}
	t.Fatalf("cancelled job still in the rotation (depth %d)", depth.Load())
}

// Close stops the workers: no further target is fetched, and the job is left live in the
// store for MarkInFlightFailed("restart") to reap, not failed target by target.
func TestCloseStopsDraining(t *testing.T) {
	started := make(chan struct{}, 1)
	var calls atomic.Int64
	fetch := func(ctx context.Context, _ *auth.Caller, _ Target) (Result, error) {
		calls.Add(1)
		select {
		case started <- struct{}{}:
		default:
		}
		<-ctx.Done()
		return Result{}, ctx.Err()
	}
	r := newRunner(t, fetch, Config{})
	id, _, err := r.Submit(context.Background(), alice, dcache.ClassFleet, urls("a", "b", "c", "d"))
	if err != nil {
		t.Fatal(err)
	}
	<-started
	r.Close()
	if n := calls.Load(); n != 1 {
		t.Fatalf("%d fetches after Close, want only the one in flight", n)
	}
	j, ts, _ := r.Status(id)
	if j.Status != "running" && j.Status != "queued" {
		t.Fatalf("job %q after Close, want it left live for the restart reaper", j.Status)
	}
	for _, x := range ts {
		if x.Status != "pending" {
			t.Fatalf("target %d is %q after Close, want pending", x.Idx, x.Status)
		}
	}
}

// Concurrent submits respect the caps and idempotency: one job per caller, and identical
// lists all get the same job.
func TestConcurrentSubmits(t *testing.T) {
	release := make(chan struct{})
	defer close(release)
	fetch := func(context.Context, *auth.Caller, Target) (Result, error) {
		<-release
		return Result{Content: "ok"}, nil
	}
	r := newRunner(t, fetch, Config{})
	var wg sync.WaitGroup
	var mu sync.Mutex
	ids := map[string]bool{}
	for i := 0; i < 20; i++ {
		wg.Add(1)
		go func(i int) {
			defer wg.Done()
			id, _, err := r.Submit(context.Background(), alice, dcache.ClassFleet, urls(fmt.Sprintf("u%d", i)))
			if err == nil {
				mu.Lock()
				ids[id] = true
				mu.Unlock()
			}
		}(i)
	}
	wg.Wait()
	if len(ids) != 1 {
		t.Fatalf("20 concurrent submits by one caller (cap 1) created %d jobs", len(ids))
	}
	same := map[string]bool{}
	for i := 0; i < 20; i++ {
		wg.Add(1)
		go func() {
			defer wg.Done()
			id, _, err := r.Submit(context.Background(), bob, dcache.ClassFleet, urls("same"))
			if err == nil {
				mu.Lock()
				same[id] = true
				mu.Unlock()
			}
		}()
	}
	wg.Wait()
	if len(same) != 1 {
		t.Fatalf("20 concurrent identical submits created %d jobs, want 1", len(same))
	}
}

// A second cancel of a job with a target in flight returns false and counts once.
func TestDoubleCancel(t *testing.T) {
	started := make(chan struct{}, 1)
	release := make(chan struct{})
	var cancels atomic.Int64
	fetch := func(context.Context, *auth.Caller, Target) (Result, error) {
		select {
		case started <- struct{}{}:
		default:
		}
		<-release
		return Result{Content: "ok"}, nil
	}
	r := newRunner(t, fetch, Config{Hooks: Hooks{Job: func(o string) {
		if o == "cancelled" {
			cancels.Add(1)
		}
	}}})
	defer close(release)
	id, _, _ := r.Submit(context.Background(), alice, dcache.ClassFleet, urls("a", "b"))
	<-started
	if !r.Cancel(id) {
		t.Fatal("first cancel returned false")
	}
	if r.Cancel(id) {
		t.Fatal("second cancel returned true")
	}
	if n := cancels.Load(); n != 1 {
		t.Fatalf("cancelled counted %d times", n)
	}
}

// Negative or zero durations and sizes fall back to the defaults.
func TestConfigDefaults(t *testing.T) {
	r := newRunner(t, func(context.Context, *auth.Caller, Target) (Result, error) { return Result{}, nil },
		Config{TTL: -time.Hour, MaxBytes: -1, RetryWait: -1, RetryFor: -1, TargetsPerJob: -1, JobsPerCaller: -1, JobsTotal: -1, Workers: -1})
	c := r.cfg
	if c.TTL != 24*time.Hour || c.MaxBytes != 64<<20 || c.RetryWait != 2*time.Second || c.RetryFor != 10*time.Minute ||
		c.TargetsPerJob != 32 || c.JobsPerCaller != 1 || c.JobsTotal != 8 || c.Workers != 1 {
		t.Fatalf("config %+v", c)
	}
}