From 6813340f51b6fbdb917cf8c41fad4d732b0cecec Mon Sep 17 00:00:00 2001 From: Eric Rakestraw Date: Tue, 1 Sep 2026 11:51:36 +0000 Subject: [PATCH] Strengthen transcription concurrency tests --- internal/stage/transcribe_test.go | 146 +++++++++++++++++++++++------- 1 file changed, 111 insertions(+), 35 deletions(-) diff --git a/internal/stage/transcribe_test.go b/internal/stage/transcribe_test.go index 1ba6d01..3e8b29c 100644 --- a/internal/stage/transcribe_test.go +++ b/internal/stage/transcribe_test.go @@ -4,9 +4,10 @@ import ( "context" "encoding/json" "errors" + "fmt" "os" "path/filepath" - "sort" + "slices" "strings" "sync/atomic" "testing" @@ -19,7 +20,7 @@ import ( ) func TestTranscribeStageTranscribesPreparedAudio(t *testing.T) { - env, m := setupTranscribeEnv(t, []string{"alice.flac", "bob.flac"}) + env, m := setupTranscribeEnv(t, []string{"bob.flac", "alice.flac"}) env.WhisperX = &whisperx.FakeClient{ Result: whisperx.TranscribeResult{ Attempts: 2, @@ -51,13 +52,11 @@ func TestTranscribeStageTranscribesPreparedAudio(t *testing.T) { gotPaths = append(gotPaths, out.AbsolutePath) verifyJSONFile(t, out.AbsolutePath) } - sort.Strings(gotPaths) wantPaths := []string{ filepath.Join(sessionPathsForEnv(env, m.SessionID).TranscriptsRawDir, "alice.json"), filepath.Join(sessionPathsForEnv(env, m.SessionID).TranscriptsRawDir, "bob.json"), } - sort.Strings(wantPaths) - if strings.Join(gotPaths, "|") != strings.Join(wantPaths, "|") { + if !slices.Equal(gotPaths, wantPaths) { t.Fatalf("output paths = %#v, want %#v", gotPaths, wantPaths) } @@ -87,15 +86,25 @@ func TestTranscribeStageTranscribesPreparedAudio(t *testing.T) { if len(perFile) != 2 { t.Fatalf("metadata per_file len = %d, want 2", len(perFile)) } + outputPaths, ok := meta["output_paths"].([]string) + if !ok || !slices.Equal(outputPaths, wantPaths) { + t.Fatalf("metadata output_paths = %#v, want %#v", meta["output_paths"], wantPaths) + } + runOutputPaths, ok := meta["run_output_paths"].([]string) + if !ok || len(runOutputPaths) != 2 || filepath.Base(runOutputPaths[0]) != "alice.json" || filepath.Base(runOutputPaths[1]) != "bob.json" { + t.Fatalf("metadata run_output_paths = %#v, want stable alice/bob order", meta["run_output_paths"]) + } } func TestTranscribeStageConcurrencyBounded(t *testing.T) { env, m := setupTranscribeEnv(t, []string{"a.flac", "b.flac", "c.flac", "d.flac", "e.flac"}) var inFlight int32 var maxInFlight int32 + started := make(chan struct{}, 2) + release := make(chan struct{}) env.WhisperX = &whisperx.FakeClient{ - TranscribeFn: func(_ context.Context, req whisperx.TranscribeRequest) (whisperx.TranscribeResult, error) { + TranscribeFn: func(ctx context.Context, req whisperx.TranscribeRequest) (whisperx.TranscribeResult, error) { n := atomic.AddInt32(&inFlight, 1) for { cur := atomic.LoadInt32(&maxInFlight) @@ -104,8 +113,15 @@ func TestTranscribeStageConcurrencyBounded(t *testing.T) { } } defer atomic.AddInt32(&inFlight, -1) - - time.Sleep(25 * time.Millisecond) + select { + case started <- struct{}{}: + default: + } + select { + case <-release: + case <-ctx.Done(): + return whisperx.TranscribeResult{}, ctx.Err() + } if err := writeJSONFile(req.OutputRawTranscriptPath, map[string]any{"speaker": req.SpeakerID}); err != nil { return whisperx.TranscribeResult{}, err } @@ -122,42 +138,81 @@ func TestTranscribeStageConcurrencyBounded(t *testing.T) { t.Fatalf("prepare.Run() error = %v", err) } - if _, err := (transcribeStage{}).Run(context.Background(), env, m); err != nil { - t.Fatalf("transcribe.Run() error = %v", err) + ctx, cancel := context.WithCancel(context.Background()) + defer cancel() + done := make(chan error, 1) + go func() { + _, err := (transcribeStage{}).Run(ctx, env, m) + done <- err + }() + for range 2 { + select { + case <-started: + case <-time.After(time.Second): + t.Fatal("configured concurrent transcription work did not start") + } } - - if got := atomic.LoadInt32(&maxInFlight); got > 2 { - t.Fatalf("max in-flight = %d, want <= 2", got) + if got := atomic.LoadInt32(&maxInFlight); got != 2 { + t.Fatalf("max in-flight = %d, want configured limit 2", got) + } + close(release) + select { + case err := <-done: + if err != nil { + t.Fatalf("transcribe.Run() error = %v", err) + } + case <-time.After(2 * time.Second): + t.Fatal("transcribe.Run() did not complete after releasing work") } } -func TestTranscribeStageFailsWhenOneFileFails(t *testing.T) { - env, m := setupTranscribeEnv(t, []string{"alice.flac", "bob.flac"}) +func TestTranscribeStageFailureCancelsRemainingWork(t *testing.T) { + env, m := setupTranscribeEnv(t, []string{"alice.flac", "bob.flac", "cara.flac", "dan.flac"}) + aliceStarted := make(chan struct{}) + upstreamErr := errors.New("upstream failed") env.WhisperX = &whisperx.FakeClient{ - TranscribeFn: func(_ context.Context, req whisperx.TranscribeRequest) (whisperx.TranscribeResult, error) { - if req.SpeakerID == "bob" { - return whisperx.TranscribeResult{}, errors.New("upstream failed") + TranscribeFn: func(ctx context.Context, req whisperx.TranscribeRequest) (whisperx.TranscribeResult, error) { + switch req.SpeakerID { + case "alice": + close(aliceStarted) + <-ctx.Done() + return whisperx.TranscribeResult{}, ctx.Err() + case "bob": + <-aliceStarted + return whisperx.TranscribeResult{}, upstreamErr + default: + return whisperx.TranscribeResult{}, fmt.Errorf("unexpected adapter call for %s", req.SpeakerID) } - if err := writeJSONFile(req.OutputRawTranscriptPath, map[string]any{"speaker": req.SpeakerID}); err != nil { - return whisperx.TranscribeResult{}, err - } - return whisperx.TranscribeResult{ - OutputRawTranscriptPath: req.OutputRawTranscriptPath, - Attempts: 1, - HTTPStatus: 200, - }, nil }, } if _, err := (prepareStage{}).Run(context.Background(), env, m); err != nil { t.Fatalf("prepare.Run() error = %v", err) } - _, err := (transcribeStage{}).Run(context.Background(), env, m) - if err == nil { - t.Fatal("expected transcribe error, got nil") + ctx, cancel := context.WithCancel(context.Background()) + defer cancel() + done := make(chan error, 1) + go func() { + _, err := (transcribeStage{}).Run(ctx, env, m) + done <- err + }() + var err error + select { + case err = <-done: + case <-time.After(2 * time.Second): + t.Fatal("transcribe.Run() did not return after a speaker failure") } - if !strings.Contains(err.Error(), "bob") { - t.Fatalf("error = %q, want speaker context", err.Error()) + if !errors.Is(err, upstreamErr) || !strings.Contains(err.Error(), "bob") { + t.Fatalf("error = %v, want wrapped upstream failure with bob speaker context", err) + } + requests := env.WhisperX.(*whisperx.FakeClient).RequestsSnapshot() + requestedSpeakers := make([]string, 0, len(requests)) + for _, req := range requests { + requestedSpeakers = append(requestedSpeakers, req.SpeakerID) + } + slices.Sort(requestedSpeakers) + if !slices.Equal(requestedSpeakers, []string{"alice", "bob"}) { + t.Fatalf("adapter speakers = %#v, want only in-flight alice and failing bob", requestedSpeakers) } } @@ -186,8 +241,25 @@ func TestTranscribeStageInvalidJSONFails(t *testing.T) { if err == nil { t.Fatal("expected transcribe error, got nil") } - if !strings.Contains(strings.ToLower(err.Error()), "json") { - t.Fatalf("error = %q, want json validation context", err.Error()) + if !strings.Contains(err.Error(), "alice") || !strings.Contains(strings.ToLower(err.Error()), "json") { + t.Fatalf("error = %q, want speaker-qualified json validation context", err.Error()) + } +} + +func TestTranscribeStageRejectsAdapterOutputPathWithSpeakerContext(t *testing.T) { + env, m := setupTranscribeEnv(t, []string{"alice.flac"}) + env.WhisperX = &whisperx.FakeClient{ + TranscribeFn: func(_ context.Context, req whisperx.TranscribeRequest) (whisperx.TranscribeResult, error) { + return whisperx.TranscribeResult{OutputRawTranscriptPath: req.OutputRawTranscriptPath + ".other"}, nil + }, + } + if _, err := (prepareStage{}).Run(context.Background(), env, m); err != nil { + t.Fatalf("prepare.Run() error = %v", err) + } + + _, err := (transcribeStage{}).Run(context.Background(), env, m) + if err == nil || !strings.Contains(err.Error(), "alice") || !strings.Contains(err.Error(), "did not match requested path") { + t.Fatalf("transcribe.Run() error = %v, want speaker-qualified output-path mismatch", err) } } @@ -284,8 +356,12 @@ func TestTranscribeStageCancellationDoesNotMaterializePartialOutputs(t *testing. cancel() t.Fatal("no transcription completed before cancellation") } - if err := <-done; err == nil || !errors.Is(err, context.Canceled) { - t.Fatalf("transcribe.Run() error = %v, want context cancellation", err) + runErr := <-done + if runErr == nil || !errors.Is(runErr, context.Canceled) { + t.Fatalf("transcribe.Run() error = %v, want context cancellation", runErr) + } + if !strings.Contains(runErr.Error(), "1 of 3") { + t.Fatalf("transcribe.Run() error = %v, want completed/planned context", runErr) } canonical := filepath.Join(sessionPathsForEnv(env, m.SessionID).TranscriptsRawDir, "alice.json")