internal/uploads/worker_progress_test.go

@@ -0,0 +1,164 @@
+package uploads_test
+
+import (
+	"bytes"
+	"context"
+	"errors"
+	"log/slog"
+	"strings"
+	"sync"
+	"testing"
+	"time"
+
+	"github.com/stretchr/testify/assert"
+	"github.com/stretchr/testify/require"
+
+	"git.d-ma.be/mathias/tapir/internal/domain"
+	"git.d-ma.be/mathias/tapir/internal/progress"
+	"git.d-ma.be/mathias/tapir/internal/uploads"
+)
+
+// Progress, heartbeat and logging of the worker (ADR-035, #42, #43).
+
+// plannedTr is a transcriber that declares its own phases and reports them, like the
+// two-engine Ensemble does.
+type plannedTr struct{ *fakeTranscriber }
+
+func (plannedTr) Plan() []progress.Planned {
+	return []progress.Planned{{Key: "engine_a", Label: "Transcribing with A"}, {Key: "engine_b", Label: "Transcribing with B"}}
+}
+
+func (p plannedTr) Transcribe(ctx context.Context, wav string) (domain.Transcription, error) {
+	r := progress.From(ctx)
+	r.Begin("engine_a", 0)
+	r.Done("engine_a")
+	r.Begin("engine_b", 0)
+	r.Done("engine_b")
+	return p.fakeTranscriber.Transcribe(ctx, wav)
+}
+
+func keysAndStates(steps []domain.ProgressStep) []string {
+	var out []string
+	for _, s := range steps {
+		out = append(out, s.Key+":"+string(s.State))
+	}
+	return out
+}
+
+// syncLog is a log sink safe to read while the worker's goroutines write.
+type syncLog struct {
+	mu sync.Mutex
+	b  bytes.Buffer
+}
+
+func (l *syncLog) Write(p []byte) (int, error) {
+	l.mu.Lock()
+	defer l.mu.Unlock()
+	return l.b.Write(p)
+}
+
+func (l *syncLog) String() string {
+	l.mu.Lock()
+	defer l.mu.Unlock()
+	return l.b.String()
+}
+
+func newProgressWorker(t *testing.T, r *wrig, tr uploads.Transcriber, opts ...uploads.WorkerOption) *uploads.Worker {
+	t.Helper()
+	return uploads.NewWorker(r.st, r.conv, tr, r.tmp, time.Minute, opts...)
+}
+
+func TestWorkerProgress_ASingleEngineJobShowsEveryPhaseDone(t *testing.T) {
+	r := newWRig(t, time.Minute)
+	r.w.Process(context.Background(), r.spool(t, "a"))
+
+	assert.Equal(t, []string{"received:done", "audio:done", "transcribe:done", "save:done"}, keysAndStates(r.st.lastSteps()))
+}
+
+func TestWorkerProgress_ATranscriberThatPlansItsOwnPhasesGetsThemBetweenAudioAndSaving(t *testing.T) {
+	r := newWRig(t, time.Minute)
+	w := newProgressWorker(t, r, plannedTr{r.tr})
+
+	w.Process(context.Background(), r.spool(t, "a"))
+
+	assert.Equal(t, []string{"received:done", "audio:done", "engine_a:done", "engine_b:done", "save:done"}, keysAndStates(r.st.lastSteps()))
+}
+
+func TestWorkerProgress_AFailureMarksWhereItStoppedAndLeavesTheRestPending(t *testing.T) {
+	tests := []struct {
+		name  string
+		setup func(*wrig)
+		want  []string
+	}{
+		{
+			"conversion fails", func(r *wrig) { r.conv.err = errors.New("moov atom not found") },
+			[]string{"received:done", "audio:failed", "transcribe:pending", "save:pending"},
+		},
+		{
+			"every engine fails", func(r *wrig) { r.tr.err = errors.New("iguana down") },
+			[]string{"received:done", "audio:done", "transcribe:failed", "save:pending"},
+		},
+	}
+	for _, tt := range tests {
+		t.Run(tt.name, func(t *testing.T) {
+			r := newWRig(t, time.Minute)
+			tt.setup(r)
+
+			r.w.Process(context.Background(), r.spool(t, "a"))
+
+			assert.Equal(t, tt.want, keysAndStates(r.st.lastSteps()))
+		})
+	}
+}
+
+func TestWorkerProgress_AShutdownIsNotRecordedAsAFailedStep(t *testing.T) {
+	r := newWRig(t, time.Minute)
+	r.tr.block = make(chan struct{})
+	ctx, cancel := context.WithCancel(context.Background())
+	done := make(chan struct{})
+	go func() { defer close(done); r.w.Process(ctx, r.spool(t, "a")) }()
+	require.Eventually(t, func() bool {
+		return strings.Contains(strings.Join(keysAndStates(r.st.lastSteps()), ","), "transcribe:running")
+	}, 2*time.Second, time.Millisecond)
+
+	cancel()
+	<-done
+
+	assert.Contains(t, keysAndStates(r.st.lastSteps()), "transcribe:running",
+		"a shutdown leaves the row for the start-up sweep; it must not pose as a failure of the step")
+}
+
+func TestWorkerProgress_TheHeartbeatBeatsWhileAnEngineCallIsStuckAndStopsWhenTheJobEnds(t *testing.T) {
+	r := newWRig(t, time.Minute)
+	r.tr.block = make(chan struct{})
+	w := newProgressWorker(t, r, r.tr, uploads.WithHeartbeat(5*time.Millisecond, time.Hour))
+	done := make(chan struct{})
+	go func() { defer close(done); w.Process(context.Background(), r.spool(t, "a")) }()
+
+	require.Eventually(t, func() bool { return r.st.beatCount() >= 3 }, 2*time.Second, 5*time.Millisecond,
+		"the page tells a slow engine from a dead service by this heartbeat, so it must not wait on the engine")
+	close(r.tr.block)
+	<-done
+	after := r.st.beatCount()
+	time.Sleep(40 * time.Millisecond)
+
+	assert.Equal(t, after, r.st.beatCount(), "no heartbeat may outlive its job")
+}
+
+func TestWorkerProgress_TheLogNamesEveryPhaseInOrderAndNeverTheTranscriptOrThePath(t *testing.T) {
+	r := newWRig(t, time.Minute)
+	r.tr.result = domain.Transcription{Text: "zebrakaka secret meeting words", Endpoint: "iguana/kb-whisper", AudioSeconds: 3}
+	buf := &syncLog{}
+	w := newProgressWorker(t, r, r.tr, uploads.WithLogger(slog.New(slog.NewTextHandler(buf, nil))))
+
+	w.Process(context.Background(), r.spool(t, "a"))
+
+	out := buf.String()
+	audio := strings.Index(out, "step=audio")
+	transcribe := strings.Index(out, "step=transcribe")
+	save := strings.Index(out, "step=save")
+	require.True(t, audio >= 0 && transcribe > audio && save > transcribe, "phases must be logged in the order they ran:\n%s", out)
+	assert.Contains(t, out, "upload=a", "every line says which upload it is about")
+	assert.NotContains(t, out, "zebrakaka", "no transcript text in the log")
+	assert.NotContains(t, out, r.tmp, "no spool path, which carries the upload's identity, in the log")
+}