From 52417e6b47d89fcdfd82321dc399fb0c0f0cf5ad Mon Sep 17 00:00:00 2001 From: Mathias Date: Thu, 1 Oct 2026 16:47:22 +0200 Subject: [PATCH] test(uploads): in-process transcription worker (red) Success path deletes both the upload and the WAV; failures are typed and still delete the audio; a shutdown mid-job leaves the row for the start-up sweep (it is not a transcription failure); a hung endpoint is failed by the job timeout, never an eternal spinner; strictly one job at a time (two long jobs on iguana were never measured); stopping deletes queued audio and refuses new jobs. Red: uploads.Worker is undefined. Co-Authored-By: Claude Sonnet 5.5 --- internal/uploads/worker_test.go | 275 ++++++++++++++++++++++++++++++++ 1 file changed, 275 insertions(+) create mode 100644 internal/uploads/worker_test.go diff --git a/internal/uploads/worker_test.go b/internal/uploads/worker_test.go new file mode 100644 index 0000000..27e2c11 --- /dev/null +++ b/internal/uploads/worker_test.go @@ -0,0 +1,275 @@ +package uploads_test + +import ( + "context" + "errors" + "os" + "path/filepath" + "sync" + "sync/atomic" + "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/uploads" +) + +type wStore struct { + mu sync.Mutex + started []string + completed map[string]domain.Transcription + failed map[string]domain.UploadErrorKind + startErr error +} + +func newWStore() *wStore { + return &wStore{completed: map[string]domain.Transcription{}, failed: map[string]domain.UploadErrorKind{}} +} + +func (s *wStore) StartTranscribing(_ context.Context, _, id string) error { + s.mu.Lock() + defer s.mu.Unlock() + s.started = append(s.started, id) + return s.startErr +} + +func (s *wStore) CompleteUpload(_ context.Context, _, id string, t domain.Transcription) error { + s.mu.Lock() + defer s.mu.Unlock() + s.completed[id] = t + return nil +} + +func (s *wStore) FailUpload(_ context.Context, _, id string, k domain.UploadErrorKind) error { + s.mu.Lock() + defer s.mu.Unlock() + s.failed[id] = k + return nil +} + +func (s *wStore) snapshot() (map[string]domain.Transcription, map[string]domain.UploadErrorKind) { + s.mu.Lock() + defer s.mu.Unlock() + c := map[string]domain.Transcription{} + f := map[string]domain.UploadErrorKind{} + for k, v := range s.completed { + c[k] = v + } + for k, v := range s.failed { + f[k] = v + } + return c, f +} + +type fakeConverter struct { + err error + srcs []string + dsts []string +} + +func (c *fakeConverter) ToWAV(_ context.Context, src, dst string) error { + c.srcs = append(c.srcs, src) + c.dsts = append(c.dsts, dst) + if c.err != nil { + return c.err + } + return os.WriteFile(dst, []byte("wav"), 0o600) +} + +type fakeTranscriber struct { + result domain.Transcription + err error + paths []string + // block, when non-nil, holds every call until it is closed or ctx ends. + block chan struct{} + running atomic.Int32 + maxSeen atomic.Int32 +} + +func (f *fakeTranscriber) Transcribe(ctx context.Context, wav string) (domain.Transcription, error) { + f.paths = append(f.paths, wav) + n := f.running.Add(1) + defer f.running.Add(-1) + for { + m := f.maxSeen.Load() + if n <= m || f.maxSeen.CompareAndSwap(m, n) { + break + } + } + if f.block != nil { + select { + case <-f.block: + case <-ctx.Done(): + return domain.Transcription{}, ctx.Err() + } + } + return f.result, f.err +} + +type wrig struct { + w *uploads.Worker + st *wStore + conv *fakeConverter + tr *fakeTranscriber + tmp string +} + +func newWRig(t *testing.T, timeout time.Duration) *wrig { + t.Helper() + r := &wrig{ + st: newWStore(), + conv: &fakeConverter{}, + tr: &fakeTranscriber{result: domain.Transcription{Text: "synthetic text", Endpoint: "iguana/kb-whisper", AudioSeconds: 3}}, + tmp: t.TempDir(), + } + r.w = uploads.NewWorker(r.st, r.conv, r.tr, r.tmp, timeout) + return r +} + +// spool writes a fake uploaded file (what Intake would hand over) and returns a Job. +func (r *wrig) spool(t *testing.T, id string) uploads.Job { + t.Helper() + p := filepath.Join(r.tmp, "upload-"+id) + require.NoError(t, os.WriteFile(p, []byte("audio"), 0o600)) + return uploads.Job{UserID: "u1", UploadID: id, Path: p} +} + +func (r *wrig) leftovers(t *testing.T) []string { + t.Helper() + ents, err := os.ReadDir(r.tmp) + require.NoError(t, err) + var n []string + for _, e := range ents { + n = append(n, e.Name()) + } + return n +} + +func TestWorkerProcess_SuccessConvertsTranscribesStoresAndDeletesTheAudio(t *testing.T) { + r := newWRig(t, time.Minute) + job := r.spool(t, "a") + + r.w.Process(context.Background(), job) + + c, f := r.st.snapshot() + assert.Equal(t, []string{"a"}, r.st.started) + assert.Equal(t, "synthetic text", c["a"].Text) + assert.Equal(t, "iguana/kb-whisper", c["a"].Endpoint) + assert.Empty(t, f) + require.Len(t, r.conv.srcs, 1) + assert.Equal(t, job.Path, r.conv.srcs[0], "the converter reads the uploaded file") + assert.Equal(t, r.conv.dsts, r.tr.paths, "whisper is sent the converted WAV, not the original") + assert.Equal(t, r.tmp, filepath.Dir(r.conv.dsts[0])) + assert.Empty(t, r.leftovers(t), "the audio is never retained: both the upload and the WAV are deleted") +} + +func TestWorkerProcess_FailuresAreTypedAndStillDeleteTheAudio(t *testing.T) { + tests := []struct { + name string + setup func(*wrig) + }{ + {"conversion fails", func(r *wrig) { r.conv.err = errors.New("moov atom not found") }}, + {"every transcription endpoint fails", func(r *wrig) { r.tr.err = errors.New("iguana down; berget down") }}, + } + for _, tt := range tests { + t.Run(tt.name, func(t *testing.T) { + r := newWRig(t, time.Minute) + tt.setup(r) + job := r.spool(t, "a") + + r.w.Process(context.Background(), job) + + c, f := r.st.snapshot() + assert.Empty(t, c) + assert.Equal(t, domain.UploadErrTranscriptionFailed, f["a"]) + assert.Empty(t, r.leftovers(t)) + }) + } + t.Run("conversion failure never reaches whisper", func(t *testing.T) { + r := newWRig(t, time.Minute) + r.conv.err = errors.New("bad") + + r.w.Process(context.Background(), r.spool(t, "a")) + + assert.Empty(t, r.tr.paths) + }) +} + +func TestWorkerProcess_ShutdownMidJobLeavesTheRowForTheStartupSweep(t *testing.T) { + r := newWRig(t, time.Minute) + r.tr.block = make(chan struct{}) + ctx, cancel := context.WithCancel(context.Background()) + done := make(chan struct{}) + job := r.spool(t, "a") + + go func() { r.w.Process(ctx, job); close(done) }() + require.Eventually(t, func() bool { return r.tr.running.Load() == 1 }, 2*time.Second, 5*time.Millisecond) + cancel() + <-done + + _, f := r.st.snapshot() + assert.Empty(t, f, "a shutdown is not a transcription failure: the row stays for FailInterruptedUploads to mark interrupted") + assert.Empty(t, r.leftovers(t), "but the audio is still deleted") +} + +func TestWorkerProcess_AJobThatOutlivesItsTimeoutIsFailedNotStuck(t *testing.T) { + r := newWRig(t, 50*time.Millisecond) + r.tr.block = make(chan struct{}) // never released + + r.w.Process(context.Background(), r.spool(t, "a")) + + _, f := r.st.snapshot() + assert.Equal(t, domain.UploadErrTranscriptionFailed, f["a"], "a hung endpoint must end in failed, not an eternal spinner (ADR-025)") + assert.Empty(t, r.leftovers(t)) +} + +func TestWorkerProcess_StartFailureDropsTheJobAndDeletesTheAudio(t *testing.T) { + r := newWRig(t, time.Minute) + r.st.startErr = errors.New("db down") + + r.w.Process(context.Background(), r.spool(t, "a")) + + assert.Empty(t, r.tr.paths, "never transcribe a job whose row could not be moved to transcribing") + _, f := r.st.snapshot() + assert.Contains(t, f, "a") + assert.Empty(t, r.leftovers(t)) +} + +func TestWorkerRun_ProcessesOneJobAtATime(t *testing.T) { + r := newWRig(t, time.Minute) + r.tr.block = make(chan struct{}) + ctx, cancel := context.WithCancel(context.Background()) + defer cancel() + go r.w.Run(ctx) + + for _, id := range []string{"a", "b", "c"} { + require.NoError(t, r.w.Submit(r.spool(t, id))) + } + require.Eventually(t, func() bool { return r.tr.running.Load() == 1 }, 2*time.Second, 5*time.Millisecond) + time.Sleep(50 * time.Millisecond) // give a (wrong) second job time to start + assert.Equal(t, int32(1), r.tr.running.Load(), "two simultaneous long jobs on iguana were never measured: concurrency is 1") + + close(r.tr.block) + require.Eventually(t, func() bool { c, _ := r.st.snapshot(); return len(c) == 3 }, 3*time.Second, 5*time.Millisecond) + assert.Equal(t, int32(1), r.tr.maxSeen.Load()) +} + +func TestWorkerRun_StoppingDeletesQueuedAudioAndRefusesNewJobs(t *testing.T) { + r := newWRig(t, time.Minute) + r.tr.block = make(chan struct{}) + ctx, cancel := context.WithCancel(context.Background()) + stopped := make(chan struct{}) + go func() { r.w.Run(ctx); close(stopped) }() + require.NoError(t, r.w.Submit(r.spool(t, "a"))) + require.Eventually(t, func() bool { return r.tr.running.Load() == 1 }, 2*time.Second, 5*time.Millisecond) + require.NoError(t, r.w.Submit(r.spool(t, "b"))) // queued behind the running job + + cancel() + <-stopped + + assert.Empty(t, r.leftovers(t), "audio of queued jobs is deleted on stop: the rows are swept to interrupted at next start") + assert.ErrorIs(t, r.w.Submit(r.spool(t, "c")), uploads.ErrStopped) +}