From dc3288d15a4c98cca1aec2f2f3f785dccd895195 Mon Sep 17 00:00:00 2001 From: Eric Rakestraw Date: Tue, 22 Sep 2026 00:15:48 +0000 Subject: [PATCH] Make WhisperX uploads replayable --- docs/roadmap/implementation.md | 15 +- internal/adapters/whisperx/http.go | 202 ++--------- internal/adapters/whisperx/http_test.go | 2 +- internal/adapters/whisperx/multipart_body.go | 263 ++++++++++++++ .../adapters/whisperx/multipart_body_test.go | 327 ++++++++++++++++++ 5 files changed, 634 insertions(+), 175 deletions(-) create mode 100644 internal/adapters/whisperx/multipart_body.go create mode 100644 internal/adapters/whisperx/multipart_body_test.go diff --git a/docs/roadmap/implementation.md b/docs/roadmap/implementation.md index 0dc47a0..651d515 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-4 complete. +Status: Implementation in progress; Stages 1-5 complete. ## Scope and Instructions @@ -291,7 +291,18 @@ expose incomplete transcripts downstream. ## Stage 5: Implement Replayable Streaming Multipart Bodies -Status: Pending. +Status: Complete. + +Completion marker (2026-09-22): Replaced the producer-goroutine/pipe upload +with bounded standard-library multipart framing and independently opened, +pull-based streaming bodies shared by the initial request and `GetBody`. +Bodies are tracked, idempotently closed, canceled with their attempt context, +and retain genuine read/close failures without treating an intentionally +abandoned original as a replay failure. Added focused coverage for identical +framing, partial-read restart, simultaneous handles, source failures, and +blocked-reader release. Verified with `go test ./internal/adapters/whisperx`, +`go test -race ./internal/adapters/whisperx` (including ten repeated runs), and +`git diff --check`. Preparation: `internal/adapters/whisperx/http.go`, `http_test.go`, `docs/integrations/whisperx.md`, and the installed Go net/http Request/GetBody diff --git a/internal/adapters/whisperx/http.go b/internal/adapters/whisperx/http.go index 93755a8..49fdc9c 100644 --- a/internal/adapters/whisperx/http.go +++ b/internal/adapters/whisperx/http.go @@ -6,14 +6,11 @@ import ( "errors" "fmt" "io" - "mime/multipart" "net" "net/http" "net/url" "os" - "path/filepath" "strings" - "sync" "time" "gitea.maximumdirect.net/eric/narratio/internal/fileops" @@ -21,7 +18,6 @@ import ( const ( defaultMaxWhisperXResponseBytes int64 = 10 * 1024 * 1024 - whisperXUploadBufferSize = 32 * 1024 ) // HTTPClientConfig contains parsed, deterministic WhisperX HTTP client settings. @@ -192,21 +188,32 @@ func (c *HTTPClient) Transcribe(ctx context.Context, req TranscribeRequest) (Tra } func (c *HTTPClient) doTranscribeAttempt(ctx context.Context, audioPath string) (int, []byte, error) { - upload := newMultipartUpload(ctx, audioPath, c.language, c.openAudio) - defer upload.Close() + upload, err := newMultipartBodyFactory(ctx, audioPath, c.language, c.openAudio) + if err != nil { + return 0, nil, fmt.Errorf("prepare whisperx request body: %w", err) + } + defer func() { _ = upload.Close() }() + body, err := upload.NewBody() + if err != nil { + return 0, nil, fmt.Errorf("open whisperx request body: %w", err) + } - req, err := http.NewRequestWithContext(ctx, http.MethodPost, c.url.String(), upload) + req, err := http.NewRequestWithContext(ctx, http.MethodPost, c.url.String(), body) if err != nil { return 0, nil, fmt.Errorf("build whisperx request: %w", err) } req.Header.Set("Content-Type", upload.contentType) req.Header.Set("Expect", "100-continue") + req.GetBody = upload.NewBody 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) + if sourceErr := upload.SourceError(true); sourceErr != nil { + return 0, nil, fmt.Errorf("stream whisperx request body: %w", sourceErr) + } + if ctxErr := ctx.Err(); ctxErr != nil { + return 0, nil, fmt.Errorf("perform whisperx request: %w", ctxErr) } return 0, nil, fmt.Errorf("perform whisperx request: %w", err) } @@ -214,8 +221,8 @@ func (c *HTTPClient) doTranscribeAttempt(ctx context.Context, audioPath string) if resp.StatusCode < 200 || resp.StatusCode >= 300 { _ = upload.Close() - if producerErr := upload.Wait(); producerErr != nil { - return resp.StatusCode, nil, fmt.Errorf("stream whisperx request body: %w", producerErr) + if sourceErr := upload.SourceError(false); sourceErr != nil { + return resp.StatusCode, nil, fmt.Errorf("stream whisperx request body: %w", sourceErr) } if _, err := readWhisperXResponse(resp.Body, c.maxResponseBytes); err != nil { return resp.StatusCode, nil, fmt.Errorf("read whisperx response body: %w", err) @@ -223,13 +230,19 @@ func (c *HTTPClient) doTranscribeAttempt(ctx context.Context, audioPath string) 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, responseErr := readWhisperXResponse(resp.Body, c.maxResponseBytes) + _ = upload.Close() + if sourceErr := upload.SourceError(false); sourceErr != nil { + if responseErr != nil { + return resp.StatusCode, nil, fmt.Errorf("stream whisperx request body: %w", errors.Join(sourceErr, responseErr)) + } + return resp.StatusCode, nil, fmt.Errorf("stream whisperx request body: %w", sourceErr) } - - data, err := readWhisperXResponse(resp.Body, c.maxResponseBytes) - if err != nil { - return resp.StatusCode, nil, fmt.Errorf("read whisperx response body: %w", err) + if err := ctx.Err(); err != nil { + return resp.StatusCode, nil, fmt.Errorf("complete whisperx request: %w", err) + } + if responseErr != nil { + return resp.StatusCode, nil, fmt.Errorf("read whisperx response body: %w", responseErr) } return resp.StatusCode, data, nil } @@ -286,161 +299,6 @@ func isHTTPURLScheme(scheme string) bool { } } -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() - 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 775c441..714bc2a 100644 --- a/internal/adapters/whisperx/http_test.go +++ b/internal/adapters/whisperx/http_test.go @@ -365,7 +365,7 @@ func TestHTTPClientEarlyServerResponseReturns(t *testing.T) { } } -func TestHTTPClientCancellationReleasesBlockedProducer(t *testing.T) { +func TestHTTPClientCancellationReleasesBlockedSource(t *testing.T) { release := make(chan struct{}) source := newGatedReadCloser([]byte("audio-data"), release) firstByteReceived := make(chan struct{}) diff --git a/internal/adapters/whisperx/multipart_body.go b/internal/adapters/whisperx/multipart_body.go new file mode 100644 index 0000000..5d4e511 --- /dev/null +++ b/internal/adapters/whisperx/multipart_body.go @@ -0,0 +1,263 @@ +package whisperx + +import ( + "bytes" + "context" + "errors" + "fmt" + "io" + "mime/multipart" + "path/filepath" + "sync" +) + +type multipartBodyFactory struct { + ctx context.Context + audioPath string + openAudio func(string) (io.ReadCloser, error) + prefix []byte + suffix []byte + contentType string + + mu sync.Mutex + closed bool + bodies []*multipartBody +} + +func newMultipartBodyFactory( + ctx context.Context, + audioPath string, + language string, + openAudio func(string) (io.ReadCloser, error), +) (*multipartBodyFactory, error) { + if ctx == nil { + ctx = context.Background() + } + if openAudio == nil { + return nil, fmt.Errorf("audio opener is required") + } + var framing bytes.Buffer + writer := multipart.NewWriter(&framing) + if _, err := writer.CreateFormFile("file", filepath.Base(audioPath)); err != nil { + return nil, fmt.Errorf("create multipart file field: %w", err) + } + prefixLength := framing.Len() + if err := writer.WriteField("language", language); err != nil { + return nil, fmt.Errorf("write language form field: %w", err) + } + if err := writer.Close(); err != nil { + return nil, fmt.Errorf("close multipart framing: %w", err) + } + framingBytes := framing.Bytes() + return &multipartBodyFactory{ + ctx: ctx, + audioPath: audioPath, + openAudio: openAudio, + prefix: append([]byte(nil), framingBytes[:prefixLength]...), + suffix: append([]byte(nil), framingBytes[prefixLength:]...), + contentType: writer.FormDataContentType(), + }, nil +} + +func (factory *multipartBodyFactory) NewBody() (io.ReadCloser, error) { + factory.mu.Lock() + if factory.closed { + factory.mu.Unlock() + return nil, fmt.Errorf("multipart body factory is closed") + } + if err := factory.ctx.Err(); err != nil { + factory.mu.Unlock() + return nil, err + } + audio, err := factory.openAudio(factory.audioPath) + if err != nil { + factory.mu.Unlock() + return nil, fmt.Errorf("open audio file %q: %w", factory.audioPath, err) + } + if err := factory.ctx.Err(); err != nil { + factory.mu.Unlock() + _ = audio.Close() + return nil, err + } + source := &multipartAudioSource{audio: audio, closeDone: make(chan struct{})} + body := &multipartBody{ + ctx: factory.ctx, + source: source, + reader: io.MultiReader(bytes.NewReader(factory.prefix), source, bytes.NewReader(factory.suffix)), + } + factory.bodies = append(factory.bodies, body) + factory.mu.Unlock() + stopCancel := context.AfterFunc(factory.ctx, func() { + _ = body.Close() + }) + body.mu.Lock() + body.stopCancel = stopCancel + closed := body.closed + body.mu.Unlock() + if closed { + stopCancel() + } + if err := factory.ctx.Err(); err != nil { + _ = body.Close() + return nil, err + } + return body, nil +} + +func (factory *multipartBodyFactory) Close() error { + factory.mu.Lock() + if factory.closed { + bodies := append([]*multipartBody(nil), factory.bodies...) + factory.mu.Unlock() + return multipartBodyCloseErrors(bodies) + } + factory.closed = true + bodies := append([]*multipartBody(nil), factory.bodies...) + factory.mu.Unlock() + for _, body := range bodies { + _ = body.Close() + } + return multipartBodyCloseErrors(bodies) +} + +func (factory *multipartBodyFactory) SourceError(includeAbandoned bool) error { + factory.mu.Lock() + bodies := append([]*multipartBody(nil), factory.bodies...) + factory.mu.Unlock() + var result error + for _, body := range bodies { + if err := body.sourceError(includeAbandoned); err != nil { + result = errors.Join(result, err) + } + } + return result +} + +func multipartBodyCloseErrors(bodies []*multipartBody) error { + var result error + for _, body := range bodies { + if err := body.sourceError(false); err != nil { + result = errors.Join(result, err) + } + } + return result +} + +type multipartBody struct { + ctx context.Context + reader io.Reader + source *multipartAudioSource + stopCancel func() bool + + mu sync.Mutex + closed bool + abandoned bool +} + +func (body *multipartBody) Read(buffer []byte) (int, error) { + if err := body.ctx.Err(); err != nil { + return 0, err + } + body.mu.Lock() + closed := body.closed + body.mu.Unlock() + if closed { + return 0, io.ErrClosedPipe + } + return body.reader.Read(buffer) +} + +func (body *multipartBody) Close() error { + body.mu.Lock() + if body.closed { + body.mu.Unlock() + return body.source.Close() + } + body.closed = true + body.abandoned = !body.source.reachedEOF() + stopCancel := body.stopCancel + body.mu.Unlock() + if stopCancel != nil { + stopCancel() + } + return body.source.Close() +} + +func (body *multipartBody) sourceError(includeAbandoned bool) error { + body.mu.Lock() + abandoned := body.abandoned + body.mu.Unlock() + err := body.source.Err() + if err != nil { + return err + } + if abandoned && !includeAbandoned { + return nil + } + return nil +} + +type multipartAudioSource struct { + audio io.ReadCloser + + mu sync.Mutex + closing bool + eof bool + readErr error + closeErr error + closeDone chan struct{} + closeResult error + closed bool +} + +func (source *multipartAudioSource) Read(buffer []byte) (int, error) { + n, err := source.audio.Read(buffer) + source.mu.Lock() + switch { + case errors.Is(err, io.EOF): + source.eof = true + case err != nil && !source.closing: + source.readErr = errors.Join(source.readErr, err) + } + source.mu.Unlock() + return n, err +} + +func (source *multipartAudioSource) Close() error { + source.mu.Lock() + if source.closed { + done := source.closeDone + source.mu.Unlock() + <-done + source.mu.Lock() + err := source.closeResult + source.mu.Unlock() + return err + } + source.closed = true + source.closing = true + reachedEOF := source.eof + source.mu.Unlock() + + err := source.audio.Close() + source.mu.Lock() + source.closeResult = err + if err != nil && reachedEOF { + source.closeErr = err + } + close(source.closeDone) + source.mu.Unlock() + return err +} + +func (source *multipartAudioSource) reachedEOF() bool { + source.mu.Lock() + defer source.mu.Unlock() + return source.eof +} + +func (source *multipartAudioSource) Err() error { + source.mu.Lock() + defer source.mu.Unlock() + return errors.Join(source.readErr, source.closeErr) +} diff --git a/internal/adapters/whisperx/multipart_body_test.go b/internal/adapters/whisperx/multipart_body_test.go new file mode 100644 index 0000000..55c1b71 --- /dev/null +++ b/internal/adapters/whisperx/multipart_body_test.go @@ -0,0 +1,327 @@ +package whisperx + +import ( + "bytes" + "context" + "errors" + "io" + "mime" + "mime/multipart" + "strings" + "sync" + "sync/atomic" + "testing" + "time" +) + +func TestMultipartBodyReplayUsesIdenticalFraming(t *testing.T) { + factory := newTestMultipartBodyFactory(t, context.Background(), "audio-data") + defer factory.Close() + + original, err := factory.NewBody() + if err != nil { + t.Fatalf("NewBody() original error = %v", err) + } + replay, err := factory.NewBody() + if err != nil { + t.Fatalf("NewBody() replay error = %v", err) + } + originalBytes, err := io.ReadAll(original) + if err != nil { + t.Fatalf("ReadAll(original) error = %v", err) + } + replayBytes, err := io.ReadAll(replay) + if err != nil { + t.Fatalf("ReadAll(replay) error = %v", err) + } + if !bytes.Equal(originalBytes, replayBytes) { + t.Fatal("original and replay multipart bytes differ") + } + + parts := parseMultipartParts(t, factory.contentType, replayBytes) + if len(parts) != 2 { + t.Fatalf("part count = %d, want 2", len(parts)) + } + if parts[0].name != "file" || parts[0].filename != "audio.flac" || string(parts[0].data) != "audio-data" { + t.Fatalf("file part = %#v, want file/audio.flac with exact audio", parts[0]) + } + if parts[1].name != "language" || string(parts[1].data) != "en" { + t.Fatalf("language part = %#v, want language=en", parts[1]) + } +} + +func TestMultipartBodyReplayRestartsAfterPartialOriginal(t *testing.T) { + factory := newTestMultipartBodyFactory(t, context.Background(), "audio-data") + defer factory.Close() + + original, err := factory.NewBody() + if err != nil { + t.Fatalf("NewBody() original error = %v", err) + } + partial := make([]byte, len(factory.prefix)+3) + if _, err := io.ReadFull(original, partial); err != nil { + t.Fatalf("ReadFull(original) error = %v", err) + } + if err := original.Close(); err != nil { + t.Fatalf("Close(original) error = %v", err) + } + + replay, err := factory.NewBody() + if err != nil { + t.Fatalf("NewBody() replay error = %v", err) + } + replayBytes, err := io.ReadAll(replay) + if err != nil { + t.Fatalf("ReadAll(replay) error = %v", err) + } + parts := parseMultipartParts(t, factory.contentType, replayBytes) + if got := string(parts[0].data); got != "audio-data" { + t.Fatalf("replayed audio = %q, want complete payload", got) + } + if err := factory.SourceError(false); err != nil { + t.Fatalf("SourceError(false) = %v, want abandoned original ignored", err) + } +} + +func TestMultipartBodySimultaneousBodiesUseIndependentSources(t *testing.T) { + var opens atomic.Int32 + var closes atomic.Int32 + factory, err := newMultipartBodyFactory(context.Background(), "audio.flac", "en", func(string) (io.ReadCloser, error) { + opens.Add(1) + return &countedReadCloser{Reader: strings.NewReader("audio-data"), closes: &closes}, nil + }) + if err != nil { + t.Fatalf("newMultipartBodyFactory() error = %v", err) + } + + first, err := factory.NewBody() + if err != nil { + t.Fatalf("NewBody() first error = %v", err) + } + second, err := factory.NewBody() + if err != nil { + t.Fatalf("NewBody() second error = %v", err) + } + var firstBytes, secondBytes []byte + var firstErr, secondErr error + var group sync.WaitGroup + group.Add(2) + go func() { + defer group.Done() + firstBytes, firstErr = io.ReadAll(first) + }() + go func() { + defer group.Done() + secondBytes, secondErr = io.ReadAll(second) + }() + group.Wait() + if firstErr != nil || secondErr != nil { + t.Fatalf("concurrent reads returned (%v, %v)", firstErr, secondErr) + } + if !bytes.Equal(firstBytes, secondBytes) { + t.Fatal("independent body bytes differ") + } + if got := opens.Load(); got != 2 { + t.Fatalf("audio opens = %d, want 2", got) + } + if err := factory.Close(); err != nil { + t.Fatalf("Close() error = %v", err) + } + if err := factory.Close(); err != nil { + t.Fatalf("second Close() error = %v", err) + } + if got := closes.Load(); got != 2 { + t.Fatalf("audio closes = %d, want 2", got) + } +} + +func TestMultipartBodyReportsSourceFailures(t *testing.T) { + t.Run("read", func(t *testing.T) { + sourceErr := errors.New("read source") + factory, err := newMultipartBodyFactory(context.Background(), "audio.flac", "en", func(string) (io.ReadCloser, error) { + return &failingReadCloser{first: []byte("partial"), err: sourceErr}, nil + }) + if err != nil { + t.Fatalf("newMultipartBodyFactory() error = %v", err) + } + body, err := factory.NewBody() + if err != nil { + t.Fatalf("NewBody() error = %v", err) + } + if _, err := io.ReadAll(body); !errors.Is(err, sourceErr) { + t.Fatalf("ReadAll() error = %v, want source failure", err) + } + _ = body.Close() + if err := factory.SourceError(false); !errors.Is(err, sourceErr) { + t.Fatalf("SourceError(false) = %v, want source failure", err) + } + }) + + t.Run("close after complete read", func(t *testing.T) { + closeErr := errors.New("close source") + factory, err := newMultipartBodyFactory(context.Background(), "audio.flac", "en", func(string) (io.ReadCloser, error) { + return &errorCloseReadCloser{Reader: strings.NewReader("audio-data"), err: closeErr}, nil + }) + if err != nil { + t.Fatalf("newMultipartBodyFactory() error = %v", err) + } + body, err := factory.NewBody() + if err != nil { + t.Fatalf("NewBody() error = %v", err) + } + if _, err := io.ReadAll(body); err != nil { + t.Fatalf("ReadAll() error = %v", err) + } + if err := body.Close(); !errors.Is(err, closeErr) { + t.Fatalf("Close() error = %v, want source close failure", err) + } + if err := factory.SourceError(false); !errors.Is(err, closeErr) { + t.Fatalf("SourceError(false) = %v, want source close failure", err) + } + }) +} + +func TestMultipartBodyCloseAndCancelReleaseBlockedSource(t *testing.T) { + for _, test := range []struct { + name string + release func(context.CancelFunc, io.ReadCloser) + }{ + {name: "close", release: func(_ context.CancelFunc, body io.ReadCloser) { _ = body.Close() }}, + {name: "cancel", release: func(cancel context.CancelFunc, _ io.ReadCloser) { cancel() }}, + } { + t.Run(test.name, func(t *testing.T) { + ctx, cancel := context.WithCancel(context.Background()) + defer cancel() + source := newSignalingBlockedReadCloser() + factory, err := newMultipartBodyFactory(ctx, "audio.flac", "en", func(string) (io.ReadCloser, error) { + return source, nil + }) + if err != nil { + t.Fatalf("newMultipartBodyFactory() error = %v", err) + } + defer factory.Close() + body, err := factory.NewBody() + if err != nil { + t.Fatalf("NewBody() error = %v", err) + } + readDone := make(chan error, 1) + go func() { + _, err := io.ReadAll(body) + readDone <- err + }() + select { + case <-source.blocked: + case <-time.After(time.Second): + t.Fatal("audio source did not block") + } + test.release(cancel, body) + select { + case <-source.closed: + case <-time.After(time.Second): + t.Fatal("audio source was not closed") + } + select { + case <-readDone: + case <-time.After(time.Second): + t.Fatal("blocked body read did not return") + } + if test.name == "cancel" { + if _, err := factory.NewBody(); !errors.Is(err, context.Canceled) { + t.Fatalf("NewBody() after cancellation error = %v, want context.Canceled", err) + } + } + }) + } +} + +func newTestMultipartBodyFactory(t *testing.T, ctx context.Context, audio string) *multipartBodyFactory { + t.Helper() + factory, err := newMultipartBodyFactory(ctx, "audio.flac", "en", func(string) (io.ReadCloser, error) { + return io.NopCloser(strings.NewReader(audio)), nil + }) + if err != nil { + t.Fatalf("newMultipartBodyFactory() error = %v", err) + } + return factory +} + +type multipartTestPart struct { + name string + filename string + data []byte +} + +func parseMultipartParts(t *testing.T, contentType string, body []byte) []multipartTestPart { + t.Helper() + mediaType, params, err := mime.ParseMediaType(contentType) + if err != nil { + t.Fatalf("ParseMediaType() error = %v", err) + } + if mediaType != "multipart/form-data" { + t.Fatalf("media type = %q, want multipart/form-data", mediaType) + } + reader := multipart.NewReader(bytes.NewReader(body), params["boundary"]) + var parts []multipartTestPart + for { + part, err := reader.NextPart() + if errors.Is(err, io.EOF) { + return parts + } + if err != nil { + t.Fatalf("NextPart() error = %v", err) + } + data, err := io.ReadAll(part) + if err != nil { + t.Fatalf("ReadAll(part %q) error = %v", part.FormName(), err) + } + parts = append(parts, multipartTestPart{name: part.FormName(), filename: part.FileName(), data: data}) + } +} + +type countedReadCloser struct { + io.Reader + closes *atomic.Int32 + once sync.Once +} + +func (reader *countedReadCloser) Close() error { + reader.once.Do(func() { reader.closes.Add(1) }) + return nil +} + +type errorCloseReadCloser struct { + io.Reader + err error +} + +func (reader *errorCloseReadCloser) Close() error { return reader.err } + +type signalingBlockedReadCloser struct { + blocked chan struct{} + closed chan struct{} + once sync.Once + read bool +} + +func newSignalingBlockedReadCloser() *signalingBlockedReadCloser { + return &signalingBlockedReadCloser{blocked: make(chan struct{}), closed: make(chan struct{})} +} + +func (reader *signalingBlockedReadCloser) Read(buffer []byte) (int, error) { + if !reader.read { + reader.read = true + return copy(buffer, "audio-data"), nil + } + reader.once.Do(func() { close(reader.blocked) }) + <-reader.closed + return 0, errors.New("audio source closed") +} + +func (reader *signalingBlockedReadCloser) Close() error { + select { + case <-reader.closed: + default: + close(reader.closed) + } + return nil +}