diff --git a/internal/uploads/worker_progress_test.go b/internal/uploads/worker_progress_test.go new file mode 100644 index 0000000..24611ea --- /dev/null +++ b/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") +}