403 lines
12 KiB
Go
403 lines
12 KiB
Go
package progress_test
|
|
|
|
import (
|
|
"context"
|
|
"errors"
|
|
"log/slog"
|
|
"strings"
|
|
"sync"
|
|
"testing"
|
|
"time"
|
|
|
|
"github.com/stretchr/testify/require"
|
|
|
|
"git.d-ma.be/mathias/tapir/internal/domain"
|
|
"git.d-ma.be/mathias/tapir/internal/progress"
|
|
)
|
|
|
|
// fakeSink records every saved snapshot and every heartbeat.
|
|
type fakeSink struct {
|
|
mu sync.Mutex
|
|
saves [][]domain.ProgressStep
|
|
at []time.Time
|
|
beats int
|
|
err error
|
|
}
|
|
|
|
func (f *fakeSink) SaveProgress(_ context.Context, steps []domain.ProgressStep, at time.Time) error {
|
|
f.mu.Lock()
|
|
defer f.mu.Unlock()
|
|
f.saves = append(f.saves, append([]domain.ProgressStep(nil), steps...))
|
|
f.at = append(f.at, at)
|
|
return f.err
|
|
}
|
|
|
|
func (f *fakeSink) Beat(context.Context) error {
|
|
f.mu.Lock()
|
|
defer f.mu.Unlock()
|
|
f.beats++
|
|
return f.err
|
|
}
|
|
|
|
func (f *fakeSink) last(t *testing.T) []domain.ProgressStep {
|
|
t.Helper()
|
|
f.mu.Lock()
|
|
defer f.mu.Unlock()
|
|
require.NotEmpty(t, f.saves, "nothing was saved")
|
|
return f.saves[len(f.saves)-1]
|
|
}
|
|
|
|
func (f *fakeSink) saveCount() int {
|
|
f.mu.Lock()
|
|
defer f.mu.Unlock()
|
|
return len(f.saves)
|
|
}
|
|
|
|
func (f *fakeSink) beatCount() int {
|
|
f.mu.Lock()
|
|
defer f.mu.Unlock()
|
|
return f.beats
|
|
}
|
|
|
|
// syncBuf is a log sink safe to read while the heartbeat goroutine is still writing.
|
|
type syncBuf struct {
|
|
mu sync.Mutex
|
|
b strings.Builder
|
|
}
|
|
|
|
func (s *syncBuf) Write(p []byte) (int, error) {
|
|
s.mu.Lock()
|
|
defer s.mu.Unlock()
|
|
return s.b.Write(p)
|
|
}
|
|
|
|
func (s *syncBuf) String() string {
|
|
s.mu.Lock()
|
|
defer s.mu.Unlock()
|
|
return s.b.String()
|
|
}
|
|
|
|
// clock is a hand-wound clock so durations and throttling are exact.
|
|
type clock struct{ t time.Time }
|
|
|
|
func (c *clock) now() time.Time { return c.t }
|
|
func (c *clock) tick(d time.Duration) { c.t = c.t.Add(d) }
|
|
func newClock() *clock { return &clock{t: time.Date(2026, 10, 2, 8, 0, 0, 0, time.UTC)} }
|
|
|
|
func plan() []progress.Planned {
|
|
return []progress.Planned{{Key: "audio", Label: "Preparing audio"}, {Key: "decide", Label: "Deciding"}, {Key: "save", Label: "Saving"}}
|
|
}
|
|
|
|
func newTracker(t *testing.T) (*progress.Tracker, *fakeSink, *clock, *syncBuf) {
|
|
t.Helper()
|
|
sink, clk, buf := &fakeSink{}, newClock(), &syncBuf{}
|
|
log := slog.New(slog.NewTextHandler(buf, &slog.HandlerOptions{Level: slog.LevelDebug}))
|
|
return progress.NewTracker(context.Background(), sink, log, plan(), progress.WithClock(clk.now), progress.WithMinSave(2*time.Second)), sink, clk, buf
|
|
}
|
|
|
|
func stateOf(steps []domain.ProgressStep, key string) domain.StepState {
|
|
for _, s := range steps {
|
|
if s.Key == key {
|
|
return s.State
|
|
}
|
|
}
|
|
return ""
|
|
}
|
|
|
|
func TestTracker_PlanIsShownUpFrontInOrderAllPending(t *testing.T) {
|
|
tr, sink, _, _ := newTracker(t)
|
|
|
|
tr.Start()
|
|
|
|
steps := sink.last(t)
|
|
require.Len(t, steps, 3)
|
|
require.Equal(t, []string{"audio", "decide", "save"}, []string{steps[0].Key, steps[1].Key, steps[2].Key})
|
|
for _, s := range steps {
|
|
require.Equal(t, domain.StepPending, s.State, s.Key)
|
|
}
|
|
require.Equal(t, "Preparing audio", steps[0].Label)
|
|
}
|
|
|
|
func TestTracker_BeginRunsAStepAndDoneRecordsItsDuration(t *testing.T) {
|
|
tr, sink, clk, _ := newTracker(t)
|
|
tr.Start()
|
|
|
|
tr.Begin("audio", 0)
|
|
require.Equal(t, domain.StepRunning, stateOf(sink.last(t), "audio"))
|
|
clk.tick(90 * time.Second)
|
|
tr.Done("audio")
|
|
|
|
s := sink.last(t)[0]
|
|
require.Equal(t, domain.StepDone, s.State)
|
|
require.Equal(t, 90*time.Second, s.FinishedAt.Sub(s.StartedAt), "the finished step must carry how long it took")
|
|
}
|
|
|
|
func TestTracker_AdvanceCountsAreSavedAndThrottled(t *testing.T) {
|
|
tr, sink, clk, _ := newTracker(t)
|
|
tr.Start()
|
|
tr.Begin("decide", 282)
|
|
base := sink.saveCount()
|
|
|
|
tr.Advance("decide", 1)
|
|
clk.tick(500 * time.Millisecond)
|
|
tr.Advance("decide", 2)
|
|
clk.tick(500 * time.Millisecond)
|
|
tr.Advance("decide", 3)
|
|
|
|
require.Equal(t, base+1, sink.saveCount(), "advances inside the minimum interval must not each hit the database")
|
|
clk.tick(2 * time.Second)
|
|
tr.Advance("decide", 4)
|
|
require.Equal(t, base+2, sink.saveCount())
|
|
got := sink.last(t)[1]
|
|
require.Equal(t, 4, got.Done)
|
|
require.Equal(t, 282, got.Total)
|
|
}
|
|
|
|
func TestTracker_DoneFlushesTheFinalCountEvenInsideTheThrottle(t *testing.T) {
|
|
tr, sink, _, _ := newTracker(t)
|
|
tr.Start()
|
|
tr.Begin("decide", 10)
|
|
tr.Advance("decide", 9)
|
|
|
|
tr.Advance("decide", 10)
|
|
tr.Done("decide")
|
|
|
|
got := sink.last(t)[1]
|
|
require.Equal(t, domain.StepDone, got.State)
|
|
require.Equal(t, 10, got.Done, "a finished step must show its full count, not the last throttled one")
|
|
}
|
|
|
|
func TestTracker_FinishSkipsWhatNeverRanAndClosesWhatDid(t *testing.T) {
|
|
tr, sink, _, _ := newTracker(t)
|
|
tr.Start()
|
|
tr.Begin("audio", 0)
|
|
tr.Done("audio")
|
|
tr.Begin("decide", 5)
|
|
|
|
tr.Finish()
|
|
|
|
steps := sink.last(t)
|
|
require.Equal(t, domain.StepDone, stateOf(steps, "audio"))
|
|
require.Equal(t, domain.StepDone, stateOf(steps, "decide"), "a step still running when the job completes is closed")
|
|
require.Equal(t, domain.StepSkipped, stateOf(steps, "save"), "a step that never ran is skipped, not left pending")
|
|
}
|
|
|
|
func TestTracker_FailRunningMarksTheRunningStepFailedAndLeavesTheRestPending(t *testing.T) {
|
|
tr, sink, _, _ := newTracker(t)
|
|
tr.Start()
|
|
tr.Begin("audio", 0)
|
|
tr.Done("audio")
|
|
tr.Begin("decide", 5)
|
|
|
|
tr.FailRunning()
|
|
|
|
steps := sink.last(t)
|
|
require.Equal(t, domain.StepDone, stateOf(steps, "audio"))
|
|
require.Equal(t, domain.StepFailed, stateOf(steps, "decide"))
|
|
require.Equal(t, domain.StepPending, stateOf(steps, "save"), "what never started stays pending: it shows where the job stopped")
|
|
}
|
|
|
|
func TestTracker_AKeyedFailMarksOneStepFailedLogsItsKindAndTheJobGoesOn(t *testing.T) {
|
|
tr, sink, _, buf := newTracker(t)
|
|
tr.Start()
|
|
tr.Begin("audio", 0)
|
|
|
|
tr.Fail("audio", errors.New("gateway said no"))
|
|
tr.Begin("decide", 5)
|
|
|
|
steps := sink.last(t)
|
|
require.Equal(t, domain.StepFailed, stateOf(steps, "audio"))
|
|
require.Equal(t, domain.StepRunning, stateOf(steps, "decide"), "a soft failure must not stop the next step from running")
|
|
require.Contains(t, buf.String(), "level=WARN")
|
|
require.NotContains(t, buf.String(), "gateway said no", "an error's text can carry provider output or a path: the log gets its kind")
|
|
require.Contains(t, buf.String(), "kind=error")
|
|
require.Contains(t, buf.String(), "step=audio")
|
|
}
|
|
|
|
func TestTracker_NoteIsALogLineAndNeverASave(t *testing.T) {
|
|
tr, sink, _, buf := newTracker(t)
|
|
tr.Start()
|
|
before := sink.saveCount()
|
|
|
|
tr.Note("reconciled", "disputes", 362, "auto_resolved", 80)
|
|
|
|
require.Equal(t, before, sink.saveCount(), "a note is for the operator; it must not cost a database write")
|
|
require.Contains(t, buf.String(), "reconciled")
|
|
require.Contains(t, buf.String(), "disputes=362")
|
|
require.Contains(t, buf.String(), "auto_resolved=80")
|
|
}
|
|
|
|
func TestTracker_AnUnknownKeyIsIgnored(t *testing.T) {
|
|
tr, sink, _, _ := newTracker(t)
|
|
tr.Start()
|
|
before := sink.saveCount()
|
|
|
|
require.NotPanics(t, func() {
|
|
tr.Begin("no-such-step", 0)
|
|
tr.Advance("no-such-step", 1)
|
|
tr.Done("no-such-step")
|
|
tr.Fail("no-such-step", errors.New("x"))
|
|
})
|
|
require.Equal(t, before, sink.saveCount())
|
|
}
|
|
|
|
func TestTracker_ASinkFailureIsLoggedAndNeverStopsTheJob(t *testing.T) {
|
|
tr, sink, _, buf := newTracker(t)
|
|
sink.err = errors.New("db down")
|
|
|
|
require.NotPanics(t, func() {
|
|
tr.Start()
|
|
tr.Begin("audio", 0)
|
|
tr.Done("audio")
|
|
})
|
|
require.Contains(t, buf.String(), "level=WARN", "a lost progress write is worth a WARN")
|
|
require.Contains(t, buf.String(), "db down")
|
|
}
|
|
|
|
func TestTracker_EveryTransitionIsLoggedByTheSameCall(t *testing.T) {
|
|
tr, _, clk, buf := newTracker(t)
|
|
tr.Start()
|
|
tr.Begin("audio", 0)
|
|
clk.tick(3 * time.Second)
|
|
tr.Done("audio")
|
|
tr.Begin("decide", 5)
|
|
clk.tick(2 * time.Second)
|
|
tr.FailRunning()
|
|
tr.Finish()
|
|
|
|
out := buf.String()
|
|
for _, want := range []string{
|
|
`msg="job started"`,
|
|
`msg="step started" step=audio`,
|
|
`msg="step finished" step=audio elapsed=3s`,
|
|
`msg="step started" step=decide total=5`,
|
|
`msg="step failed" step=decide elapsed=2s`,
|
|
`msg="job finished"`,
|
|
} {
|
|
require.Contains(t, out, want, "a transition with no log line is invisible to the operator")
|
|
}
|
|
}
|
|
|
|
func TestTracker_AKeyedFailIsSavedAtOnceNotOnTheNextTransition(t *testing.T) {
|
|
tr, sink, _, _ := newTracker(t)
|
|
tr.Start()
|
|
tr.Begin("audio", 0)
|
|
before := sink.saveCount()
|
|
|
|
tr.Fail("audio", errors.New("down"))
|
|
|
|
require.Equal(t, before+1, sink.saveCount(), "a soft-failed phase must show as failed on the page now")
|
|
require.Equal(t, domain.StepFailed, stateOf(sink.last(t), "audio"))
|
|
}
|
|
|
|
// retainingSink keeps the very slice it is handed, like a sink that queues it for later.
|
|
type retainingSink struct{ got [][]domain.ProgressStep }
|
|
|
|
func (r *retainingSink) SaveProgress(_ context.Context, steps []domain.ProgressStep, _ time.Time) error {
|
|
r.got = append(r.got, steps)
|
|
return nil
|
|
}
|
|
func (r *retainingSink) Beat(context.Context) error { return nil }
|
|
|
|
func TestTracker_ASnapshotHandedToTheSinkIsNotChangedByLaterTransitions(t *testing.T) {
|
|
sink := &retainingSink{}
|
|
tr := progress.NewTracker(context.Background(), sink, slog.New(slog.NewTextHandler(&syncBuf{}, nil)), plan(),
|
|
progress.WithClock(newClock().now))
|
|
tr.Start()
|
|
first := sink.got[0]
|
|
|
|
tr.Begin("audio", 0)
|
|
tr.Done("audio")
|
|
|
|
require.Equal(t, domain.StepPending, first[0].State, "a sink that keeps a snapshot must not see it change under it")
|
|
}
|
|
|
|
func TestTracker_AdvanceIsLoggedAtMostOncePerMinute(t *testing.T) {
|
|
tr, _, clk, buf := newTracker(t)
|
|
tr.Start()
|
|
tr.Begin("decide", 100)
|
|
|
|
for i := 1; i <= 30; i++ {
|
|
clk.tick(1 * time.Second)
|
|
tr.Advance("decide", i)
|
|
}
|
|
require.LessOrEqual(t, strings.Count(buf.String(), "step progress"), 1,
|
|
"30 seconds of progress is at most one log line, not thirty")
|
|
clk.tick(61 * time.Second)
|
|
tr.Advance("decide", 31)
|
|
require.Equal(t, 2, strings.Count(buf.String(), "step progress"))
|
|
require.Contains(t, buf.String(), "done=31")
|
|
require.Contains(t, buf.String(), "total=100")
|
|
}
|
|
|
|
func TestTracker_HeartbeatBeatsOnItsOwnUntilStopped(t *testing.T) {
|
|
tr, sink, _, _ := newTracker(t)
|
|
ctx, cancel := context.WithCancel(context.Background())
|
|
done := make(chan struct{})
|
|
|
|
go func() {
|
|
defer close(done)
|
|
tr.Heartbeat(ctx, 5*time.Millisecond, time.Hour)
|
|
}()
|
|
|
|
require.Eventually(t, func() bool { return sink.beatCount() >= 3 }, 2*time.Second, 5*time.Millisecond,
|
|
"a heartbeat that does not depend on any engine call is the point")
|
|
cancel()
|
|
select {
|
|
case <-done:
|
|
case <-time.After(2 * time.Second):
|
|
t.Fatal("Heartbeat did not stop when its context ended")
|
|
}
|
|
}
|
|
|
|
func TestTracker_HeartbeatSaysWhichStepIsStillRunningAndForHowLong(t *testing.T) {
|
|
tr, _, clk, buf := newTracker(t)
|
|
tr.Start()
|
|
tr.Begin("decide", 100)
|
|
clk.tick(7 * time.Minute)
|
|
ctx, cancel := context.WithCancel(context.Background())
|
|
defer cancel()
|
|
|
|
go tr.Heartbeat(ctx, time.Hour, 5*time.Millisecond)
|
|
|
|
require.Eventually(t, func() bool { return strings.Contains(buf.String(), "still running") }, 2*time.Second, 5*time.Millisecond)
|
|
require.Contains(t, buf.String(), "step=decide")
|
|
require.Contains(t, buf.String(), "elapsed=7m0s")
|
|
}
|
|
|
|
func TestFrom_WithoutAReporterReturnsOneThatDoesNothing(t *testing.T) {
|
|
r := progress.From(context.Background())
|
|
|
|
require.NotNil(t, r)
|
|
require.NotPanics(t, func() {
|
|
r.Begin("x", 3)
|
|
r.Advance("x", 1)
|
|
r.Done("x")
|
|
r.Fail("x", errors.New("e"))
|
|
r.Note("n", "k", 1)
|
|
})
|
|
}
|
|
|
|
func TestWithAndFrom_ACarriedReporterIsTheOneReturned(t *testing.T) {
|
|
tr, sink, _, _ := newTracker(t)
|
|
tr.Start()
|
|
ctx := progress.With(context.Background(), tr)
|
|
|
|
progress.From(ctx).Begin("audio", 0)
|
|
|
|
require.Equal(t, domain.StepRunning, stateOf(sink.last(t), "audio"))
|
|
}
|
|
|
|
func TestTracker_AnnotateAttachesANoteToAStepAndSavesItAtOnce(t *testing.T) {
|
|
tr, sink, _, _ := newTracker(t)
|
|
tr.Start()
|
|
tr.Begin("audio", 0)
|
|
before := sink.saveCount()
|
|
|
|
tr.Annotate("audio", "20 min of the 79 the file claimed")
|
|
|
|
require.Equal(t, before+1, sink.saveCount())
|
|
require.Equal(t, "20 min of the 79 the file claimed", sink.last(t)[0].Note)
|
|
require.NotPanics(t, func() { tr.Annotate("no-such-step", "x") })
|
|
}
|