tapir/internal/uploads/worker_progress_test.go

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{})
	r.st.beatDelay = 20 * time.Millisecond
	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_OnASuccessfulJobTheLogNamesEveryPhaseInOrderAndNeverTheTranscriptOrThePath(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")
	assert.Contains(t, out, "upload transcribed", "the job's own summary line")
	assert.Contains(t, out, "endpoint=iguana/kb-whisper", "which endpoint served it is a quality fact")
}

func TestWorkerProgress_AFailureLogsAKindNeverTheErrorTextOrThePath(t *testing.T) {
	tests := []struct {
		name  string
		setup func(*wrig)
		line  string
	}{
		{"conversion fails and names the spool file", func(r *wrig) {
			r.conv.err = errors.New("Error opening input file " + r.tmp + "/upload-a: Invalid data zebrakaka")
		}, "upload: convert"},
		{"every engine fails and the gateway echoed a word", func(r *wrig) {
			r.tr.err = errors.New("gateway echoed: zebrakaka secret words")
		}, "upload: transcribe"},
	}
	for _, tt := range tests {
		t.Run(tt.name, func(t *testing.T) {
			r := newWRig(t, time.Minute)
			tt.setup(r)
			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()
			assert.NotContains(t, out, "zebrakaka", "provider text or file contents in the log")
			assert.NotContains(t, out, r.tmp, "the spool path in the log")
			assert.Contains(t, out, tt.line, "the failure must be logged at all, not only scrubbed")
			assert.Contains(t, out, "kind=error")
			assert.Contains(t, out, "upload=a")
		})
	}
}

func TestWorkerProgress_AStoreFailureLogsItsSQLStateNeverTheValueItQuotes(t *testing.T) {
	// A Postgres error can quote the offending value; for a transcript row that is the
	// transcript (fourth adversarial review). The log gets the SQLSTATE.
	r := newWRig(t, time.Minute)
	r.st.completeErr = dbErr{}
	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()
	assert.NotContains(t, out, "zebrakaka")
	assert.Contains(t, out, "upload: store transcript")
	assert.Contains(t, out, "kind=sql_22P02")
}