diff --git a/internal/progress/tracker_stall_test.go b/internal/progress/tracker_stall_test.go new file mode 100644 index 0000000..ce59436 --- /dev/null +++ b/internal/progress/tracker_stall_test.go @@ -0,0 +1,98 @@ +package progress_test + +// Observes what a database that stops answering does to the job it reports on (#51): +// progress is cosmetic, so a save that never returns must not stall the engine calls that +// report through the tracker, nor the heartbeat that says the process is alive. + +import ( + "bytes" + "context" + "log/slog" + "strings" + "sync" + "sync/atomic" + "testing" + "time" + + "github.com/stretchr/testify/require" + + "git.d-ma.be/mathias/tapir/internal/domain" + "git.d-ma.be/mathias/tapir/internal/progress" +) + +// stallSink never answers until its context ends, like a database connection that hung. +type stallSink struct{ beats atomic.Int32 } + +func (s *stallSink) SaveProgress(ctx context.Context, _ []domain.ProgressStep, _ time.Time) error { + <-ctx.Done() + return ctx.Err() +} + +func (s *stallSink) Beat(ctx context.Context) error { + s.beats.Add(1) + <-ctx.Done() + return ctx.Err() +} + +type lockedBuf struct { + mu sync.Mutex + b bytes.Buffer +} + +func (l *lockedBuf) Write(p []byte) (int, error) { + l.mu.Lock() + defer l.mu.Unlock() + return l.b.Write(p) +} + +func (l *lockedBuf) String() string { + l.mu.Lock() + defer l.mu.Unlock() + return l.b.String() +} + +func stallTracker(sink progress.Sink, buf *lockedBuf) *progress.Tracker { + log := slog.New(slog.NewTextHandler(buf, nil)) + return progress.NewTracker(context.Background(), sink, log, + []progress.Planned{{Key: "a", Label: "A"}}, progress.WithSaveTimeout(50*time.Millisecond)) +} + +func TestTracker_AHungSaveDoesNotBlockTheReporterPastItsTimeout(t *testing.T) { + buf := &lockedBuf{} + tr := stallTracker(&stallSink{}, buf) + + done := make(chan struct{}) + go func() { tr.Begin("a", 0); tr.Advance("a", 1); tr.Done("a"); close(done) }() + + select { + case <-done: + case <-time.After(3 * time.Second): + t.Fatal("the reporter was blocked by a save that never returned") + } + require.Contains(t, buf.String(), "progress: not saved", "a dropped save is logged") +} + +func TestTracker_AHungBeatDoesNotStopTheHeartbeat(t *testing.T) { + sink := &stallSink{} + tr := stallTracker(sink, &lockedBuf{}) + ctx, cancel := context.WithCancel(context.Background()) + defer cancel() + + go tr.Heartbeat(ctx, 5*time.Millisecond, time.Hour) + + require.Eventually(t, func() bool { return sink.beats.Load() >= 3 }, 3*time.Second, 5*time.Millisecond, + "each beat is abandoned at its timeout and the next one is tried") +} + +func TestTracker_TheStillRunningLineIsNotHeldUpBehindAHungSave(t *testing.T) { + buf := &lockedBuf{} + tr := stallTracker(&stallSink{}, buf) + go tr.Begin("a", 0) // holds the tracker's lock inside a save that hangs until its timeout + time.Sleep(10 * time.Millisecond) + 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") }, 3*time.Second, 5*time.Millisecond) +}