tapir/internal/progress/tracker_stall_test.go

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, _ time.Time) 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)
}

// deadlineSink records how long each write was given, and answers at once.
type deadlineSink struct {
	mu     sync.Mutex
	saves  []time.Duration
	beats  []time.Duration
	noDLOK bool
}

func (d *deadlineSink) record(ctx context.Context, into *[]time.Duration) {
	d.mu.Lock()
	defer d.mu.Unlock()
	dl, ok := ctx.Deadline()
	if !ok {
		d.noDLOK = true
		return
	}
	*into = append(*into, time.Until(dl))
}

func (d *deadlineSink) SaveProgress(ctx context.Context, _ []domain.ProgressStep, _ time.Time) error {
	d.record(ctx, &d.saves)
	return nil
}

func (d *deadlineSink) Beat(ctx context.Context, _ time.Time) error {
	d.record(ctx, &d.beats)
	return nil
}

// The production value is the one the worker gets: it builds a Tracker without options.
// A zero or tiny default would drop every write, a huge one would bound nothing.
func TestTracker_TheDefaultSaveAndBeatDeadlineIsAFewSecondsAway(t *testing.T) {
	sink := &deadlineSink{}
	tr := progress.NewTracker(context.Background(), sink, slog.New(slog.NewTextHandler(&lockedBuf{}, nil)),
		[]progress.Planned{{Key: "a", Label: "A"}})
	ctx, cancel := context.WithCancel(context.Background())
	defer cancel()
	go tr.Heartbeat(ctx, 5*time.Millisecond, time.Hour)

	tr.Begin("a", 0)
	require.Eventually(t, func() bool {
		sink.mu.Lock()
		defer sink.mu.Unlock()
		return len(sink.beats) > 0
	}, 3*time.Second, 5*time.Millisecond)

	sink.mu.Lock()
	defer sink.mu.Unlock()
	require.False(t, sink.noDLOK, "a write was given no deadline at all")
	require.NotEmpty(t, sink.saves)
	for _, d := range append(append([]time.Duration{}, sink.saves...), sink.beats...) {
		require.Greater(t, d, time.Second, "the deadline is already (nearly) spent when the write starts")
		require.LessOrEqual(t, d, 5*time.Second, "the default bound is 5 s")
	}
}