diff --git a/voice/audio/codec.go b/voice/audio/codec.go index ae1a07a2..1775c5aa 100644 --- a/voice/audio/codec.go +++ b/voice/audio/codec.go @@ -10,6 +10,7 @@ const ( CodecMP3 CodecOpus CodecWAV + CodecFLAC ) // Format describes the encoding parameters of an audio stream. diff --git a/voice/audio/resample.go b/voice/audio/resample.go index 3e95fff9..eedd229d 100644 --- a/voice/audio/resample.go +++ b/voice/audio/resample.go @@ -62,6 +62,15 @@ func ResamplePCM16(data []byte, fromRate, toRate, channels int) []byte { // ResampleStream wraps a Stream[Frame] and resamples every PCM16 frame to // targetRate on the fly. Non-PCM frames or frames already at the target // rate are passed through unchanged. +// +// Leak note: the worker goroutine blocks in input.Read(). Read is a +// blocking, non-interruptible call, so if the consumer interrupts the +// returned stream while the worker is parked inside input.Read() on a +// stalled upstream, the worker cannot exit until that Read returns. The +// upstream input MUST therefore eventually become readable or return an +// error (e.g. by being closed/interrupted) for the worker to terminate. +// As a cheap safeguard the worker checks for an output interrupt before +// each Read so it exits promptly when interrupted between frames. func ResampleStream(input Stream[Frame], targetRate int) Stream[Frame] { if targetRate <= 0 { return input @@ -70,6 +79,13 @@ func ResampleStream(input Stream[Frame], targetRate int) Stream[Frame] { go func() { defer p.Close() for { + // Exit early if the consumer already interrupted the output, + // avoiding a blocking Read we would immediately discard. + select { + case <-p.ctx.Done(): + return + default: + } f, err := input.Read() if err != nil { if err != io.EOF { diff --git a/voice/audio/stream.go b/voice/audio/stream.go index 548476da..fd350c3c 100644 --- a/voice/audio/stream.go +++ b/voice/audio/stream.go @@ -53,6 +53,24 @@ func (p *Pipe[T]) Read() (T, error) { } p.mu.Unlock() + // Interrupt takes precedence over any pending value or a normal Close. + // Without this pre-check the select below is chosen at random when both a + // value/closed channel and ctx.Done() are ready, which would let an + // abnormal termination be observed as a clean io.EOF — e.g. a producer + // that calls Interrupt() and then runs a deferred Close(). Checking + // ctx.Done() first makes "Interrupt returns context.Canceled immediately, + // skipping buffered values" a deterministic guarantee. + select { + case <-p.ctx.Done(): + var zero T + err := p.ctx.Err() + p.mu.Lock() + p.lastErr = err + p.mu.Unlock() + return zero, err + default: + } + select { case v, ok := <-p.ch: if !ok { diff --git a/voice/audio/stream_test.go b/voice/audio/stream_test.go index 606e5371..4b779963 100644 --- a/voice/audio/stream_test.go +++ b/voice/audio/stream_test.go @@ -78,19 +78,14 @@ func TestPipe_InterruptSkipsBuffer(t *testing.T) { } pipe.Interrupt() - // Read must eventually return context.Canceled; it may return buffered values first - // (select is non-deterministic when both channel and ctx.Done() are ready). - var err error - for err == nil { - _, err = pipe.Read() - } - if err != context.Canceled { + // Interrupt is deterministic: the very first Read skips all buffered + // values and returns context.Canceled. + if _, err := pipe.Read(); err != context.Canceled { t.Errorf("Read: got error %v, want context.Canceled", err) } // Subsequent reads also return the same error - _, err = pipe.Read() - if err != context.Canceled { + if _, err := pipe.Read(); err != context.Canceled { t.Errorf("subsequent Read: got error %v, want context.Canceled", err) } } @@ -149,9 +144,10 @@ func TestPipe_InterruptThenClose(t *testing.T) { pipe.Interrupt() pipe.Close() // must not panic - // Read returns an error (either context.Canceled or io.EOF; select is non-deterministic when both are ready) - _, err := pipe.Read() - if err == nil { - t.Error("Read after Interrupt+Close: expected error, got nil") + // Interrupt takes precedence over Close: Read must deterministically + // report context.Canceled, never a clean io.EOF, so an abnormal + // termination is never observed as a normal end-of-stream. + if _, err := pipe.Read(); err != context.Canceled { + t.Errorf("Read after Interrupt+Close: got %v, want context.Canceled", err) } } diff --git a/voice/go.mod b/voice/go.mod index 1666018c..7b2ab735 100644 --- a/voice/go.mod +++ b/voice/go.mod @@ -8,6 +8,7 @@ require ( github.com/pion/webrtc/v4 v4.2.9 github.com/rs/xid v1.6.0 go.opentelemetry.io/otel/metric v1.40.0 + go.uber.org/goleak v1.3.0 ) require ( diff --git a/voice/provider/circuit_halfopen_test.go b/voice/provider/circuit_halfopen_test.go new file mode 100644 index 00000000..7d03e2a4 --- /dev/null +++ b/voice/provider/circuit_halfopen_test.go @@ -0,0 +1,122 @@ +package provider + +import ( + "testing" + "time" +) + +// TestCircuit_HalfOpenConsecutiveFailuresStayBounded is a white-box regression +// test for the half-open bounded-count fix (issue #4b). +// +// When a circuit's open window elapses, Allow transitions it to half-open and +// decays consecutiveFailures to 0, and a probe failure re-opens the circuit +// from that clean count. As a result the internal consecutiveFailures counter +// must stay bounded no matter how many open -> probe-fail cycles occur. +// +// The existing black-box test (TestCircuit_HalfOpenProbeFailureReopensFromClean +// Count) only checks a single re-open; it does not catch a counter that grows +// without bound across many cycles. This test drives many cycles and asserts +// the counter never exceeds the break threshold — before the fix it would grow +// roughly linearly with the number of cycles. +func TestCircuit_HalfOpenConsecutiveFailuresStayBounded(t *testing.T) { + c := NewCircuit(1) + policy := FallbackPolicy{CircuitBreakAfter: 2, CircuitOpen: 50 * time.Millisecond} + base := time.Now() + + // Open the circuit initially: two failures reach the break threshold. + c.OnFailure(0, base, policy, true) + if !c.OnFailure(0, base, policy, true) { + t.Fatal("circuit should open after reaching CircuitBreakAfter") + } + + maxSeen := 0 + record := func() { + // Same-package white-box access to the internal counter. + c.mu.Lock() + got := c.states[0].consecutiveFailures + c.mu.Unlock() + if got > maxSeen { + maxSeen = got + } + } + record() + + now := base + const cycles = 200 + for cycle := 0; cycle < cycles; cycle++ { + // Let the open window elapse so Allow yields a half-open probe. + now = now.Add(60 * time.Millisecond) + if !c.Allow(0, now) { + t.Fatalf("cycle %d: expected a half-open probe to be allowed", cycle) + } + // The probe fails, which must re-open the circuit from a clean count. + if !c.OnFailure(0, now, policy, true) { + t.Fatalf("cycle %d: half-open probe failure should re-open the circuit", cycle) + } + record() + } + + // Bounded property: the counter must not grow with the number of cycles. + // Before the fix it would be ~= 2 + cycles. + if maxSeen > policy.CircuitBreakAfter { + t.Fatalf("consecutiveFailures grew unbounded across %d cycles: max=%d, want <= %d", + cycles, maxSeen, policy.CircuitBreakAfter) + } +} + +// TestCircuit_HalfOpenAllowsSingleProbe verifies that once the open window +// elapses, Allow hands out exactly one half-open probe and holds every other +// caller until the probe is resolved via OnSuccess/OnFailure. Before the fix, +// the first probe zeroed openUntil and any concurrent/subsequent Allow saw a +// zero window and returned true, letting a post-cooldown burst hit a possibly +// still-unhealthy provider. +func TestCircuit_HalfOpenAllowsSingleProbe(t *testing.T) { + policy := FallbackPolicy{CircuitBreakAfter: 1, CircuitOpen: 50 * time.Millisecond} + base := time.Now() + + open := func() *Circuit { + c := NewCircuit(1) + if !c.OnFailure(0, base, policy, true) { + t.Fatal("circuit should open after reaching CircuitBreakAfter") + } + return c + } + afterCooldown := base.Add(60 * time.Millisecond) + + t.Run("holds other callers until probe resolves", func(t *testing.T) { + c := open() + if !c.Allow(0, afterCooldown) { + t.Fatal("first caller after cooldown should get the half-open probe") + } + // Probe is in flight; all other callers must be held. + for i := 0; i < 5; i++ { + if c.Allow(0, afterCooldown) { + t.Fatalf("call %d: expected probe to be held while one is in flight", i) + } + } + }) + + t.Run("probe success reopens the gate", func(t *testing.T) { + c := open() + if !c.Allow(0, afterCooldown) { + t.Fatal("expected half-open probe") + } + c.OnSuccess(0) + if !c.Allow(0, afterCooldown) { + t.Fatal("after a successful probe the circuit should be closed and allow traffic") + } + }) + + t.Run("probe failure re-opens and blocks", func(t *testing.T) { + c := open() + if !c.Allow(0, afterCooldown) { + t.Fatal("expected half-open probe") + } + if !c.OnFailure(0, afterCooldown, policy, true) { + t.Fatal("failed probe should re-open the circuit") + } + if c.Allow(0, afterCooldown) { + t.Fatal("re-opened circuit should block until the new cooldown elapses") + } + }) +} diff --git a/voice/provider/executor.go b/voice/provider/executor.go index 5782edca..884293a6 100644 --- a/voice/provider/executor.go +++ b/voice/provider/executor.go @@ -14,9 +14,16 @@ import ( var ErrSkipProvider = errors.New("provider: skip") // SleepWithContext blocks for duration d or until ctx is cancelled. +// It reports true if the full duration elapsed and false if ctx was cancelled. +// For d<=0 it still consults ctx so a cancelled context is honored immediately. func SleepWithContext(ctx context.Context, d time.Duration) bool { if d <= 0 { - return true + select { + case <-ctx.Done(): + return false + default: + return true + } } timer := time.NewTimer(d) defer timer.Stop() @@ -54,9 +61,21 @@ func RunWithFallback[T any]( var zero T var lastErr error - now := time.Now() for i := 0; i < n; i++ { + // Fail fast if the context is already done: return the context error + // directly so it is never misclassified as a provider failure that + // could trip the breaker. + if err := ctx.Err(); err != nil { + report.Error = err.Error() + emit() + return zero, err + } + + // Recompute the clock per iteration: a slow multi-provider run can let + // a provider's open window elapse mid-loop, so a single pre-loop + // timestamp would keep skipping providers that are actually recovered. + now := time.Now() name := nameFn(i) if !circuit.Allow(i, now) { report.Attempts = append(report.Attempts, Attempt{ @@ -69,6 +88,14 @@ func RunWithFallback[T any]( skipped := false for attempt := 0; attempt < policy.MaxAttempts; attempt++ { + // Guard each attempt on the context so a cancelled/expired ctx + // stops retries deterministically even when RetryBackoff==0 (the + // backoff sleep is otherwise the only cancellation checkpoint). + if err := ctx.Err(); err != nil { + report.Error = err.Error() + emit() + return zero, err + } result, err := execFn(ctx, i) if err == nil { circuit.OnSuccess(i) @@ -91,7 +118,10 @@ func RunWithFallback[T any]( lastErr = err retryable := policy.ShouldRetry(err) fallbackable := policy.ShouldFallback(err) - circuitOpened := circuit.OnFailure(i, time.Now(), policy, retryable) + // Only provider-attributable faults count toward the breaker; + // client-fault errors (bad input, cancellation) must not open a + // healthy provider. + circuitOpened := circuit.OnFailure(i, time.Now(), policy, IsProviderFault(err)) report.Attempts = append(report.Attempts, Attempt{ Provider: name, Attempt: attempt + 1, diff --git a/voice/provider/executor_test.go b/voice/provider/executor_test.go index 4c50f52d..11da92fe 100644 --- a/voice/provider/executor_test.go +++ b/voice/provider/executor_test.go @@ -262,6 +262,129 @@ func TestRunWithFallback_ContextCancelDuringBackoff(t *testing.T) { } } +func TestRunWithFallback_ContextCancelledBeforeStart(t *testing.T) { + circuit := NewCircuit(2) + policy := DefaultFallbackPolicy() + + ctx, cancel := context.WithCancel(context.Background()) + cancel() + + called := false + _, err := RunWithFallback(ctx, circuit, policy, "test.op", + 2, + func(i int) string { return "provider" }, + func(ctx context.Context, i int) (string, error) { + called = true + return "ok", nil + }, + ) + if !errors.Is(err, context.Canceled) { + t.Fatalf("err = %v, want context.Canceled", err) + } + if called { + t.Fatal("execFn should not be called when ctx is already cancelled") + } +} + +func TestRunWithFallback_ContextCancelDuringRetryNoBackoff(t *testing.T) { + circuit := NewCircuit(1) + // RetryBackoff==0 means the backoff sleep is not a cancellation checkpoint; + // retries must still stop deterministically once ctx is cancelled. + policy := FallbackPolicy{MaxAttempts: 5, RetryBackoff: 0} + + ctx, cancel := context.WithCancel(context.Background()) + + calls := 0 + _, err := RunWithFallback(ctx, circuit, policy, "test.op", + 1, + func(i int) string { return "provider" }, + func(ctx context.Context, i int) (string, error) { + calls++ + cancel() // cancel while returning a retryable error that ignores ctx + return "", errors.New("503 unavailable") + }, + ) + if !errors.Is(err, context.Canceled) { + t.Fatalf("err = %v, want context.Canceled", err) + } + if calls != 1 { + t.Fatalf("calls = %d, want 1 (retries must stop after cancellation)", calls) + } +} + +func TestCircuit_NonRetryableFailuresTripBreaker(t *testing.T) { + c := NewCircuit(1) + policy := FallbackPolicy{CircuitBreakAfter: 2, CircuitOpen: time.Minute} + base := time.Now() + + // A non-retryable but provider-attributable hard failure (e.g. an internal + // provider error) must still trip the breaker after CircuitBreakAfter — the + // breaker must not require retryable errors to open (#19). + hardErr := &ProviderError{Code: "internal_error", Op: "op", Message: "boom", Retry: false, Fallback: true} + if !IsProviderFault(hardErr) { + t.Fatal("internal_error must be provider-attributable") + } + if c.OnFailure(0, base, policy, IsProviderFault(hardErr)) { + t.Fatal("circuit should not open on the first hard failure") + } + if !c.OnFailure(0, base, policy, IsProviderFault(hardErr)) { + t.Fatal("persistent hard (non-retryable) failures should open the breaker") + } + if c.Allow(0, base.Add(time.Second)) { + t.Fatal("circuit should be open after consecutive hard failures") + } +} + +func TestCircuit_HalfOpenRecovery(t *testing.T) { + c := NewCircuit(1) + policy := FallbackPolicy{CircuitBreakAfter: 2, CircuitOpen: 50 * time.Millisecond} + base := time.Now() + + c.OnFailure(0, base, policy, true) + if !c.OnFailure(0, base, policy, true) { + t.Fatal("circuit should open after 2 failures") + } + if c.Allow(0, base.Add(10*time.Millisecond)) { + t.Fatal("circuit should stay open within the open window") + } + + after := base.Add(60 * time.Millisecond) + if !c.Allow(0, after) { + t.Fatal("circuit should allow a probe (half-open) after the window elapses") + } + // A successful probe fully closes the circuit. + c.OnSuccess(0) + if !c.Allow(0, after) { + t.Fatal("circuit should be closed after a successful probe") + } +} + +func TestCircuit_HalfOpenProbeFailureReopensFromCleanCount(t *testing.T) { + c := NewCircuit(1) + policy := FallbackPolicy{CircuitBreakAfter: 2, CircuitOpen: 50 * time.Millisecond} + base := time.Now() + + c.OnFailure(0, base, policy, true) + c.OnFailure(0, base, policy, true) // opens + + after := base.Add(60 * time.Millisecond) + if !c.Allow(0, after) { + t.Fatal("expected a half-open probe to be allowed") + } + // A single probe failure re-opens immediately from the decayed count, + // rather than requiring an ever-growing streak. + if !c.OnFailure(0, after, policy, true) { + t.Fatal("half-open probe failure should re-open the circuit") + } + if c.Allow(0, after.Add(10*time.Millisecond)) { + t.Fatal("circuit should be open again after the probe failure") + } + // And it remains bounded: the next window again yields a single probe. + if !c.Allow(0, after.Add(60*time.Millisecond)) { + t.Fatal("expected another probe after the second window elapses") + } +} + func TestRunWithFallback_ZeroProviders(t *testing.T) { circuit := NewCircuit(0) policy := DefaultFallbackPolicy() @@ -304,3 +427,56 @@ func TestSleepWithContext_Cancelled(t *testing.T) { t.Fatal("expected false for cancelled context") } } + +// TestCircuit_ClientFaultDoesNotTripBreaker locks in that client-attributable +// failures (bad input) never open a healthy provider's breaker, while +// provider-attributable faults still do (regression guard for the breaker +// accounting change). +func TestCircuit_ClientFaultDoesNotTripBreaker(t *testing.T) { + c := NewCircuit(1) + policy := DefaultFallbackPolicy() // CircuitBreakAfter = 2 + now := time.Now() + + badInput := &ProviderError{Code: "bad_audio", Op: "tts.synthesize", Message: "invalid sample rate", Retry: false, Fallback: true} + for i := 0; i < 10; i++ { + if c.OnFailure(0, now, policy, IsProviderFault(badInput)) { + t.Fatalf("client-fault failure #%d wrongly opened the breaker", i+1) + } + } + if !c.Allow(0, now) { + t.Fatal("healthy provider should stay allowed after client-fault failures") + } + + // Sanity: provider-attributable faults still trip after CircuitBreakAfter. + provErr := &ProviderError{Code: "provider_unavailable", Op: "tts.synthesize", Message: "503", Retry: true, Fallback: true} + _ = c.OnFailure(0, now, policy, IsProviderFault(provErr)) + if !c.OnFailure(0, now, policy, IsProviderFault(provErr)) { + t.Fatal("provider-attributable faults should still trip the breaker") + } +} + +// TestRunWithFallback_ClientFaultDoesNotSkipHealthyProvider verifies that a +// client repeatedly sending bad input never causes a healthy provider to be +// skipped with an open circuit (the regression the diff review flagged). +func TestRunWithFallback_ClientFaultDoesNotSkipHealthyProvider(t *testing.T) { + circuit := NewCircuit(1) + policy := DefaultFallbackPolicy() + + calls := 0 + for req := 0; req < 5; req++ { + _, err := RunWithFallback(context.Background(), circuit, policy, "tts.synthesize", + 1, + func(i int) string { return "primary" }, + func(ctx context.Context, i int) (string, error) { + calls++ + return "", &ProviderError{Code: "bad_audio", Op: "tts.synthesize", Message: "bad input", Retry: false, Fallback: true} + }, + ) + if err == nil { + t.Fatalf("req %d: expected an error for bad input", req) + } + } + if calls != 5 { + t.Fatalf("execFn called %d times, want 5 — a healthy provider was wrongly skipped by an open breaker on client-fault input", calls) + } +} diff --git a/voice/provider/fallback.go b/voice/provider/fallback.go index 666ec074..5c5af8c2 100644 --- a/voice/provider/fallback.go +++ b/voice/provider/fallback.go @@ -140,6 +140,27 @@ func IsRetryable(err error) bool { return code == "timeout" || code == "transport_error" || code == "provider_unavailable" } +// IsProviderFault reports whether err should count against a provider's circuit +// breaker. Failures attributable to the caller/client — context cancellation and +// bad input (bad audio/codec/sample-rate) — must NOT penalize provider health; +// otherwise a client repeatedly sending bad input would trip a healthy +// provider's breaker (and cascade to fallbacks). Everything else (timeout, +// transport, provider_unavailable, internal_error) is treated as a +// provider-attributable fault. +func IsProviderFault(err error) bool { + if err == nil { + return false + } + if errors.Is(err, context.Canceled) { + return false + } + var ce ClassifiedError + if errors.As(err, &ce) { + return ce.ErrorCode() != "bad_audio" + } + return ClassifyByMessage(err.Error()) != "bad_audio" +} + type Circuit struct { mu sync.Mutex states []circuitState @@ -148,6 +169,12 @@ type Circuit struct { type circuitState struct { consecutiveFailures int openUntil time.Time + // halfOpen is set when the open window has elapsed and a single probe has + // been allowed. While it is set, Allow holds every other caller so only one + // probe is in flight at a time. A probe success (OnSuccess) fully closes the + // circuit, while a probe failure (OnFailure) re-opens it immediately from a + // clean failure count so recovery is gradual and the counter stays bounded. + halfOpen bool } func NewCircuit(size int) *Circuit { @@ -163,7 +190,26 @@ func (c *Circuit) Allow(index int, now time.Time) bool { } c.mu.Lock() defer c.mu.Unlock() - return !now.Before(c.states[index].openUntil) + state := &c.states[index] + if state.openUntil.IsZero() { + // A zero open window means the circuit is either fully closed or a + // half-open probe is already in flight. When a probe is in flight, hold + // every other caller (return false) until OnSuccess/OnFailure resolves + // it, so exactly one probe reaches a provider that may still be + // unhealthy instead of a post-cooldown burst. + return !state.halfOpen + } + if now.Before(state.openUntil) { + // Open: still within the cool-down window. + return false + } + // Open window elapsed: transition to half-open and allow one probe. Decay + // the failure count to zero so a probe failure re-opens from a clean slate + // (bounding the counter) rather than compounding the previous streak. + state.openUntil = time.Time{} + state.consecutiveFailures = 0 + state.halfOpen = true + return true } func (c *Circuit) OnSuccess(index int) { @@ -175,8 +221,19 @@ func (c *Circuit) OnSuccess(index int) { c.states[index] = circuitState{} } -func (c *Circuit) OnFailure(index int, now time.Time, policy FallbackPolicy, retryable bool) bool { - if c == nil || index < 0 || index >= len(c.states) || !retryable { +// OnFailure records a provider failure and reports whether the circuit opened. +// +// attributable gates breaker accounting: only provider-attributable faults +// (see IsProviderFault) count toward consecutiveFailures. Retryability does NOT +// gate accounting — a persistently failing provider must eventually trip its +// breaker even when it returns hard (non-retryable) errors — but client-fault +// errors (bad input, caller cancellation) are excluded so they never open a +// healthy provider. +func (c *Circuit) OnFailure(index int, now time.Time, policy FallbackPolicy, attributable bool) bool { + if c == nil || index < 0 || index >= len(c.states) { + return false + } + if !attributable { return false } policy = policy.Normalize() @@ -184,10 +241,20 @@ func (c *Circuit) OnFailure(index int, now time.Time, policy FallbackPolicy, ret defer c.mu.Unlock() state := &c.states[index] state.consecutiveFailures++ + + canOpen := policy.CircuitBreakAfter > 0 && policy.CircuitOpen > 0 opened := false - if policy.CircuitBreakAfter > 0 && - policy.CircuitOpen > 0 && - state.consecutiveFailures >= policy.CircuitBreakAfter { + switch { + case state.halfOpen: + // A half-open probe failed: re-open immediately from the clean count so + // recovery restarts a fresh cool-down window instead of an ever-growing + // streak. + state.halfOpen = false + if canOpen { + state.openUntil = now.Add(policy.CircuitOpen) + opened = true + } + case canOpen && state.consecutiveFailures >= policy.CircuitBreakAfter: state.openUntil = now.Add(policy.CircuitOpen) opened = true } diff --git a/voice/stt/bytedance/bytedance.go b/voice/stt/bytedance/bytedance.go index 954022ac..3b4f079c 100644 --- a/voice/stt/bytedance/bytedance.go +++ b/voice/stt/bytedance/bytedance.go @@ -227,11 +227,15 @@ func (s *STT) RecognizeStream( } } for { - select { - case <-innerCtx.Done(): - writerDone <- innerCtx.Err() + // input is owned by the caller (it is passed into + // RecognizeStream), so we must not Close/Interrupt it to unblock + // a pending Read. Stream.Read carries no context either, so the + // pragmatic fix against a writer leak is to observe innerCtx + // before each Read and return promptly once the reader goroutine + // cancels it (via its deferred innerCancel on every exit path). + if err := innerCtx.Err(); err != nil { + writerDone <- err return - default: } frame, readErr := input.Read() if readErr == io.EOF { @@ -258,6 +262,14 @@ func (s *STT) RecognizeStream( return case err := <-writerDone: if err != nil && err != io.EOF && innerCtx.Err() == nil { + // A genuine writer send-failure means the server never + // received the full stream, so the transcript is truncated. + // Interrupt the output (mirroring the server-error path) + // instead of falling through to defer out.Close(), which + // would surface a clean io.EOF indistinguishable from a + // normal end-of-stream. + telemetry.Error(innerCtx, fmt.Sprintf("bytedance stt: writer error: %v", err)) + out.Interrupt() return } writerDone = nil @@ -311,7 +323,9 @@ func (s *STT) emitResults(ctx context.Context, resp *asrResponse, lang string, o case <-ctx.Done(): return default: - out.Send(stt.STTResult{Text: resp.payload.Result.Text, IsFinal: resp.isLast, Lang: lang}) + if !out.Send(stt.STTResult{Text: resp.payload.Result.Text, IsFinal: resp.isLast, Lang: lang}) { + return + } } } } diff --git a/voice/stt/bytedance/bytedance_leak_test.go b/voice/stt/bytedance/bytedance_leak_test.go new file mode 100644 index 00000000..13feedfe --- /dev/null +++ b/voice/stt/bytedance/bytedance_leak_test.go @@ -0,0 +1,191 @@ +package bytedance + +import ( + "context" + "encoding/binary" + "encoding/json" + "net/http" + "net/http/httptest" + "strings" + "testing" + "time" + + "github.com/GizClaw/flowcraft/voice/audio" + "github.com/coder/websocket" + "go.uber.org/goleak" +) + +// wsTestURL converts an httptest server URL to a ws:// URL. +func wsTestURL(httpURL string) string { + return strings.Replace(httpURL, "http://", "ws://", 1) +} + +// singleFrameInput returns a closed input pipe carrying one audio frame, which +// drives RecognizeStream's writer to send the frame then a finish marker. +func singleFrameInput() *audio.Pipe[audio.Frame] { + input := audio.NewPipe[audio.Frame](2) + input.Send(audio.Frame{ + Data: make([]byte, 3200), + Format: audio.Format{SampleRate: 16000, Channels: 1, BitDepth: 16}, + }) + input.Close() + return input +} + +// loopingFrameStream yields empty-data frames forever — it never returns io.EOF +// and never blocks. This keeps RecognizeStream's writer goroutine spinning in +// its read loop with NO side effects: because Data is empty the writer never +// calls sendAudio, so the writer's only clean exit is the `innerCtx.Err()` +// guard before input.Read (the fix under test), which the reader trips via its +// deferred innerCancel on early/server-error/ctx-cancel exit. Frames carrying +// real data would let the writer also exit via a sendAudio error after the +// connection closes, masking a missing guard — empty frames isolate the guard. +// +// It owns no goroutine, so there is nothing for the test itself to leak. The +// tiny pause avoids a hot CPU spin without affecting the guarantees. The first +// frame's Format is what RecognizeStream reads to build the ASR request. +type loopingFrameStream struct { + format audio.Format +} + +func (s loopingFrameStream) Read() (audio.Frame, error) { + time.Sleep(time.Millisecond) + return audio.Frame{Format: s.format}, nil +} + +func loopingInput() loopingFrameStream { + return loopingFrameStream{format: audio.Format{SampleRate: 16000, Channels: 1, BitDepth: 16}} +} + +// TestRecognizeStream_NoLeak_NormalCompletion locks in the writer/reader +// goroutine teardown on normal completion (finish sent, final frame received, +// output drained to EOF). +func TestRecognizeStream_NoLeak_NormalCompletion(t *testing.T) { + defer goleak.VerifyNone(t) + + srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + c, err := websocket.Accept(w, r, nil) + if err != nil { + return + } + defer func() { _ = c.Close(websocket.StatusNormalClosure, "") }() + + if _, _, err := c.Read(r.Context()); err != nil { + return + } + // Consume audio until the finish (negative-seq) frame. + for { + _, raw, err := c.Read(r.Context()) + if err != nil { + return + } + if len(raw) < 8 { + continue + } + if binary.BigEndian.Uint32(raw[0:4])&msgTypeFlagMask == flagNegativeSeq { + break + } + } + final := asrResponsePayload{} + final.Result.Utterances = []asrUtterance{{Text: "你好", Definite: true}} + data, _ := json.Marshal(final) + _ = c.Write(r.Context(), websocket.MessageBinary, buildServerFullFrame(data, true)) + })) + defer srv.Close() + + s, _ := New(WithAppID("app"), WithToken("tok"), WithHost(wsTestURL(srv.URL))) + out, err := s.RecognizeStream(context.Background(), singleFrameInput()) + if err != nil { + t.Fatal(err) + } + for { + if _, err := out.Read(); err != nil { + break + } + } +} + +// TestRecognizeStream_NoLeak_ServerError locks in the writer-goroutine teardown +// on early server-error termination (the fix under test). The input stream +// keeps yielding frames forever (loopingInput), so the writer never reaches its +// own io.EOF/send-finish exit; teardown of the writer depends entirely on the +// reader's deferred innerCancel unblocking the `innerCtx.Err()` guard. If that +// guard is removed the writer spins forever and goleak fires. +func TestRecognizeStream_NoLeak_ServerError(t *testing.T) { + defer goleak.VerifyNone(t) + + srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + c, err := websocket.Accept(w, r, nil) + if err != nil { + return + } + defer func() { _ = c.Close(websocket.StatusNormalClosure, "") }() + + if _, _, err := c.Read(r.Context()); err != nil { + return + } + errPayload, _ := json.Marshal(asrResponsePayload{Error: "quota exceeded"}) + _ = c.Write(r.Context(), websocket.MessageBinary, buildErrorFrame(1013, errPayload)) + // Keep the connection open so the client observes the error frame + // before the normal-closure handshake. + _, _, _ = c.Read(r.Context()) + })) + defer srv.Close() + + s, _ := New(WithAppID("app"), WithToken("tok"), WithHost(wsTestURL(srv.URL))) + out, err := s.RecognizeStream(context.Background(), loopingInput()) + if err != nil { + t.Fatal(err) + } + for { + if _, err := out.Read(); err != nil { + break + } + } +} + +// TestRecognizeStream_NoLeak_CtxCancelled locks in goroutine teardown when the +// caller cancels the context while the reader is blocked awaiting results. The +// input stream keeps yielding frames forever (loopingInput), so the writer +// stays looping and can only exit via the `innerCtx.Err()` guard once the +// reader's deferred innerCancel fires on ctx cancel. If that guard is removed +// the writer spins forever and goleak fires. +func TestRecognizeStream_NoLeak_CtxCancelled(t *testing.T) { + defer goleak.VerifyNone(t) + + srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + c, err := websocket.Accept(w, r, nil) + if err != nil { + return + } + defer func() { _ = c.Close(websocket.StatusNormalClosure, "") }() + // Consume everything (request, audio, finish) and then keep the + // connection open, never sending a result, so the client reader blocks + // until the caller cancels. + for { + if _, _, err := c.Read(r.Context()); err != nil { + return + } + } + })) + defer srv.Close() + + s, _ := New(WithAppID("app"), WithToken("tok"), WithHost(wsTestURL(srv.URL))) + + ctx, cancel := context.WithCancel(context.Background()) + defer cancel() + out, err := s.RecognizeStream(ctx, loopingInput()) + if err != nil { + t.Fatal(err) + } + + go func() { + time.Sleep(50 * time.Millisecond) + cancel() + }() + for { + if _, err := out.Read(); err != nil { + break + } + } +} diff --git a/voice/stt/bytedance/bytedance_test.go b/voice/stt/bytedance/bytedance_test.go index 4264b28f..e87fed0b 100644 --- a/voice/stt/bytedance/bytedance_test.go +++ b/voice/stt/bytedance/bytedance_test.go @@ -463,6 +463,71 @@ func buildServerFullFrame(data []byte, isLast bool) []byte { return buf } +// buildErrorFrame constructs a server error response frame for testing. +func buildErrorFrame(code uint32, data []byte) []byte { + toc := uint32(protoVersion | protoHdrSize | msgTypeError | serializationJSON | flagNoSeq) + buf := make([]byte, 4+4+4+len(data)) + binary.BigEndian.PutUint32(buf[0:], toc) + binary.BigEndian.PutUint32(buf[4:], code) + binary.BigEndian.PutUint32(buf[8:], uint32(len(data))) + copy(buf[12:], data) + return buf +} + +// TestRecognizeStream_ServerError verifies that an early server-error +// termination surfaces an error to the consumer rather than a clean io.EOF, +// which would make a truncated transcript look complete. +func TestRecognizeStream_ServerError(t *testing.T) { + srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + c, err := websocket.Accept(w, r, nil) + if err != nil { + return + } + defer func() { _ = c.Close(websocket.StatusNormalClosure, "") }() + + // Read full request, then report a server error instead of transcribing. + if _, _, err = c.Read(r.Context()); err != nil { + return + } + errPayload, _ := json.Marshal(asrResponsePayload{Error: "quota exceeded"}) + _ = c.Write(r.Context(), websocket.MessageBinary, buildErrorFrame(1013, errPayload)) + + // Keep the connection open so the client observes the error frame + // before the normal-closure handshake. + _, _, _ = c.Read(r.Context()) + })) + defer srv.Close() + + wsURL := strings.Replace(srv.URL, "http://", "ws://", 1) + s, _ := New(WithAppID("app"), WithToken("tok"), WithHost(wsURL)) + + input := audio.NewPipe[audio.Frame](2) + input.Send(audio.Frame{ + Data: make([]byte, 3200), + Format: audio.Format{SampleRate: 16000, Channels: 1, BitDepth: 16}, + }) + input.Close() + + out, err := s.RecognizeStream(context.Background(), input) + if err != nil { + t.Fatal(err) + } + + var readErr error + for { + if _, e := out.Read(); e != nil { + readErr = e + break + } + } + if readErr == io.EOF { + t.Fatal("server error surfaced as clean EOF; truncated transcript looks complete") + } + if readErr != context.Canceled { + t.Errorf("err = %v, want context.Canceled", readErr) + } +} + func TestUnpackFrame_ServerFull(t *testing.T) { payload := []byte(`{"result":{"text":"hello"}}`) frame := buildServerFullFrame(payload, false) diff --git a/voice/tts/minimax/minimax.go b/voice/tts/minimax/minimax.go index aada31fa..76e95f96 100644 --- a/voice/tts/minimax/minimax.go +++ b/voice/tts/minimax/minimax.go @@ -6,6 +6,7 @@ import ( "context" "encoding/hex" "encoding/json" + "errors" "fmt" "io" "net/http" @@ -16,6 +17,11 @@ import ( "github.com/rs/xid" ) +// errConsumerGone signals that the output pipe's consumer stopped reading +// (out.Send returned false). It is distinguished from provider/decode errors +// so SynthesizeStream can Interrupt (nothing left to drain) instead of Close. +var errConsumerGone = errors.New("minimax tts stream: consumer gone") + const ( defaultBaseURL = "https://api.minimaxi.com" defaultModel = "speech-2.8-hd" @@ -311,13 +317,31 @@ func (t *TTS) SynthesizeStream( out := speechaudio.NewPipe[speechtts.Utterance](16) go func() { + // done is closed when this goroutine returns so the watcher below + // exits instead of leaking. The watcher only interrupts the output + // when the caller's ctx is cancelled, leaving the normal EOF + // (out.Close) and provider-error (out.Interrupt) paths untouched. + done := make(chan struct{}) + defer close(done) go func() { - <-ctx.Done() - out.Interrupt() + select { + case <-ctx.Done(): + out.Interrupt() + case <-done: + } }() seq := 0 for { + // input.Read below ignores ctx, so bail out between sentences + // once the caller cancels. Interrupt directly (the watcher may + // race with done) to guarantee the consumer is unblocked. + select { + case <-ctx.Done(): + out.Interrupt() + return + default: + } text, err := input.Read() if err == io.EOF { out.Close() @@ -328,6 +352,13 @@ func (t *TTS) SynthesizeStream( return } if err := t.synthesizeStreamChunk(ctx, text, o, out, &seq); err != nil { + // Any mid-stream failure — consumer gone, a MiniMax base_resp + // API error, or a response read/decode error — is an abnormal + // termination. Interrupt so the terminal Read returns an error + // per the audio.Stream contract; closing instead would surface + // a clean io.EOF and let direct StreamTTS callers mistake a + // provider failure (or a truncated/empty utterance) for a + // normally completed one. out.Interrupt() return } @@ -354,96 +385,151 @@ func (t *TTS) synthesizeStreamChunk( } defer func() { _ = resp.Body.Close() }() - scanner := bufio.NewScanner(resp.Body) - scanner.Buffer(make([]byte, 0, 64*1024), 1024*1024) - for scanner.Scan() { - line := scanner.Text() - if !strings.HasPrefix(line, "data:") { - continue - } - data := strings.TrimPrefix(line, "data:") - data = strings.TrimSpace(data) - if data == "" { - continue - } - - var respChunk t2aResponse - if err := json.Unmarshal([]byte(data), &respChunk); err != nil { - continue - } - if respChunk.BaseResp != nil && respChunk.BaseResp.StatusCode != 0 { - return fmt.Errorf("minimax tts stream: api error %d: %s", - respChunk.BaseResp.StatusCode, respChunk.BaseResp.StatusMsg) - } - if respChunk.Data == nil || respChunk.Data.Audio == "" { - continue + // Use a bufio.Reader (unbounded line length) instead of a bufio.Scanner: + // the status=2 final SSE event carries the FULL cumulative hex-encoded + // audio, which for longer utterances exceeds Scanner's max-token cap and + // makes Scan() fail with bufio.ErrTooLong, failing the whole synthesis. + // handleStreamLine skips that redundant final payload since all audio has + // already been streamed incrementally via the status!=2 events. + reader := bufio.NewReader(resp.Body) + for { + line, readErr := reader.ReadString('\n') + if err := t.handleStreamLine(strings.TrimSpace(line), text, chunkID, o, out, seq, &firstChunk); err != nil { + return err } - // status=2 is the final event containing the full cumulative audio; - // skip it because we already streamed all incremental chunks. - if respChunk.Data.Status == 2 { - continue - } - - audioBytes, err := hex.DecodeString(respChunk.Data.Audio) - if err != nil { - continue + if readErr != nil { + if readErr == io.EOF { + return nil + } + return fmt.Errorf("minimax tts stream: read response: %w", readErr) } + } +} - effectiveCodec := o.Codec - if effectiveCodec == speechaudio.CodecPCM { - effectiveCodec = speechaudio.CodecMP3 - } +// handleStreamLine parses one SSE line from the T2A streaming response and, +// when it carries an incremental audio chunk, sends it to out. It returns +// errConsumerGone if the consumer stopped reading, a provider error for API +// failures, or nil for lines that should be skipped (non-data lines, the +// redundant status=2 final event, or undecodable payloads). +func (t *TTS) handleStreamLine( + line, text, chunkID string, + o *speechtts.TTSOptions, + out *speechaudio.Pipe[speechtts.Utterance], + seq *int, + firstChunk *bool, +) error { + if !strings.HasPrefix(line, "data:") { + return nil + } + data := strings.TrimSpace(strings.TrimPrefix(line, "data:")) + if data == "" { + return nil + } - // Only the first audio chunk of a sentence carries Text; - // subsequent chunks for the same sentence leave Text empty - // so downstream consumers can treat it as a sync-safe delta. - uttText := "" - if firstChunk { - uttText = text - firstChunk = false - } + var respChunk t2aResponse + if err := json.Unmarshal([]byte(data), &respChunk); err != nil { + return nil + } + if respChunk.BaseResp != nil && respChunk.BaseResp.StatusCode != 0 { + return fmt.Errorf("minimax tts stream: api error %d: %s", + respChunk.BaseResp.StatusCode, respChunk.BaseResp.StatusMsg) + } + if respChunk.Data == nil || respChunk.Data.Audio == "" { + return nil + } + // status=2 is the final event containing the full cumulative audio; + // skip it because we already streamed all incremental chunks. + if respChunk.Data.Status == 2 { + return nil + } - outChunk := speechtts.Utterance{ - Frame: speechaudio.Frame{ - Data: audioBytes, - Format: speechaudio.Format{ - Codec: effectiveCodec, - SampleRate: o.Rate, - Channels: 1, - BitDepth: 0, - }, + audioBytes, err := hex.DecodeString(respChunk.Data.Audio) + if err != nil { + return nil + } + + // Label the frame with the codec of the bytes MiniMax actually returned + // (e.g. an opus request yields mp3 bytes, a wav request yields flac), not + // the requested codec, so downstream decoders keyed off Format.Codec decode + // the real audio rather than mis-decoding it. + output := resolveOutput(o.Codec) + + // Only the first audio chunk of a sentence carries Text; + // subsequent chunks for the same sentence leave Text empty + // so downstream consumers can treat it as a sync-safe delta. + uttText := "" + if *firstChunk { + uttText = text + *firstChunk = false + } + + outChunk := speechtts.Utterance{ + Frame: speechaudio.Frame{ + Data: audioBytes, + Format: speechaudio.Format{ + Codec: output.codec, + SampleRate: effectiveSampleRate(o), + Channels: 1, + // BitDepth is left 0: MiniMax's mp3/flac output is a + // compressed stream whose sample bit depth is only defined + // after decoding, so we cannot state it here without guessing. + BitDepth: 0, }, - Text: uttText, - ChunkID: chunkID, - Sequence: *seq, - } - *seq++ - if !out.Send(outChunk) { - return ctx.Err() - } + }, + Text: uttText, + ChunkID: chunkID, + Sequence: *seq, } - - return scanner.Err() + *seq++ + if !out.Send(outChunk) { + return errConsumerGone + } + return nil } // --- helpers --- -// codecToMinimaxFormat maps speechaudio.Codec to the MiniMax API format string. -func codecToMinimaxFormat(c speechaudio.Codec) string { +// minimaxOutput describes the audio MiniMax actually returns for a requested +// codec: the format string sent in audio_setting.format and the codec of the +// bytes the API produces. MiniMax does not support every codec, so the +// requested codec and the produced bytes can differ. Keeping both fields +// together is the single source of truth so the format sent to the API and the +// Format.Codec label stamped on emitted frames can never diverge. +type minimaxOutput struct { + format string + codec speechaudio.Codec +} + +// resolveOutput maps a requested codec to the MiniMax API format string and the +// codec of the bytes the API actually returns for that format. +func resolveOutput(c speechaudio.Codec) minimaxOutput { switch c { case speechaudio.CodecMP3: - return "mp3" + return minimaxOutput{format: "mp3", codec: speechaudio.CodecMP3} case speechaudio.CodecOpus: - return "mp3" // MiniMax does not support opus; fallback to mp3 + // MiniMax does not support opus; it returns mp3 bytes. + return minimaxOutput{format: "mp3", codec: speechaudio.CodecMP3} case speechaudio.CodecWAV: - return "flac" + // MiniMax returns flac bytes for a "flac" format request. + return minimaxOutput{format: "flac", codec: speechaudio.CodecFLAC} case speechaudio.CodecPCM: fallthrough default: - return "mp3" + // MiniMax has no raw PCM output; it returns mp3 bytes. + return minimaxOutput{format: "mp3", codec: speechaudio.CodecMP3} } } +// effectiveSampleRate returns the sample rate MiniMax is asked to encode at. +// It must match the value sent in audio_setting.sample_rate so the rate stamped +// on emitted frames describes the actual audio. +func effectiveSampleRate(o *speechtts.TTSOptions) int { + if o.Rate <= 0 { + return 32000 + } + return o.Rate +} + func (t *TTS) buildRequest(text string, stream bool, o *speechtts.TTSOptions) *t2aRequest { voiceID := t.voiceID if o.Voice != "" { @@ -463,15 +549,9 @@ func (t *TTS) buildRequest(text string, stream bool, o *speechtts.TTSOptions) *t Emotion: o.ExtraString(KeyEmotion, t.emotion), } - format := codecToMinimaxFormat(o.Codec) - rate := o.Rate - if rate <= 0 { - rate = 32000 - } - as := &audioSetting{ - SampleRate: rate, - Format: format, + SampleRate: effectiveSampleRate(o), + Format: resolveOutput(o.Codec).format, Channel: 1, } @@ -660,8 +740,17 @@ func (t *TTS) Warmup(ctx context.Context) error { if err != nil { return fmt.Errorf("minimax tts: warmup: %w", err) } + // Drain the body before closing so net/http's Transport can return the + // connection to the idle pool for reuse — the whole point of warmup. A + // bare Close on an unread body discards the connection instead. + _, _ = io.Copy(io.Discard, resp.Body) _ = resp.Body.Close() - if resp.StatusCode >= 400 { + // The TCP/TLS handshake — the expensive part warmup pre-heats — is complete + // once we have any HTTP response. Posting "{}" to the T2A endpoint typically + // yields HTTP 400, so only treat 5xx as a warmup failure; a client-side + // (4xx) status still means the connection is established and pooled. A real + // connection error would already have been returned above. + if resp.StatusCode >= 500 { return fmt.Errorf("minimax tts: warmup: HTTP %d", resp.StatusCode) } return nil diff --git a/voice/tts/minimax/minimax_leak_test.go b/voice/tts/minimax/minimax_leak_test.go new file mode 100644 index 00000000..fc3086bb --- /dev/null +++ b/voice/tts/minimax/minimax_leak_test.go @@ -0,0 +1,203 @@ +package minimax + +import ( + "context" + "encoding/hex" + "encoding/json" + "errors" + "io" + "net/http" + "net/http/httptest" + "testing" + "time" + + speechaudio "github.com/GizClaw/flowcraft/voice/audio" + "go.uber.org/goleak" +) + +// newLeakTestTTS builds a TTS whose HTTP client uses a dedicated Transport so +// each test can deterministically close its idle connections before goleak +// verification (the default http.DefaultClient pools connections globally, +// whose reaper goroutines would otherwise be reported as false positives). +func newLeakTestTTS(t *testing.T, baseURL string) (*TTS, *http.Transport) { + t.Helper() + p, err := New(WithAPIKey("test-key"), WithBaseURL(baseURL)) + if err != nil { + t.Fatal(err) + } + tr := &http.Transport{} + p.client = &http.Client{Transport: tr} + return p, tr +} + +func newSSEServer(handler func(w http.ResponseWriter, flush func())) *httptest.Server { + return httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + flusher, _ := w.(http.Flusher) + flush := func() { + if flusher != nil { + flusher.Flush() + } + } + handler(w, flush) + })) +} + +func writeSSE(w http.ResponseWriter, flush func(), ev t2aResponse) { + data, _ := json.Marshal(ev) + _, _ = w.Write([]byte("data: " + string(data) + "\n\n")) + flush() +} + +// TestSynthesizeStream_NoLeak_NormalCompletion locks in the fix that the +// ctx-watcher goroutine and synth loop exit on normal completion (input closed +// -> EOF -> output drained to EOF). +func TestSynthesizeStream_NoLeak_NormalCompletion(t *testing.T) { + defer goleak.VerifyNone(t) // runs last (see defer ordering below) + + chunk1 := hex.EncodeToString([]byte("chunk-1")) + chunk2 := hex.EncodeToString([]byte("chunk-2")) + srv := newSSEServer(func(w http.ResponseWriter, flush func()) { + writeSSE(w, flush, t2aResponse{Data: &t2aData{Audio: chunk1, Status: 1}, BaseResp: &baseResp{StatusCode: 0}}) + writeSSE(w, flush, t2aResponse{Data: &t2aData{Audio: chunk2, Status: 1}, BaseResp: &baseResp{StatusCode: 0}}) + writeSSE(w, flush, t2aResponse{Data: &t2aData{Audio: "", Status: 2}, BaseResp: &baseResp{StatusCode: 0, StatusMsg: "success"}}) + }) + defer srv.Close() + + p, tr := newLeakTestTTS(t, srv.URL) + defer tr.CloseIdleConnections() + + textPipe := speechaudio.NewPipe[string](1) + textPipe.Send("hello") + textPipe.Close() + + stream, err := p.SynthesizeStream(context.Background(), textPipe) + if err != nil { + t.Fatal(err) + } + + var chunks int + for { + _, err := stream.Read() + if err == io.EOF { + break + } + if err != nil { + t.Fatalf("read: %v", err) + } + chunks++ + } + if chunks != 2 { + t.Fatalf("got %d chunks, want 2", chunks) + } +} + +// TestSynthesizeStream_NoLeak_CtxCancelledMidStream locks in the fix that a +// caller-ctx cancellation mid-stream tears down both goroutines instead of +// leaking them, and that teardown surfaces via the ctx-watcher (context.Canceled) +// rather than only via an eventual transport abort. +// +// The SSE handler sends exactly one chunk then BLOCKS (never sends more, never +// returns) until test cleanup, so the worker is genuinely parked mid-stream — +// blocked reading the next SSE line — when the caller ctx is cancelled. The +// load-bearing production fix here is the done-channel watcher goroutine in +// SynthesizeStream: on ctx.Done() it calls out.Interrupt(), which unblocks the +// consumer's Read with context.Canceled promptly. If that watcher is removed, +// the consumer never observes context.Canceled (it would instead see a delayed +// io.EOF from the transport-abort path, or block), so the assertion below fails. +func TestSynthesizeStream_NoLeak_CtxCancelledMidStream(t *testing.T) { + defer goleak.VerifyNone(t) + + chunk := hex.EncodeToString([]byte("chunk-1")) + // release unblocks the handler at cleanup so the server goroutine never + // outlives the test (otherwise goleak would flag the parked handler). + release := make(chan struct{}) + srv := newSSEServer(func(w http.ResponseWriter, flush func()) { + // Deliver exactly one chunk, then park so the worker stays genuinely + // mid-stream (blocked on the next SSE line) until the test releases it. + writeSSE(w, flush, t2aResponse{Data: &t2aData{Audio: chunk, Status: 1}, BaseResp: &baseResp{StatusCode: 0}}) + <-release + }) + // Ordering (LIFO): CloseIdleConnections -> close(release) -> srv.Close -> + // goleak.VerifyNone. close(release) lets the parked handler return so + // srv.Close() can join it before the leak check runs. + defer srv.Close() + defer close(release) + + p, tr := newLeakTestTTS(t, srv.URL) + defer tr.CloseIdleConnections() + + ctx, cancel := context.WithCancel(context.Background()) + textPipe := speechaudio.NewPipe[string](1) + textPipe.Send("hello") + textPipe.Close() + + stream, err := p.SynthesizeStream(ctx, textPipe) + if err != nil { + t.Fatal(err) + } + + // Read the first delivered chunk, confirming the worker is now mid-stream. + if _, err := stream.Read(); err != nil { + t.Fatalf("first read: %v", err) + } + + // Cancel mid-stream. The ctx-watcher must interrupt the output promptly. + cancel() + + errCh := make(chan error, 1) + go func() { + var last error + for { + if _, err := stream.Read(); err != nil { + last = err + break + } + } + errCh <- last + }() + + select { + case err := <-errCh: + if !errors.Is(err, context.Canceled) { + t.Fatalf("terminal error = %v, want context.Canceled (ctx-watcher must interrupt mid-stream)", err) + } + case <-time.After(2 * time.Second): + t.Fatal("stream did not terminate promptly after ctx cancel (worker leaked mid-stream)") + } +} + +// TestSynthesizeStream_NoLeak_ProviderError locks in the fix that a provider +// error mid-stream tears down the goroutines (here the server returns a +// base_resp error after some audio). +func TestSynthesizeStream_NoLeak_ProviderError(t *testing.T) { + defer goleak.VerifyNone(t) + + chunk := hex.EncodeToString([]byte("chunk-1")) + srv := newSSEServer(func(w http.ResponseWriter, flush func()) { + writeSSE(w, flush, t2aResponse{Data: &t2aData{Audio: chunk, Status: 1}, BaseResp: &baseResp{StatusCode: 0}}) + // Provider error frame mid-stream. + writeSSE(w, flush, t2aResponse{BaseResp: &baseResp{StatusCode: 1002, StatusMsg: "rate limited"}}) + }) + defer srv.Close() + + p, tr := newLeakTestTTS(t, srv.URL) + defer tr.CloseIdleConnections() + + textPipe := speechaudio.NewPipe[string](1) + textPipe.Send("hello") + textPipe.Close() + + stream, err := p.SynthesizeStream(context.Background(), textPipe) + if err != nil { + t.Fatal(err) + } + + // Drain to the terminal error. The provider-error path interrupts the + // output, so the terminal Read reports context.Canceled (abnormal + // termination); this test only asserts no goroutine leaks on that path. + for { + if _, err := stream.Read(); err != nil { + break + } + } +} diff --git a/voice/tts/minimax/minimax_test.go b/voice/tts/minimax/minimax_test.go index 584cbcae..22d82081 100644 --- a/voice/tts/minimax/minimax_test.go +++ b/voice/tts/minimax/minimax_test.go @@ -1,6 +1,7 @@ package minimax import ( + "bytes" "context" "encoding/hex" "encoding/json" @@ -288,6 +289,203 @@ func TestSynthesizeStream(t *testing.T) { } } +func TestSynthesizeStream_FormatCodecMatchesOutput(t *testing.T) { + // MiniMax may return bytes in a codec other than the one requested + // (opus -> mp3, wav -> flac, pcm -> mp3). The Format.Codec label on emitted + // frames must describe the bytes the API actually returned, and the format + // string sent to the API must correspond to that same codec. + cases := []struct { + name string + reqCodec speechaudio.Codec + wantFormat string + wantCodec speechaudio.Codec + }{ + {"pcm->mp3", speechaudio.CodecPCM, "mp3", speechaudio.CodecMP3}, + {"opus->mp3", speechaudio.CodecOpus, "mp3", speechaudio.CodecMP3}, + {"wav->flac", speechaudio.CodecWAV, "flac", speechaudio.CodecFLAC}, + {"mp3->mp3", speechaudio.CodecMP3, "mp3", speechaudio.CodecMP3}, + } + + for _, tc := range cases { + t.Run(tc.name, func(t *testing.T) { + chunk := hex.EncodeToString([]byte("audio")) + var gotFormat string + srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + var req t2aRequest + if err := json.NewDecoder(r.Body).Decode(&req); err != nil { + t.Fatalf("decode request: %v", err) + } + gotFormat = req.AudioSetting.Format + + flusher, _ := w.(http.Flusher) + events := []t2aResponse{ + {Data: &t2aData{Audio: chunk, Status: 1}, BaseResp: &baseResp{StatusCode: 0}}, + {Data: &t2aData{Audio: "", Status: 2}, BaseResp: &baseResp{StatusCode: 0}}, + } + for _, ev := range events { + data, _ := json.Marshal(ev) + if _, err := w.Write([]byte("data: " + string(data) + "\n\n")); err != nil { + t.Fatalf("write sse: %v", err) + } + flusher.Flush() + } + })) + defer srv.Close() + + p, _ := New(WithAPIKey("k"), WithBaseURL(srv.URL)) + textPipe := speechaudio.NewPipe[string](1) + textPipe.Send("测试") + textPipe.Close() + + stream, err := p.SynthesizeStream(context.Background(), textPipe, speechtts.WithCodec(tc.reqCodec)) + if err != nil { + t.Fatal(err) + } + frame, err := stream.Read() + if err != nil { + t.Fatalf("read: %v", err) + } + // Drain the stream. + for { + if _, err := stream.Read(); err == io.EOF { + break + } else if err != nil { + t.Fatalf("drain: %v", err) + } + } + + if gotFormat != tc.wantFormat { + t.Errorf("api format = %q, want %q", gotFormat, tc.wantFormat) + } + if frame.Format.Codec != tc.wantCodec { + t.Errorf("frame codec = %v, want %v", frame.Format.Codec, tc.wantCodec) + } + }) + } +} + +// TestSynthesizeStream_ProviderErrorMidStreamSignalsError guards the stream +// contract for abnormal termination: when MiniMax returns a base_resp error +// after some audio has already streamed, the terminal Read must report an error +// (context.Canceled from Interrupt), never a clean io.EOF. Surfacing io.EOF +// would let a direct StreamTTS caller mistake a mid-stream provider failure for +// a normally completed utterance. +func TestSynthesizeStream_ProviderErrorMidStreamSignalsError(t *testing.T) { + chunk := hex.EncodeToString([]byte("chunk-1")) + srv := newSSEServer(func(w http.ResponseWriter, flush func()) { + writeSSE(w, flush, t2aResponse{Data: &t2aData{Audio: chunk, Status: 1}, BaseResp: &baseResp{StatusCode: 0}}) + writeSSE(w, flush, t2aResponse{BaseResp: &baseResp{StatusCode: 1002, StatusMsg: "rate limited"}}) + }) + defer srv.Close() + + p, _ := New(WithAPIKey("k"), WithBaseURL(srv.URL)) + textPipe := speechaudio.NewPipe[string](1) + textPipe.Send("hello") + textPipe.Close() + + stream, err := p.SynthesizeStream(context.Background(), textPipe) + if err != nil { + t.Fatal(err) + } + + var termErr error + for { + if _, err := stream.Read(); err != nil { + termErr = err + break + } + } + if termErr == io.EOF { + t.Fatal("mid-stream provider error surfaced as clean io.EOF; caller cannot tell failure from normal completion") + } + if termErr != context.Canceled { + t.Fatalf("terminal error = %v, want context.Canceled", termErr) + } +} + +func TestSynthesizeStream_LargeFinalStatus2Line(t *testing.T) { + // Issue #4: production reads the SSE stream line-by-line. The original + // bufio.Scanner imposes a max-token cap (even when bumped to 1MB via + // scanner.Buffer), so any single `data:` line longer than that cap makes + // Scan() fail with bufio.ErrTooLong — aborting the read loop and silently + // dropping every remaining chunk on the stream. The fix switched to + // bufio.Reader, whose ReadString has no line-length cap. + // + // To catch a regression back to Scanner, the oversized line MUST sit on an + // INCREMENTAL audio event (status != 2 — the branch that decodes + Sends), + // followed by a further small incremental chunk. Under Scanner the >1MB + // line trips ErrTooLong: the large chunk is never decoded/sent AND the + // trailing small chunk is lost. Under Reader both are delivered. Putting + // the big payload on the skipped status=2 final line (as an earlier version + // did) is indistinguishable between the two implementations, so this test + // deliberately does not do that. + small := []byte("small-trailing-chunk") + smallHex := hex.EncodeToString(small) + + // A > 1 MiB raw payload => > 2 MiB hex on the wire, well past both the + // default 64KB Scanner cap and a 1MB scanner.Buffer cap. Fill with a + // recognizable byte so the decoded bytes can be asserted exactly. + bigAudio := make([]byte, 1<<20+512) + for i := range bigAudio { + bigAudio[i] = 0xAB + } + bigHex := hex.EncodeToString(bigAudio) + + srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + flusher, _ := w.(http.Flusher) + write := func(ev t2aResponse) { + data, _ := json.Marshal(ev) + if _, err := w.Write([]byte("data: " + string(data) + "\n\n")); err != nil { + t.Errorf("write sse: %v", err) + } + flusher.Flush() + } + // Oversized INCREMENTAL chunk first, then a small incremental chunk, + // then the redundant status=2 final (skipped by production). + write(t2aResponse{Data: &t2aData{Audio: bigHex, Status: 1}, BaseResp: &baseResp{StatusCode: 0}}) + write(t2aResponse{Data: &t2aData{Audio: smallHex, Status: 1}, BaseResp: &baseResp{StatusCode: 0}}) + write(t2aResponse{Data: &t2aData{Audio: "", Status: 2}, BaseResp: &baseResp{StatusCode: 0, StatusMsg: "success"}}) + })) + defer srv.Close() + + p, _ := New(WithAPIKey("test-key"), WithBaseURL(srv.URL)) + textPipe := speechaudio.NewPipe[string](1) + textPipe.Send("测试大行") + textPipe.Close() + + stream, err := p.SynthesizeStream(context.Background(), textPipe) + if err != nil { + t.Fatal(err) + } + + var got [][]byte + var termErr error + for { + u, err := stream.Read() + if err != nil { + termErr = err + break + } + got = append(got, append([]byte(nil), u.Data...)) + } + + if termErr != io.EOF { + t.Fatalf("terminal error = %v, want io.EOF (large incremental line must not fail synthesis)", termErr) + } + // Both incremental chunks must be delivered: the >1MB one AND the small + // one after it. Under the old bufio.Scanner the oversized line trips + // ErrTooLong and BOTH are lost, so this count assertion fails. + if len(got) != 2 { + t.Fatalf("got %d chunks, want 2 (>1MB incremental + trailing small incremental)", len(got)) + } + if !bytes.Equal(got[0], bigAudio) { + t.Errorf("large chunk: got %d bytes, want %d bytes with exact payload match", len(got[0]), len(bigAudio)) + } + if string(got[1]) != string(small) { + t.Errorf("trailing chunk = %q, want %q", string(got[1]), string(small)) + } +} + func TestBuildRequest_FormatMapping(t *testing.T) { p, _ := New(WithAPIKey("k")) @@ -337,6 +535,35 @@ func TestWarmup(t *testing.T) { } } +func TestWarmup_StatusHandling(t *testing.T) { + // A completed handshake is warmup's goal, so a client-side (4xx) status — + // which posting "{}" typically produces — must count as success, while a + // server-side (5xx) status is reported as a failure. + cases := []struct { + status int + wantErr bool + }{ + {http.StatusOK, false}, + {http.StatusBadRequest, false}, + {http.StatusInternalServerError, true}, + } + for _, tc := range cases { + srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + w.WriteHeader(tc.status) + _, _ = w.Write([]byte(`{"base_resp":{"status_code":1004}}`)) + })) + p, _ := New(WithAPIKey("k"), WithBaseURL(srv.URL)) + err := p.Warmup(context.Background()) + if tc.wantErr && err == nil { + t.Errorf("status %d: expected error, got nil", tc.status) + } + if !tc.wantErr && err != nil { + t.Errorf("status %d: unexpected error: %v", tc.status, err) + } + srv.Close() + } +} + func TestRegistration(t *testing.T) { providers := speechtts.ListTTSProviders() found := false diff --git a/voice/webrtc/export_test.go b/voice/webrtc/export_test.go index d51369ad..0f7d3a42 100644 --- a/voice/webrtc/export_test.go +++ b/voice/webrtc/export_test.go @@ -1,6 +1,14 @@ package webrtc -import "github.com/pion/webrtc/v4/pkg/media" +import ( + "reflect" + "sync/atomic" + "testing" + "unsafe" + + "github.com/pion/webrtc/v4" + "github.com/pion/webrtc/v4/pkg/media" +) // NewSinkForTest creates a Sink with the given track and encoder. // Exported only for testing via the _test build tag. @@ -8,6 +16,57 @@ func NewSinkForTest(track interface{ WriteSample(media.Sample) error }, encoder return newSink(track, encoder) } +// InitPCForTest runs the real initPC so the production OnConnectionStateChange +// handler is registered on a live PeerConnection, without performing an SDP +// handshake. Exported only for testing. +func (t *Transport) InitPCForTest() error { + t.mu.Lock() + defer t.mu.Unlock() + return t.initPC(false) +} + +// SimulateConnectionState invokes the REAL OnConnectionStateChange handler that +// initPC registered on the underlying pion PeerConnection, passing the given +// state. This lets tests exercise the production state -> source-interrupt +// decision (only Failed/Closed interrupt; Disconnected is transient) directly, +// without provoking a real ICE failure. +// +// The handler is an unexported closure stored in an unexported atomic.Value +// field of pion's PeerConnection with no public trigger, so it is read via +// reflection. This is deterministic (not flaky) but pinned to the pion version +// in go.mod; if pion renames that field this shim breaks. +// +// If the reflection shim cannot resolve the handler (e.g. a pion upgrade +// renamed/moved the field), it returns nil. Silently skipping the call would +// make every connection-state test pass VACUOUSLY — in particular the +// Disconnected test, whose success condition is merely "the source pipe was NOT +// interrupted", would then pass even though the production handler never ran. +// To make a broken shim fail loudly instead of masquerading as a green test, +// this fatals when the handler is nil. +func (t *Transport) SimulateConnectionState(tb testing.TB, state webrtc.PeerConnectionState) { + tb.Helper() + h := connectionStateHandler(t.pc) + if h == nil { + tb.Fatalf("SimulateConnectionState: connectionStateHandler returned nil — the pion " + + "reflection shim is broken (field renamed/moved after a pion upgrade?). " + + "Connection-state tests cannot run and must not pass vacuously.") + } + h(state) +} + +func connectionStateHandler(pc *webrtc.PeerConnection) func(webrtc.PeerConnectionState) { + if pc == nil { + return nil + } + f := reflect.ValueOf(pc).Elem().FieldByName("onConnectionStateChangeHandler") + if !f.IsValid() || !f.CanAddr() { + return nil + } + av := (*atomic.Value)(unsafe.Pointer(f.UnsafeAddr())) + h, _ := av.Load().(func(webrtc.PeerConnectionState)) + return h +} + // NotifyTrackReady exposes the trackReady signal for concurrency testing. func (t *Transport) NotifyTrackReady() { t.trackReadyOnce.Do(func() { close(t.trackReady) }) diff --git a/voice/webrtc/transport.go b/voice/webrtc/transport.go index eac4e7d1..4ab48cc3 100644 --- a/voice/webrtc/transport.go +++ b/voice/webrtc/transport.go @@ -271,8 +271,13 @@ func (t *Transport) initPC(asOfferer bool) error { if t.config.OnConnectionStateChange != nil { t.config.OnConnectionStateChange(cs) } + // Only tear down on terminal states. Disconnected is a TRANSIENT + // ICE state that routinely flaps on packet loss and can recover to + // connected; interrupting here would latch context.Canceled on the + // source pipe (see audio.Pipe.Read) and irrecoverably kill an + // otherwise-recoverable session. Callers still observe Disconnected + // via OnConnectionStateChange above. if state == webrtc.PeerConnectionStateFailed || - state == webrtc.PeerConnectionStateDisconnected || state == webrtc.PeerConnectionStateClosed { t.source.pipe.Interrupt() } diff --git a/voice/webrtc/transport_state_test.go b/voice/webrtc/transport_state_test.go new file mode 100644 index 00000000..c9e59e9b --- /dev/null +++ b/voice/webrtc/transport_state_test.go @@ -0,0 +1,89 @@ +package webrtc_test + +import ( + "context" + "errors" + "testing" + "time" + + "github.com/GizClaw/flowcraft/voice/audio" + "github.com/pion/webrtc/v4" +) + +// readErrAsync reads one value from the stream in a goroutine and reports the +// resulting error. The goroutine is unblocked at test cleanup (tr.Close +// interrupts the source pipe) when the stream is never interrupted by the state +// change under test. +func readErrAsync(s audio.Stream[audio.Frame]) <-chan error { + ch := make(chan error, 1) + go func() { + _, err := s.Read() + ch <- err + }() + return ch +} + +// TestTransport_ConnectionStateDisconnected_DoesNotInterruptSource locks in the +// issue #2 fix: PeerConnectionStateDisconnected is a transient ICE state that +// routinely flaps and can recover, so it must NOT interrupt the source pipe (an +// interrupt would latch context.Canceled and kill a recoverable session). +func TestTransport_ConnectionStateDisconnected_DoesNotInterruptSource(t *testing.T) { + tr := newTransport(t) + if err := tr.InitPCForTest(); err != nil { + t.Fatalf("InitPCForTest: %v", err) + } + + tr.SimulateConnectionState(t, webrtc.PeerConnectionStateDisconnected) + + errCh := readErrAsync(tr.Source().Stream()) + select { + case err := <-errCh: + t.Fatalf("Disconnected interrupted the source pipe (Read returned %v); "+ + "it is transient and must not interrupt", err) + case <-time.After(300 * time.Millisecond): + // Still blocking with no data => not interrupted, as required. + // t.Cleanup(tr.Close) unblocks the pending Read. + } +} + +// TestTransport_ConnectionStateFailed_InterruptsSourcePipe locks in that the +// terminal Failed state DOES interrupt the source pipe. +func TestTransport_ConnectionStateFailed_InterruptsSourcePipe(t *testing.T) { + tr := newTransport(t) + if err := tr.InitPCForTest(); err != nil { + t.Fatalf("InitPCForTest: %v", err) + } + + tr.SimulateConnectionState(t, webrtc.PeerConnectionStateFailed) + + errCh := readErrAsync(tr.Source().Stream()) + select { + case err := <-errCh: + if !errors.Is(err, context.Canceled) { + t.Fatalf("Failed: Read err = %v, want context.Canceled", err) + } + case <-time.After(2 * time.Second): + t.Fatal("Failed did not interrupt the source pipe") + } +} + +// TestTransport_ConnectionStateClosed_InterruptsSourcePipe locks in that the +// terminal Closed state DOES interrupt the source pipe. +func TestTransport_ConnectionStateClosed_InterruptsSourcePipe(t *testing.T) { + tr := newTransport(t) + if err := tr.InitPCForTest(); err != nil { + t.Fatalf("InitPCForTest: %v", err) + } + + tr.SimulateConnectionState(t, webrtc.PeerConnectionStateClosed) + + errCh := readErrAsync(tr.Source().Stream()) + select { + case err := <-errCh: + if !errors.Is(err, context.Canceled) { + t.Fatalf("Closed: Read err = %v, want context.Canceled", err) + } + case <-time.After(2 * time.Second): + t.Fatal("Closed did not interrupt the source pipe") + } +}