package work
import (
"context"
"errors"
"testing"
"time"
)
// A unit that has to wait on something other than its lane partway
// through — an ingest round queued for a rebuild slot — held its lane
// slot and a queue position the whole time, and ordinary units starved
// behind it. Aside gives both back for the length of the wait.
func TestAsideFreesTheSlotForTheLengthOfTheWait(t *testing.T) {
l := NewLane("test", 1, 0)
proceed := make(chan struct{})
aside := make(chan struct{})
done := make(chan error, 1)
go func() {
done <- l.DoNamedAside(context.Background(), "ingest", "a", func(stepAside func(func() error) error) error {
return stepAside(func() error {
close(aside)
<-proceed
return nil
})
})
}()
<-aside
if st := l.Stats(); st.InFlight != 0 || st.Waiting != 0 {
t.Errorf("FAIL: a unit set aside still counts against the lane: %+v", st)
}
if err := l.Do(context.Background(), func() error { return nil }); err != nil {
t.Fatalf("FAIL: another unit was refused while the first was aside: %v", err)
}
t.Log("ok: another unit runs while the first is aside")
close(proceed)
if err := <-done; err != nil {
t.Fatal(err)
}
if st := l.Stats(); st.InFlight != 0 || st.Waiting != 0 || st.Shed != 0 {
t.Errorf("FAIL: the lane is unbalanced after an aside: %+v", st)
}
}
// A wait that fails, or a context that ends while the unit waits to get
// back in, returns without the slot, and the lane stays balanced.
func TestAsideThatDoesNotComeBackLeavesTheLaneBalanced(t *testing.T) {
l := NewLane("test", 1, 1)
stopped := errors.New("stopped")
err := l.DoNamedAside(context.Background(), "ingest", "a", func(stepAside func(func() error) error) error {
return stepAside(func() error { return stopped })
})
if !errors.Is(err, stopped) {
t.Errorf("FAIL: a failed wait returned %v", err)
}
// Someone else takes the slot while this unit is aside, and the
// context ends while it waits to get back in.
hold := make(chan struct{})
held := make(chan struct{})
others := make(chan error, 1)
ctx, cancel := context.WithCancel(context.Background())
err = l.DoNamedAside(ctx, "ingest", "b", func(stepAside func(func() error) error) error {
return stepAside(func() error {
go func() {
others <- l.Do(context.Background(), func() error { close(held); <-hold; return nil })
}()
<-held
time.AfterFunc(20*time.Millisecond, cancel)
return nil
})
})
if !errors.Is(err, context.Canceled) {
t.Errorf("FAIL: a unit that could not get back in returned %v", err)
}
close(hold)
if err := <-others; err != nil {
t.Fatal(err)
}
if st := l.Stats(); st.InFlight != 0 || st.Waiting != 0 {
t.Errorf("FAIL: the lane is unbalanced: %+v", st)
} else {
t.Log("ok: the lane is balanced after asides that did not come back")
}
}