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) }