package work
import (
"context"
"errors"
"runtime"
"strings"
"sync"
"sync/atomic"
"testing"
"time"
"craftdesign.group/quarry/internal/fault"
)
// The limit is the whole point: without it the file page forked a
// subprocess per request and throughput went backwards under load.
func TestLaneNeverExceedsItsLimit(t *testing.T) {
l := NewLane("test", 3, 100)
var now, peak atomic.Int64
var wg sync.WaitGroup
for i := 0; i < 40; i++ {
wg.Add(1)
go func() {
defer wg.Done()
l.Do(context.Background(), func() error {
n := now.Add(1)
for {
p := peak.Load()
if n <= p || peak.CompareAndSwap(p, n) {
break
}
}
time.Sleep(2 * time.Millisecond)
now.Add(-1)
return nil
})
}()
}
wg.Wait()
if got := peak.Load(); got > 3 {
t.Errorf("ran %d concurrently, limit is 3", got)
}
if got := peak.Load(); got < 2 {
t.Errorf("peak concurrency %d — the lane serialized instead of admitting its limit", got)
}
}
// Past the cap a lane must REFUSE, not accept work it cannot serve.
// An accepted request that times out consumed the capacity that would
// have served someone else.
func TestLaneShedsPastItsQueue(t *testing.T) {
l := NewLane("test", 1, 0) // one at a time, no waiting room
release := make(chan struct{})
started := make(chan struct{})
go l.Do(context.Background(), func() error {
close(started)
<-release
return nil
})
<-started
err := l.Do(context.Background(), func() error {
t.Error("ran a function the lane should have refused")
return nil
})
if !errors.Is(err, ErrBusy) {
t.Fatalf("err = %v, want ErrBusy", err)
}
if s := l.Stats(); s.Shed != 1 {
t.Errorf("Shed = %d, want 1 — the refusal must be counted, it is the scaling signal", s.Shed)
}
close(release)
}
// A caller who gives up while queued must not run later against a
// context nobody is waiting on.
func TestLaneHonoursCancellation(t *testing.T) {
l := NewLane("test", 1, 10)
release := make(chan struct{})
started := make(chan struct{})
go l.Do(context.Background(), func() error {
close(started)
<-release
return nil
})
<-started
ctx, cancel := context.WithCancel(context.Background())
cancel()
var ran bool
err := l.Do(ctx, func() error { ran = true; return nil })
if !errors.Is(err, context.Canceled) {
t.Errorf("err = %v, want context.Canceled", err)
}
if ran {
t.Error("ran the function for a caller who had already given up")
}
close(release)
}
// The error from fn must reach the caller unchanged; a lane is a gate,
// not a wrapper.
func TestLanePassesErrorsThrough(t *testing.T) {
l := NewLane("test", 2, 2)
want := errors.New("boom")
if got := l.Do(context.Background(), func() error { return want }); !errors.Is(got, want) {
t.Errorf("err = %v, want %v", got, want)
}
}
// A nil lane runs the work — so a caller with no lane configured is not
// a special case at every call site.
func TestNilLaneRuns(t *testing.T) {
var l *Lane
ran := false
if err := l.Do(context.Background(), func() error { ran = true; return nil }); err != nil {
t.Fatal(err)
}
if !ran {
t.Error("nil lane did not run the work")
}
}
// A refusal has to say which lane, how full, and what it costs: the
// bare sentinel filled the journal 833 times naming none of it, while
// the lane raising it knew every number.
func TestShedRefusalCarriesTheLanesLoad(t *testing.T) {
l := NewLane("ingest", 1, 0)
release := make(chan struct{})
go l.Do(context.Background(), func() error { <-release; return nil })
// Wait for the goroutine to hold the only slot.
for l.Stats().InFlight == 0 {
runtime.Gosched()
}
err := l.Do(context.Background(), func() error { return nil })
close(release)
if !errors.Is(err, ErrBusy) {
t.Fatalf("a shed call must still compare as busy: %v", err)
}
f := fault.Of(err)
if f == nil {
t.Fatal("the refusal carries no context")
}
if f.Why == "" || f.Fix == "" {
t.Error("a refusal that says neither what it costs nor what to do is not actionable")
}
want := []string{"lane", "limit", "queue", "running", "shed", "waiting"}
if strings.Join(f.Keys(), " ") != strings.Join(want, " ") {
t.Errorf("fields %v, want %v", f.Keys(), want)
}
for _, fld := range f.Fields {
if fld.Key == "lane" && fld.Value != "ingest" {
t.Errorf("the refusal names lane %q", fld.Value)
}
if fld.Key == "shed" && fld.Value != "1" {
t.Errorf("shed count %q, want the refusal to be counted", fld.Value)
}
}
// The message stays stable so every refusal from this lane is one
// row; the numbers live in the fields, which move freely.
if f.Msg != "ingest lane is full" {
t.Errorf("message %q carries something that moves", f.Msg)
}
}