diff --git a/docs/roadmap/implementation.md b/docs/roadmap/implementation.md index 651d515..7b7a2e3 100644 --- a/docs/roadmap/implementation.md +++ b/docs/roadmap/implementation.md @@ -1,6 +1,6 @@ # Implementation Plan: Speaker Checkpoints and Replayable Uploads -Status: Implementation in progress; Stages 1-5 complete. +Status: Implementation in progress; Stages 1-6 complete. ## Scope and Instructions @@ -337,7 +337,20 @@ ambiguity, and all existing HTTP behavior tests still pass. ## Stage 6: Verify Replay Through the HTTP Adapter Boundary -Status: Pending. +Status: Complete. + +Completion marker (2026-09-22): Added a deterministic transport regression +that abandons a partially consumed original request, obtains and parses a fresh +multipart replay from `GetBody`, and installs the returned transcript within one +application attempt. Covered replay open/read failures, early responses, +parent cancellation during replay, and disposal of every opened source, while +proving an abandoned original does not poison a valid replay. Added a real HTTP +retry-exhaustion integration that durably retains an earlier speaker and resumes +only the uncommitted speakers. This controlled boundary verifies Narratio's +replay contract, not recovery from every real proxy protocol failure. Verified +with `go test ./internal/adapters/whisperx ./internal/app ./internal/stage`, +`go test -race ./internal/adapters/whisperx ./internal/app ./internal/stage`, +ten repeated race-enabled adapter runs, and `git diff --check`. Preparation: Existing injectable HTTPClient/RoundTripper seam and local-server tests in `internal/adapters/whisperx/http_test.go`; app WhisperX wiring tests. diff --git a/internal/adapters/whisperx/http_replay_test.go b/internal/adapters/whisperx/http_replay_test.go new file mode 100644 index 0000000..13d7b26 --- /dev/null +++ b/internal/adapters/whisperx/http_replay_test.go @@ -0,0 +1,261 @@ +package whisperx + +import ( + "context" + "errors" + "io" + "net/http" + "os" + "path/filepath" + "strings" + "sync/atomic" + "testing" + "time" +) + +func TestHTTPClientTransportReplaysAbandonedMultipartBody(t *testing.T) { + var opens atomic.Int32 + var closes atomic.Int32 + transport := roundTripFunc(func(request *http.Request) (*http.Response, error) { + partial := make([]byte, 17) + if _, err := io.ReadFull(request.Body, partial); err != nil { + return nil, err + } + if err := request.Body.Close(); err != nil { + return nil, err + } + if request.GetBody == nil { + return nil, errors.New("request GetBody is nil") + } + replay, err := request.GetBody() + if err != nil { + return nil, err + } + replayedBytes, err := io.ReadAll(replay) + if err != nil { + _ = replay.Close() + return nil, err + } + if err := replay.Close(); err != nil { + return nil, err + } + parts := parseMultipartParts(t, request.Header.Get("Content-Type"), replayedBytes) + if len(parts) != 2 || parts[0].name != "file" || parts[0].filename != "audio.flac" || string(parts[0].data) != "audio-data" { + t.Fatalf("replayed file part = %#v, want exact audio.flac payload", parts) + } + if parts[1].name != "language" || string(parts[1].data) != "en" { + t.Fatalf("replayed language part = %#v, want language=en", parts[1]) + } + return whisperXJSONResponse(request), nil + }) + client := newReplayTestHTTPClient(t, transport) + client.openAudio = func(string) (io.ReadCloser, error) { + opens.Add(1) + return &countedReadCloser{Reader: strings.NewReader("audio-data"), closes: &closes}, nil + } + outPath := filepath.Join(t.TempDir(), "raw.json") + + result, err := client.Transcribe(context.Background(), TranscribeRequest{ + AudioPath: "audio.flac", + OutputRawTranscriptPath: outPath, + }) + if err != nil { + t.Fatalf("Transcribe() error = %v", err) + } + if result.Attempts != 1 { + t.Fatalf("application attempts = %d, want 1", result.Attempts) + } + verifyJSONFile(t, outPath) + if got := opens.Load(); got != 2 { + t.Fatalf("audio opens = %d, want original and replay", got) + } + if got := closes.Load(); got != 2 { + t.Fatalf("audio closes = %d, want all body handles disposed", got) + } +} + +func TestHTTPClientTransportReturnsReplayOpenFailure(t *testing.T) { + replayErr := errors.New("replay open failed") + var opens atomic.Int32 + var closes atomic.Int32 + transport := roundTripFunc(func(request *http.Request) (*http.Response, error) { + if _, err := io.ReadFull(request.Body, make([]byte, 1)); err != nil { + return nil, err + } + _ = request.Body.Close() + _, err := request.GetBody() + return nil, err + }) + client := newReplayTestHTTPClient(t, transport) + client.openAudio = func(string) (io.ReadCloser, error) { + if opens.Add(1) == 2 { + return nil, replayErr + } + return &countedReadCloser{Reader: strings.NewReader("audio-data"), closes: &closes}, nil + } + outPath := filepath.Join(t.TempDir(), "raw.json") + + result, err := client.Transcribe(context.Background(), TranscribeRequest{ + AudioPath: "audio.flac", + OutputRawTranscriptPath: outPath, + }) + if !errors.Is(err, replayErr) { + t.Fatalf("Transcribe() error = %v, want replay open failure", err) + } + if result.Attempts != 1 { + t.Fatalf("application attempts = %d, want 1", result.Attempts) + } + if got := closes.Load(); got != 1 { + t.Fatalf("audio closes = %d, want original disposed", got) + } + if _, statErr := os.Stat(outPath); !errors.Is(statErr, os.ErrNotExist) { + t.Fatalf("output should not exist, stat error = %v", statErr) + } +} + +func TestHTTPClientTransportCannotSwallowReplayReadFailure(t *testing.T) { + replayErr := errors.New("replay read failed") + var opens atomic.Int32 + transport := roundTripFunc(func(request *http.Request) (*http.Response, error) { + if _, err := io.ReadFull(request.Body, make([]byte, 1)); err != nil { + return nil, err + } + _ = request.Body.Close() + replay, err := request.GetBody() + if err != nil { + return nil, err + } + _, _ = io.ReadAll(replay) + _ = replay.Close() + return whisperXJSONResponse(request), nil + }) + client := newReplayTestHTTPClient(t, transport) + client.openAudio = func(string) (io.ReadCloser, error) { + if opens.Add(1) == 2 { + return &failingReadCloser{first: []byte("partial"), err: replayErr}, nil + } + return io.NopCloser(strings.NewReader("audio-data")), nil + } + outPath := filepath.Join(t.TempDir(), "raw.json") + + _, err := client.Transcribe(context.Background(), TranscribeRequest{ + AudioPath: "audio.flac", + OutputRawTranscriptPath: outPath, + }) + if !errors.Is(err, replayErr) { + t.Fatalf("Transcribe() error = %v, want replay read failure", err) + } + if _, statErr := os.Stat(outPath); !errors.Is(statErr, os.ErrNotExist) { + t.Fatalf("output should not exist, stat error = %v", statErr) + } +} + +func TestHTTPClientTransportEarlyResponseDisposesOriginalBody(t *testing.T) { + var closes atomic.Int32 + client := newReplayTestHTTPClient(t, roundTripFunc(func(request *http.Request) (*http.Response, error) { + return whisperXJSONResponse(request), nil + })) + client.openAudio = func(string) (io.ReadCloser, error) { + return &countedReadCloser{Reader: strings.NewReader("audio-data"), closes: &closes}, nil + } + + _, err := client.Transcribe(context.Background(), TranscribeRequest{ + AudioPath: "audio.flac", + OutputRawTranscriptPath: filepath.Join(t.TempDir(), "raw.json"), + }) + if err != nil { + t.Fatalf("Transcribe() error = %v", err) + } + if got := closes.Load(); got != 1 { + t.Fatalf("audio closes = %d, want early-response body disposed", got) + } +} + +func TestHTTPClientParentCancellationDuringReplayClosesEveryBody(t *testing.T) { + var opens atomic.Int32 + var originalCloses atomic.Int32 + replaySource := newSignalingBlockedReadCloser() + transport := roundTripFunc(func(request *http.Request) (*http.Response, error) { + if _, err := io.ReadFull(request.Body, make([]byte, 1)); err != nil { + return nil, err + } + _ = request.Body.Close() + replay, err := request.GetBody() + if err != nil { + return nil, err + } + _, _ = io.ReadAll(replay) + _ = replay.Close() + return nil, request.Context().Err() + }) + client := newReplayTestHTTPClient(t, transport) + client.openAudio = func(string) (io.ReadCloser, error) { + if opens.Add(1) == 1 { + return &countedReadCloser{Reader: strings.NewReader("audio-data"), closes: &originalCloses}, nil + } + return replaySource, nil + } + ctx, cancel := context.WithCancel(context.Background()) + done := make(chan error, 1) + go func() { + _, err := client.Transcribe(ctx, TranscribeRequest{ + AudioPath: "audio.flac", + OutputRawTranscriptPath: filepath.Join(t.TempDir(), "raw.json"), + }) + done <- err + }() + + select { + case <-replaySource.blocked: + cancel() + case <-time.After(time.Second): + cancel() + t.Fatal("replay source did not block") + } + select { + case err := <-done: + if !errors.Is(err, context.Canceled) { + t.Fatalf("Transcribe() error = %v, want context cancellation", err) + } + case <-time.After(time.Second): + t.Fatal("Transcribe() did not return after replay cancellation") + } + if got := originalCloses.Load(); got != 1 { + t.Fatalf("original audio closes = %d, want 1", got) + } + select { + case <-replaySource.closed: + case <-time.After(time.Second): + t.Fatal("replay audio source was not closed") + } +} + +type roundTripFunc func(*http.Request) (*http.Response, error) + +func (roundTrip roundTripFunc) RoundTrip(request *http.Request) (*http.Response, error) { + return roundTrip(request) +} + +func newReplayTestHTTPClient(t *testing.T, transport http.RoundTripper) *HTTPClient { + t.Helper() + client, err := NewHTTPClient(HTTPClientConfig{ + TranscribeURL: "http://whisperx.test/transcribe", + Language: "en", + Timeout: time.Second, + HTTPClient: &http.Client{Transport: transport}, + }) + if err != nil { + t.Fatalf("NewHTTPClient() error = %v", err) + } + return client +} + +func whisperXJSONResponse(request *http.Request) *http.Response { + return &http.Response{ + StatusCode: http.StatusOK, + Status: "200 OK", + Header: make(http.Header), + Body: io.NopCloser(strings.NewReader(`{"source":"transport-replay","segments":[]}`)), + Request: request, + } +} diff --git a/internal/app/whisperx_wiring_test.go b/internal/app/whisperx_wiring_test.go index 89dbcf7..ccb10e3 100644 --- a/internal/app/whisperx_wiring_test.go +++ b/internal/app/whisperx_wiring_test.go @@ -7,9 +7,12 @@ import ( "net/http/httptest" "os" "path/filepath" + "slices" "strings" + "sync" "sync/atomic" "testing" + "time" "gitea.maximumdirect.net/eric/narratio/internal/adapters/whisperx" "gitea.maximumdirect.net/eric/narratio/internal/config" @@ -118,6 +121,86 @@ func TestExecuteStagesDefaultWiringWhisperXFailureMarksManifestFailed(t *testing } } +func TestExhaustedWhisperXFailurePreservesEarlierSpeakerCheckpoint(t *testing.T) { + var successful atomic.Bool + var requestMu sync.Mutex + var requested []string + srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + file, header, err := r.FormFile("file") + if err != nil { + t.Errorf("FormFile(file) error = %v", err) + http.Error(w, "invalid multipart", http.StatusBadRequest) + return + } + _, readErr := io.Copy(io.Discard, file) + _ = file.Close() + if readErr != nil { + t.Errorf("read uploaded file: %v", readErr) + return + } + speaker := strings.TrimSuffix(header.Filename, filepath.Ext(header.Filename)) + requestMu.Lock() + requested = append(requested, speaker) + requestMu.Unlock() + if speaker == "b" && !successful.Load() { + http.Error(w, "temporary", http.StatusInternalServerError) + return + } + _, _ = w.Write([]byte(`{"segments":[]}`)) + })) + defer srv.Close() + + cfg, store := setupTranscribeRecoveryFixture(t, []string{"a", "b", "c"}) + setWhisperXConfig(cfg, srv.URL, 1) + concurrency := 1 + cfg.Pipeline.WhisperX.Concurrency = &concurrency + httpClient, err := whisperx.NewHTTPClient(whisperx.HTTPClientConfig{ + TranscribeURL: srv.URL, + Language: "en", + Timeout: 2 * time.Second, + Retries: 1, + RetryDelay: time.Millisecond, + }) + if err != nil { + t.Fatalf("NewHTTPClient() error = %v", err) + } + transcribe := mustSelectStage(t, "transcribe") + + if _, err := executeStages(context.Background(), cfg, []stage.Stage{transcribe}, RunOptions{ + Env: &Env{ManifestStore: store, WhisperX: httpClient}, + }); err == nil { + t.Fatal("executeStages() error = nil, want exhausted HTTP failure") + } + failed := loadTranscribeRecoveryManifest(t, store, cfg) + if got := transcribeSpeakerKeys(failed); !slices.Equal(got, []string{"a"}) { + t.Fatalf("checkpointed speakers = %#v, want a", got) + } + requestMu.Lock() + firstRequests := append([]string(nil), requested...) + requested = nil + requestMu.Unlock() + if !slices.Equal(firstRequests, []string{"a", "b", "b"}) { + t.Fatalf("initial HTTP speakers = %#v, want a then two attempts for b", firstRequests) + } + + successful.Store(true) + if _, err := executeStages(context.Background(), cfg, []stage.Stage{transcribe}, RunOptions{ + Env: &Env{ManifestStore: store, WhisperX: httpClient}, + }); err != nil { + t.Fatalf("retry executeStages() error = %v", err) + } + requestMu.Lock() + retryRequests := append([]string(nil), requested...) + requestMu.Unlock() + if !slices.Equal(retryRequests, []string{"b", "c"}) { + t.Fatalf("retry HTTP speakers = %#v, want only b and c", retryRequests) + } + completed := loadTranscribeRecoveryManifest(t, store, cfg) + if got := transcribeSpeakerKeys(completed); !slices.Equal(got, []string{"a", "b", "c"}) { + t.Fatalf("completed speakers = %#v, want all speakers", got) + } +} + func TestExecuteStagesExplicitWhisperXInjectionOverridesDefaultWiring(t *testing.T) { cfg := testConfig(t) setWhisperXConfig(cfg, "https://127.0.0.1:1/transcribe", 0)