From 9da2c1e1441c0717e847abec8792c116a46d2372 Mon Sep 17 00:00:00 2001 From: Eric Rakestraw Date: Mon, 10 Aug 2026 21:49:44 +0000 Subject: [PATCH] Stream WhisperX uploads safely --- docs/config.md | 2 +- docs/integrations/whisperx.md | 7 +- docs/roadmap/implementation.md | 2 + internal/adapters/whisperx/fake.go | 15 +- internal/adapters/whisperx/fake_test.go | 25 +- internal/adapters/whisperx/http.go | 238 +++++++++++++++--- internal/adapters/whisperx/http_test.go | 259 ++++++++++++++++++++ internal/app/whisperx_wiring_test.go | 5 +- internal/config/validate.go | 6 +- internal/config/validation_duration_test.go | 27 ++ internal/stage/placeholders_test.go | 5 +- internal/stage/transcribe_test.go | 7 +- 12 files changed, 542 insertions(+), 56 deletions(-) diff --git a/docs/config.md b/docs/config.md index 98f0b98..a4161f2 100644 --- a/docs/config.md +++ b/docs/config.md @@ -178,7 +178,7 @@ Rules: | `pipeline.publish.locks[]` | list | No | empty | | `pipeline.publish.locks[].source` | string | Yes (per lock) | must reference supported publish source | | `pipeline.publish.locks[].reason` | string | No | empty | -| `pipeline.whisperx.transcribe_url` | string | Yes | valid URL | +| `pipeline.whisperx.transcribe_url` | string | Yes | absolute `http` or `https` URL | | `pipeline.whisperx.language` | string | No | `en` | | `pipeline.whisperx.timeout` | duration | No | `30m` | | `pipeline.whisperx.retries` | int | No | `3` | diff --git a/docs/integrations/whisperx.md b/docs/integrations/whisperx.md index 038c695..3850b3a 100644 --- a/docs/integrations/whisperx.md +++ b/docs/integrations/whisperx.md @@ -17,6 +17,11 @@ Narratio sends an HTTP `POST` to the configured transcription URL using The server must return a `2xx` response whose body is valid JSON. Narratio does not currently require a more specific response schema at this boundary. +The transcription URL must be an absolute `http` or `https` URL. The audio body +is streamed through a fresh multipart writer for every attempt, so its memory +use is bounded by the transport buffer rather than by the complete audio file. +WhisperX response acquisition is capped at 10 MiB. + ## Request And Result Contract Each adapter request identifies a speaker, a readable audio file, and the @@ -40,7 +45,7 @@ transcript output. ## Validation And Failure Semantics -Client construction rejects a missing or invalid absolute transcription URL, +Client construction rejects a missing or non-HTTP(S) absolute transcription URL, a missing language, a non-positive timeout, negative retries, or a negative retry delay. A request fails before transmission when its audio or output path is missing. diff --git a/docs/roadmap/implementation.md b/docs/roadmap/implementation.md index 9834a96..1265270 100644 --- a/docs/roadmap/implementation.md +++ b/docs/roadmap/implementation.md @@ -810,6 +810,8 @@ server early close, cancellation, blocked producer, exact payloads, bounded responses, and goroutine completion. Run Whisper packages and their callers under `go test -race`; this stage unblocks the full race suite in Stage 31. +**Status:** Completed. + ## Stage 24 — Correct prepare/transcribe transition semantics **Read first:** `audit-findings.md` lines 2282–2379 (COR-017 through COR-019) diff --git a/internal/adapters/whisperx/fake.go b/internal/adapters/whisperx/fake.go index 178934e..805b21d 100644 --- a/internal/adapters/whisperx/fake.go +++ b/internal/adapters/whisperx/fake.go @@ -3,6 +3,7 @@ package whisperx import ( "context" "path/filepath" + "sync" "gitea.maximumdirect.net/eric/narratio/internal/fileops" ) @@ -32,7 +33,8 @@ func (n *NoopClient) Transcribe(ctx context.Context, req TranscribeRequest) (Tra // FakeClient captures requests and returns deterministic responses for tests. type FakeClient struct { - Requests []TranscribeRequest + requestsMu sync.RWMutex + requests []TranscribeRequest Err error Result TranscribeResult TranscribeFn func(ctx context.Context, req TranscribeRequest) (TranscribeResult, error) @@ -43,7 +45,9 @@ func (f *FakeClient) Transcribe(ctx context.Context, req TranscribeRequest) (Tra if err := ctx.Err(); err != nil { return TranscribeResult{}, err } - f.Requests = append(f.Requests, req) + f.requestsMu.Lock() + f.requests = append(f.requests, req) + f.requestsMu.Unlock() if f.TranscribeFn != nil { return f.TranscribeFn(ctx, req) } @@ -66,6 +70,13 @@ func (f *FakeClient) Transcribe(ctx context.Context, req TranscribeRequest) (Tra return res, nil } +// RequestsSnapshot returns a copy of captured requests safe for concurrent test assertions. +func (f *FakeClient) RequestsSnapshot() []TranscribeRequest { + f.requestsMu.RLock() + defer f.requestsMu.RUnlock() + return append([]TranscribeRequest(nil), f.requests...) +} + func writeMinimalJSON(path string) error { if path == "" { return nil diff --git a/internal/adapters/whisperx/fake_test.go b/internal/adapters/whisperx/fake_test.go index e7feb05..8139932 100644 --- a/internal/adapters/whisperx/fake_test.go +++ b/internal/adapters/whisperx/fake_test.go @@ -3,6 +3,7 @@ package whisperx import ( "context" "errors" + "sync" "testing" ) @@ -14,8 +15,9 @@ func TestFakeClientCapturesRequestAndReturnsPath(t *testing.T) { if err != nil { t.Fatalf("Transcribe() error = %v", err) } - if len(fake.Requests) != 1 || fake.Requests[0].SpeakerID != "alice" { - t.Fatalf("requests = %#v, want one alice request", fake.Requests) + requests := fake.RequestsSnapshot() + if len(requests) != 1 || requests[0].SpeakerID != "alice" { + t.Fatalf("requests = %#v, want one alice request", requests) } if res.OutputRawTranscriptPath != req.OutputRawTranscriptPath { t.Fatalf("output path = %q, want %q", res.OutputRawTranscriptPath, req.OutputRawTranscriptPath) @@ -29,3 +31,22 @@ func TestFakeClientError(t *testing.T) { t.Fatal("expected error, got nil") } } + +func TestFakeClientRequestsSnapshotSupportsConcurrentCalls(t *testing.T) { + fake := &FakeClient{} + const callers = 16 + var group sync.WaitGroup + group.Add(callers) + for i := 0; i < callers; i++ { + go func() { + defer group.Done() + if _, err := fake.Transcribe(context.Background(), TranscribeRequest{}); err != nil { + t.Errorf("Transcribe() error = %v", err) + } + }() + } + group.Wait() + if got := len(fake.RequestsSnapshot()); got != callers { + t.Fatalf("captured requests = %d, want %d", got, callers) + } +} diff --git a/internal/adapters/whisperx/http.go b/internal/adapters/whisperx/http.go index 4c51eb7..ecdd223 100644 --- a/internal/adapters/whisperx/http.go +++ b/internal/adapters/whisperx/http.go @@ -1,7 +1,6 @@ package whisperx import ( - "bytes" "context" "encoding/json" "errors" @@ -14,12 +13,16 @@ import ( "os" "path/filepath" "strings" + "sync" "time" "gitea.maximumdirect.net/eric/narratio/internal/fileops" ) -const defaultMaxResponseBytes int64 = 10 * 1024 * 1024 +const ( + defaultMaxWhisperXResponseBytes int64 = 10 * 1024 * 1024 + whisperXUploadBufferSize = 32 * 1024 +) // HTTPClientConfig contains parsed, deterministic WhisperX HTTP client settings. type HTTPClientConfig struct { @@ -41,6 +44,7 @@ type HTTPClient struct { retryDelay time.Duration httpClient *http.Client maxResponseBytes int64 + openAudio func(string) (io.ReadCloser, error) } // NewHTTPClientFromConfigValues builds a client from config values and parses durations once. @@ -74,11 +78,11 @@ func NewHTTPClient(cfg HTTPClientConfig) (*HTTPClient, error) { return nil, fmt.Errorf("whisperx transcribe_url is required") } u, err := url.Parse(cfg.TranscribeURL) - if err != nil || u.Scheme == "" || u.Host == "" { + if err != nil || !u.IsAbs() || u.Host == "" || !isHTTPURLScheme(u.Scheme) { if err != nil { return nil, fmt.Errorf("invalid whisperx transcribe_url %q: %w", cfg.TranscribeURL, err) } - return nil, fmt.Errorf("invalid whisperx transcribe_url %q", cfg.TranscribeURL) + return nil, fmt.Errorf("invalid whisperx transcribe_url %q: must be an absolute http or https URL", cfg.TranscribeURL) } if cfg.Timeout <= 0 { return nil, fmt.Errorf("whisperx timeout must be > 0") @@ -100,7 +104,7 @@ func NewHTTPClient(cfg HTTPClientConfig) (*HTTPClient, error) { maxBytes := cfg.MaxResponseBytes if maxBytes <= 0 { - maxBytes = defaultMaxResponseBytes + maxBytes = defaultMaxWhisperXResponseBytes } return &HTTPClient{ @@ -111,6 +115,7 @@ func NewHTTPClient(cfg HTTPClientConfig) (*HTTPClient, error) { retryDelay: cfg.RetryDelay, httpClient: client, maxResponseBytes: maxBytes, + openAudio: func(path string) (io.ReadCloser, error) { return os.Open(path) }, }, nil } @@ -185,52 +190,40 @@ func (c *HTTPClient) Transcribe(ctx context.Context, req TranscribeRequest) (Tra } func (c *HTTPClient) doTranscribeAttempt(ctx context.Context, audioPath string) (int, []byte, error) { - bodyBuf := &bytes.Buffer{} - writer := multipart.NewWriter(bodyBuf) + upload := newMultipartUpload(ctx, audioPath, c.language, c.openAudio) + defer upload.Close() - fileWriter, err := writer.CreateFormFile("file", filepath.Base(audioPath)) - if err != nil { - return 0, nil, fmt.Errorf("create multipart file field: %w", err) - } - - audioFile, err := os.Open(audioPath) - if err != nil { - return 0, nil, fmt.Errorf("open audio file %q: %w", audioPath, err) - } - if _, err := io.Copy(fileWriter, audioFile); err != nil { - _ = audioFile.Close() - return 0, nil, fmt.Errorf("copy audio file %q: %w", audioPath, err) - } - if err := audioFile.Close(); err != nil { - return 0, nil, fmt.Errorf("close audio file %q: %w", audioPath, err) - } - - if err := writer.WriteField("language", c.language); err != nil { - return 0, nil, fmt.Errorf("write language form field: %w", err) - } - if err := writer.Close(); err != nil { - return 0, nil, fmt.Errorf("close multipart writer: %w", err) - } - - req, err := http.NewRequestWithContext(ctx, http.MethodPost, c.url.String(), bodyBuf) + req, err := http.NewRequestWithContext(ctx, http.MethodPost, c.url.String(), upload) if err != nil { return 0, nil, fmt.Errorf("build whisperx request: %w", err) } - req.Header.Set("Content-Type", writer.FormDataContentType()) + req.Header.Set("Content-Type", upload.contentType) resp, err := c.httpClient.Do(req) if err != nil { + _ = upload.Close() + if producerErr := upload.Wait(); producerErr != nil { + return 0, nil, fmt.Errorf("stream whisperx request body: %w", producerErr) + } return 0, nil, fmt.Errorf("perform whisperx request: %w", err) } defer resp.Body.Close() - data, err := readBounded(resp.Body, c.maxResponseBytes) - if err != nil { - return resp.StatusCode, nil, fmt.Errorf("read whisperx response body: %w", err) + if resp.StatusCode < 200 || resp.StatusCode >= 300 { + _ = upload.Close() + if _, err := readWhisperXResponse(resp.Body, c.maxResponseBytes); err != nil { + return resp.StatusCode, nil, fmt.Errorf("read whisperx response body: %w", err) + } + return resp.StatusCode, nil, fmt.Errorf("whisperx returned status %d", resp.StatusCode) } - if resp.StatusCode < 200 || resp.StatusCode >= 300 { - return resp.StatusCode, nil, fmt.Errorf("whisperx returned status %d", resp.StatusCode) + if err := upload.Wait(); err != nil { + return resp.StatusCode, nil, fmt.Errorf("stream whisperx request body: %w", err) + } + + data, err := readWhisperXResponse(resp.Body, c.maxResponseBytes) + if err != nil { + return resp.StatusCode, nil, fmt.Errorf("read whisperx response body: %w", err) } return resp.StatusCode, data, nil } @@ -266,18 +259,183 @@ func (c *HTTPClient) shouldRetry(parent context.Context, err error, status int) return false } -func readBounded(r io.Reader, maxBytes int64) ([]byte, error) { +func readWhisperXResponse(r io.Reader, maxBytes int64) ([]byte, error) { limited := io.LimitReader(r, maxBytes+1) data, err := io.ReadAll(limited) if err != nil { return nil, err } if int64(len(data)) > maxBytes { - return nil, fmt.Errorf("response exceeds max size %d bytes", maxBytes) + return nil, fmt.Errorf("whisperx response exceeds configured limit of %d bytes", maxBytes) } return data, nil } +func isHTTPURLScheme(scheme string) bool { + switch strings.ToLower(scheme) { + case "http", "https": + return true + default: + return false + } +} + +type multipartUpload struct { + reader *io.PipeReader + writer *io.PipeWriter + contentType string + done chan struct{} + + mu sync.Mutex + audio io.Closer + err error + aborted bool +} + +func newMultipartUpload(ctx context.Context, audioPath, language string, openAudio func(string) (io.ReadCloser, error)) *multipartUpload { + reader, writer := io.Pipe() + multipartWriter := multipart.NewWriter(writer) + upload := &multipartUpload{ + reader: reader, + writer: writer, + contentType: multipartWriter.FormDataContentType(), + done: make(chan struct{}), + } + + go func() { + err := upload.write(ctx, multipartWriter, audioPath, language, openAudio) + if err != nil { + _ = writer.CloseWithError(err) + } else { + _ = writer.Close() + } + upload.mu.Lock() + upload.err = err + upload.audio = nil + upload.mu.Unlock() + close(upload.done) + }() + + go func() { + select { + case <-ctx.Done(): + upload.abort() + case <-upload.done: + } + }() + + return upload +} + +func (u *multipartUpload) Read(p []byte) (int, error) { + return u.reader.Read(p) +} + +func (u *multipartUpload) Close() error { + u.abort() + <-u.done + return nil +} + +func (u *multipartUpload) Wait() error { + <-u.done + u.mu.Lock() + defer u.mu.Unlock() + return u.err +} + +func (u *multipartUpload) write(ctx context.Context, writer *multipart.Writer, audioPath, language string, openAudio func(string) (io.ReadCloser, error)) error { + fileWriter, err := writer.CreateFormFile("file", filepath.Base(audioPath)) + if err != nil { + return u.producerError(ctx, fmt.Errorf("create multipart file field: %w", err)) + } + + audioFile, err := openAudio(audioPath) + if err != nil { + return u.producerError(ctx, fmt.Errorf("open audio file %q: %w", audioPath, err)) + } + u.setAudio(audioFile) + + _, copyErr := io.CopyBuffer(fileWriter, &contextReader{ctx: ctx, reader: audioFile}, make([]byte, whisperXUploadBufferSize)) + closeErr := audioFile.Close() + u.clearAudio(audioFile) + if copyErr != nil { + return u.producerError(ctx, fmt.Errorf("copy audio file %q: %w", audioPath, copyErr)) + } + if closeErr != nil { + return u.producerError(ctx, fmt.Errorf("close audio file %q: %w", audioPath, closeErr)) + } + + if err := writer.WriteField("language", language); err != nil { + return u.producerError(ctx, fmt.Errorf("write language form field: %w", err)) + } + if err := writer.Close(); err != nil { + return u.producerError(ctx, fmt.Errorf("close multipart writer: %w", err)) + } + return nil +} + +func (u *multipartUpload) producerError(ctx context.Context, err error) error { + if ctx.Err() != nil { + return ctx.Err() + } + u.mu.Lock() + aborted := u.aborted + u.mu.Unlock() + if aborted { + return nil + } + return err +} + +func (u *multipartUpload) setAudio(audio io.Closer) { + u.mu.Lock() + u.audio = audio + aborted := u.aborted + u.mu.Unlock() + if aborted { + _ = audio.Close() + } +} + +func (u *multipartUpload) clearAudio(audio io.Closer) { + u.mu.Lock() + if u.audio == audio { + u.audio = nil + } + u.mu.Unlock() +} + +func (u *multipartUpload) abort() { + u.mu.Lock() + if u.aborted { + u.mu.Unlock() + return + } + u.aborted = true + audio := u.audio + u.mu.Unlock() + + _ = u.reader.Close() + if audio != nil { + _ = audio.Close() + } +} + +type contextReader struct { + ctx context.Context + reader io.Reader +} + +func (r *contextReader) Read(p []byte) (int, error) { + select { + case <-r.ctx.Done(): + return 0, r.ctx.Err() + default: + return r.reader.Read(p) + } +} + func writeFileAtomic(path string, data []byte, perm os.FileMode) error { if strings.TrimSpace(path) == "" { return fmt.Errorf("path is required") diff --git a/internal/adapters/whisperx/http_test.go b/internal/adapters/whisperx/http_test.go index e03b194..937bbed 100644 --- a/internal/adapters/whisperx/http_test.go +++ b/internal/adapters/whisperx/http_test.go @@ -5,6 +5,7 @@ import ( "encoding/json" "errors" "io" + "mime/multipart" "net/http" "net/http/httptest" "os" @@ -19,6 +20,7 @@ func TestHTTPClientTranscribeSuccess(t *testing.T) { var gotLanguage string var gotFileField string var gotFileSize int + var gotFileData string srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { if r.Method != http.MethodPost { @@ -40,6 +42,7 @@ func TestHTTPClientTranscribeSuccess(t *testing.T) { t.Fatalf("ReadAll(file) error = %v", err) } gotFileSize = len(data) + gotFileData = string(data) w.Header().Set("Content-Type", "application/json") _, _ = w.Write([]byte(`{"schema":"speaker_transcript.v1","segments":[]}`)) @@ -77,13 +80,27 @@ func TestHTTPClientTranscribeSuccess(t *testing.T) { if gotFileSize == 0 { t.Fatal("file size = 0, want >0") } + if gotFileData != "audio-data" { + t.Fatalf("file data = %q, want exact payload", gotFileData) + } verifyJSONFile(t, outPath) } func TestHTTPClientRetriesOnTransientAndSucceeds(t *testing.T) { var calls atomic.Int32 + var payloads []string srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { n := calls.Add(1) + file, _, err := r.FormFile("file") + if err != nil { + t.Fatalf("FormFile(file) error = %v", err) + } + data, err := io.ReadAll(file) + _ = file.Close() + if err != nil { + t.Fatalf("ReadAll(file) error = %v", err) + } + payloads = append(payloads, string(data)) if n == 1 { http.Error(w, "temporary", http.StatusInternalServerError) return @@ -112,6 +129,9 @@ func TestHTTPClientRetriesOnTransientAndSucceeds(t *testing.T) { if calls.Load() != 2 { t.Fatalf("calls = %d, want 2", calls.Load()) } + if len(payloads) != 2 || payloads[0] != "audio-data" || payloads[1] != "audio-data" { + t.Fatalf("retry payloads = %#v, want two exact audio payloads", payloads) + } verifyJSONFile(t, outPath) } @@ -247,8 +267,247 @@ func TestHTTPClientConstructorValidation(t *testing.T) { if err == nil { t.Fatal("expected bad retry_delay error") } + for _, endpoint := range []string{"ftp://example.com/transcribe", "file:///tmp/transcribe", "//example.com/transcribe", "https:/missing-host"} { + if _, err := NewHTTPClientFromConfigValues(endpoint, "en", "30m", "2s", 1); err == nil { + t.Errorf("NewHTTPClientFromConfigValues(%q) error = nil, want endpoint validation error", endpoint) + } + } + for _, endpoint := range []string{"http://example.com/transcribe", "https://example.com/transcribe"} { + if _, err := NewHTTPClientFromConfigValues(endpoint, "en", "30m", "2s", 1); err != nil { + t.Errorf("NewHTTPClientFromConfigValues(%q) error = %v", endpoint, err) + } + } } +func TestHTTPClientStreamsUploadBeforeSourceCompletes(t *testing.T) { + release := make(chan struct{}) + source := newGatedReadCloser([]byte("audio-data"), release) + firstByteReceived := make(chan struct{}) + + srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + part := firstMultipartFilePart(t, r) + buf := make([]byte, 1) + if _, err := part.Read(buf); err != nil { + t.Errorf("Read(file) error = %v", err) + return + } + close(firstByteReceived) + if _, err := io.Copy(io.Discard, part); err != nil { + t.Errorf("discard remaining file data: %v", err) + return + } + _, _ = w.Write([]byte(`{"ok":true}`)) + })) + defer srv.Close() + + client := newTestHTTPClient(t, srv.URL) + client.openAudio = func(string) (io.ReadCloser, error) { return source, nil } + + done := make(chan error, 1) + go func() { + _, err := client.Transcribe(context.Background(), TranscribeRequest{AudioPath: "audio.flac", OutputRawTranscriptPath: filepath.Join(t.TempDir(), "raw.json")}) + done <- err + }() + + select { + case <-firstByteReceived: + close(release) + case <-time.After(time.Second): + t.Fatal("server did not receive streamed audio before source completed") + } + if err := <-done; err != nil { + t.Fatalf("Transcribe() error = %v", err) + } +} + +func TestHTTPClientSourceReadFailureReachesCaller(t *testing.T) { + sourceErr := errors.New("source read failed") + srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + _, _ = io.Copy(io.Discard, r.Body) + })) + defer srv.Close() + + client := newTestHTTPClient(t, srv.URL) + client.openAudio = func(string) (io.ReadCloser, error) { + return &failingReadCloser{first: []byte("partial"), err: sourceErr}, nil + } + + _, err := client.Transcribe(context.Background(), TranscribeRequest{AudioPath: "audio.flac", OutputRawTranscriptPath: filepath.Join(t.TempDir(), "raw.json")}) + if !errors.Is(err, sourceErr) { + t.Fatalf("Transcribe() error = %v, want source read failure", err) + } +} + +func TestHTTPClientEarlyServerResponseReleasesBlockedProducer(t *testing.T) { + release := make(chan struct{}) + source := newGatedReadCloser([]byte("audio-data"), release) + srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + http.Error(w, "bad request", http.StatusBadRequest) + })) + defer srv.Close() + + client := newTestHTTPClient(t, srv.URL) + client.openAudio = func(string) (io.ReadCloser, error) { return source, nil } + + done := make(chan error, 1) + go func() { + _, err := client.Transcribe(context.Background(), TranscribeRequest{AudioPath: "audio.flac", OutputRawTranscriptPath: filepath.Join(t.TempDir(), "raw.json")}) + done <- err + }() + + select { + case err := <-done: + if err == nil { + t.Fatal("Transcribe() error = nil, want HTTP status error") + } + case <-time.After(time.Second): + t.Fatal("Transcribe() did not finish after server closed the request early") + } + select { + case <-source.closed: + case <-time.After(time.Second): + t.Fatal("blocked audio source was not closed") + } +} + +func TestHTTPClientCancellationReleasesBlockedProducer(t *testing.T) { + release := make(chan struct{}) + source := newGatedReadCloser([]byte("audio-data"), release) + firstByteReceived := make(chan struct{}) + srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + part := firstMultipartFilePart(t, r) + buf := make([]byte, 1) + if _, err := part.Read(buf); err != nil { + t.Errorf("Read(file) error = %v", err) + return + } + close(firstByteReceived) + <-r.Context().Done() + })) + defer srv.Close() + + client := newTestHTTPClient(t, srv.URL) + client.openAudio = func(string) (io.ReadCloser, error) { return source, 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 <-firstByteReceived: + cancel() + case <-time.After(time.Second): + cancel() + t.Fatal("server did not receive initial streamed audio") + } + 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 finish after cancellation") + } + select { + case <-source.closed: + case <-time.After(time.Second): + t.Fatal("blocked audio source was not closed on cancellation") + } +} + +func TestHTTPClientBoundsWhisperXResponse(t *testing.T) { + srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + _, _ = io.Copy(io.Discard, r.Body) + _, _ = w.Write([]byte(`{"ok":true}`)) + })) + defer srv.Close() + + client, err := NewHTTPClient(HTTPClientConfig{TranscribeURL: srv.URL, Language: "en", Timeout: time.Second, MaxResponseBytes: 4}) + if err != nil { + t.Fatalf("NewHTTPClient() error = %v", err) + } + audioPath := writeWhisperXTestFile(t, "audio.flac", "audio-data") + _, err = client.Transcribe(context.Background(), TranscribeRequest{AudioPath: audioPath, OutputRawTranscriptPath: filepath.Join(t.TempDir(), "raw.json")}) + if err == nil || !strings.Contains(err.Error(), "whisperx response exceeds configured limit") { + t.Fatalf("Transcribe() error = %v, want bounded WhisperX response error", err) + } +} + +func newTestHTTPClient(t *testing.T, endpoint string) *HTTPClient { + t.Helper() + client, err := NewHTTPClientFromConfigValues(endpoint, "en", "2s", "1ms", 0) + if err != nil { + t.Fatalf("NewHTTPClientFromConfigValues() error = %v", err) + } + return client +} + +func firstMultipartFilePart(t *testing.T, r *http.Request) *multipart.Part { + t.Helper() + reader, err := r.MultipartReader() + if err != nil { + t.Fatalf("MultipartReader() error = %v", err) + } + part, err := reader.NextPart() + if err != nil { + t.Fatalf("NextPart() error = %v", err) + } + if part.FormName() != "file" { + t.Fatalf("first form field = %q, want file", part.FormName()) + } + return part +} + +type gatedReadCloser struct { + first []byte + release <-chan struct{} + closed chan struct{} + sent bool + once atomic.Bool +} + +func newGatedReadCloser(first []byte, release <-chan struct{}) *gatedReadCloser { + return &gatedReadCloser{first: first, release: release, closed: make(chan struct{})} +} + +func (r *gatedReadCloser) Read(p []byte) (int, error) { + if !r.sent { + r.sent = true + return copy(p, r.first), nil + } + select { + case <-r.release: + return 0, io.EOF + case <-r.closed: + return 0, errors.New("audio source closed") + } +} + +func (r *gatedReadCloser) Close() error { + if r.once.CompareAndSwap(false, true) { + close(r.closed) + } + return nil +} + +type failingReadCloser struct { + first []byte + err error + sent bool +} + +func (r *failingReadCloser) Read(p []byte) (int, error) { + if !r.sent { + r.sent = true + return copy(p, r.first), nil + } + return 0, r.err +} + +func (r *failingReadCloser) Close() error { return nil } + func writeWhisperXTestFile(t *testing.T, name, contents string) string { t.Helper() path := filepath.Join(t.TempDir(), name) diff --git a/internal/app/whisperx_wiring_test.go b/internal/app/whisperx_wiring_test.go index 268cb57..89dbcf7 100644 --- a/internal/app/whisperx_wiring_test.go +++ b/internal/app/whisperx_wiring_test.go @@ -132,8 +132,9 @@ func TestExecuteStagesExplicitWhisperXInjectionOverridesDefaultWiring(t *testing if err != nil { t.Fatalf("executeStages() error = %v", err) } - if len(fake.Requests) != 1 { - t.Fatalf("fake whisperx requests = %d, want 1", len(fake.Requests)) + requests := fake.RequestsSnapshot() + if len(requests) != 1 { + t.Fatalf("fake whisperx requests = %d, want 1", len(requests)) } } diff --git a/internal/config/validate.go b/internal/config/validate.go index 74bb514..a5781f7 100644 --- a/internal/config/validate.go +++ b/internal/config/validate.go @@ -395,11 +395,11 @@ func validateWhisperX(cfg WhisperXConfig) error { return fmt.Errorf("pipeline.whisperx.transcribe_url is required") } u, err := url.Parse(cfg.TranscribeURL) - if err != nil || u.Scheme == "" || u.Host == "" { + if err != nil || !u.IsAbs() || u.Host == "" || (strings.ToLower(u.Scheme) != "http" && strings.ToLower(u.Scheme) != "https") { if err != nil { - return fmt.Errorf("pipeline.whisperx.transcribe_url must be a valid URL: %w", err) + return fmt.Errorf("pipeline.whisperx.transcribe_url must be an absolute http or https URL: %w", err) } - return fmt.Errorf("pipeline.whisperx.transcribe_url must be a valid URL") + return fmt.Errorf("pipeline.whisperx.transcribe_url must be an absolute http or https URL") } if err := validateDuration("pipeline.whisperx.timeout", cfg.Timeout); err != nil { return err diff --git a/internal/config/validation_duration_test.go b/internal/config/validation_duration_test.go index 43dd60d..7d41a96 100644 --- a/internal/config/validation_duration_test.go +++ b/internal/config/validation_duration_test.go @@ -39,6 +39,33 @@ func TestValidateDurationsRequirePositiveValues(t *testing.T) { } } +func TestValidateWhisperXRequiresAbsoluteHTTPSEndpoint(t *testing.T) { + retries := 0 + concurrency := 1 + base := WhisperXConfig{ + Language: "en", + Timeout: "1s", + Retries: &retries, + RetryDelay: "0s", + Concurrency: &concurrency, + } + + for _, endpoint := range []string{"ftp://example.com/transcribe", "file:///tmp/transcribe", "//example.com/transcribe", "https:/missing-host"} { + cfg := base + cfg.TranscribeURL = endpoint + if err := validateWhisperX(cfg); err == nil || !strings.Contains(err.Error(), "absolute http or https URL") { + t.Errorf("validateWhisperX(%q) error = %v, want HTTP(S) endpoint validation", endpoint, err) + } + } + for _, endpoint := range []string{"http://example.com/transcribe", "https://example.com/transcribe"} { + cfg := base + cfg.TranscribeURL = endpoint + if err := validateWhisperX(cfg); err != nil { + t.Errorf("validateWhisperX(%q) error = %v", endpoint, err) + } + } +} + func TestValidateNotariusTimeoutRequiresPositiveValue(t *testing.T) { for _, value := range []string{"0s", "-1ms"} { t.Run(value, func(t *testing.T) { diff --git a/internal/stage/placeholders_test.go b/internal/stage/placeholders_test.go index 7c96b0f..c6f2e7c 100644 --- a/internal/stage/placeholders_test.go +++ b/internal/stage/placeholders_test.go @@ -235,8 +235,9 @@ func TestStagesReturnExpectedMetadata(t *testing.T) { } } - if len(wf.Requests) != 1 { - t.Fatalf("whisperx calls = %d, want 1", len(wf.Requests)) + whisperXRequests := wf.RequestsSnapshot() + if len(whisperXRequests) != 1 { + t.Fatalf("whisperx calls = %d, want 1", len(whisperXRequests)) } if len(sf.Requests) != 1 { t.Fatalf("seriatim calls = %d, want 1", len(sf.Requests)) diff --git a/internal/stage/transcribe_test.go b/internal/stage/transcribe_test.go index d5e3008..f1ec2c8 100644 --- a/internal/stage/transcribe_test.go +++ b/internal/stage/transcribe_test.go @@ -206,10 +206,11 @@ func TestTranscribeStageUsesRunLocalOutputAndMaterializesCanonical(t *testing.T) if err != nil { t.Fatalf("transcribe.Run() error = %v", err) } - if len(fake.Requests) != 1 { - t.Fatalf("requests = %d, want 1", len(fake.Requests)) + requests := fake.RequestsSnapshot() + if len(requests) != 1 { + t.Fatalf("requests = %d, want 1", len(requests)) } - runOut := fake.Requests[0].OutputRawTranscriptPath + runOut := requests[0].OutputRawTranscriptPath if !strings.Contains(runOut, filepath.Join("runs", m.RunID, "transcribe", "outputs")) { t.Fatalf("run-local output path = %q, want runs/{run_id}/transcribe/outputs path", runOut) }