fix(voice): harden streaming concurrency, provider resilience, and codec labeling#226
Conversation
Add audio.CodecFLAC so streamed audio can be labeled to match the bytes a provider actually returns. Make the ResampleStream worker check the output pipe's context before each blocking input.Read so it exits promptly when the consumer interrupts, and document that the upstream must be interruptible. Co-authored-by: Cursor <cursoragent@cursor.com>
…-aware Fail fast on a cancelled context before trying each provider and before each retry attempt; previously a retryable, ctx-ignoring execFn could be retried MaxAttempts times after cancellation when RetryBackoff==0. SleepWithContext now honors ctx even for non-positive durations, and the clock is recomputed per provider so a breaker that has actually cooled down is not skipped. Add a half-open state so recovery is gradual and consecutiveFailures stays bounded, count provider-attributable failures (not only retryable ones) toward the breaker, and exclude client-fault errors (bad input, caller cancellation) via IsProviderFault so bad client input can never open a healthy provider. Co-authored-by: Cursor <cursoragent@cursor.com>
…ntics Make the writer goroutine observe inner-context cancellation before each input.Read so it exits when the reader returns early (server error frame or ctx cancel) instead of leaking while parked on the caller-owned input stream. Surface genuine writer send failures via out.Interrupt() instead of a clean io.EOF so a truncated transcript is no longer mistaken for a complete one, and honor out.Send()'s result consistently. Adds go.uber.org/goleak as a test dependency for the leak regression tests. Co-authored-by: Cursor <cursoragent@cursor.com>
…ec labels Stop the SynthesizeStream ctx-watcher goroutine from leaking on the normal and error return paths (exit via a done channel) and bail out of the synth loop promptly on ctx cancellation. Replace the 1MB-capped bufio.Scanner with a bufio.Reader so a long utterance's cumulative status=2 final line no longer fails with ErrTooLong, and skip that redundant cumulative line. Centralize request-codec -> (API format, output codec) mapping in resolveOutput so an emitted Frame.Format matches the bytes the API actually returns (Opus -> mp3, WAV -> flac), and drain the warmup response body before closing so the connection returns to the idle pool. Co-authored-by: Cursor <cursoragent@cursor.com>
Only tear down the source pipe on PeerConnectionStateFailed/Closed, not on the transient Disconnected state, which routinely flaps on packet loss and can recover to connected. Because Pipe.Interrupt latches context.Canceled permanently, interrupting on Disconnected irrecoverably killed otherwise recoverable sessions. Co-authored-by: Cursor <cursoragent@cursor.com>
![chatgpt-codex-connector[bot]](https://avatars.githubusercontent.com/in/1144995?s=60&v=4){kind=small}
There was a problem hiding this comment.
💡 Codex Review
Here are some automated review suggestions for this pull request.
Reviewed commit: 3733acbe29
ℹ️ About Codex in GitHub
Codex has been enabled to automatically review pull requests in this repo. Reviews are triggered when you
- Open a pull request for review
- Mark a draft as ready
- Comment "@codex review".
If Codex has suggestions, it will comment; otherwise it will react with 👍.
When you sign up for Codex through ChatGPT, Codex can also answer questions or update the PR, like "@codex address that feedback".
| // 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() |
There was a problem hiding this comment.
On this writer-error path the goroutine still has defer out.Close() at the top of RecognizeStream, so returning after out.Interrupt() immediately closes the pipe too. audio.Pipe.Read is nondeterministic when both cancel and close are ready (see TestPipe_InterruptThenClose), so callers that read after the goroutine exits can cache io.EOF and still treat a truncated upload as a clean transcript; avoid closing the pipe on interrupt/error paths or otherwise make the terminal error deterministic.
Useful? React with 👍 / 👎.
| // loses no error signal that callers rely on, and | ||
| // FallbackTTS only falls back on stream-setup failure, | ||
| // not on mid-stream errors. | ||
| out.Close() |
There was a problem hiding this comment.
This branch handles synthesizeStreamChunk errors such as MiniMax base_resp API errors or response read failures, but out.Close() makes the next Read end with io.EOF when no audio, or only partial audio, was produced. Direct StreamTTS callers and the pipeline cannot distinguish provider failure or rate limiting from a normal empty/completed utterance, violating the stream contract for abnormal termination; preserve an interrupt/error signal instead of closing cleanly for provider/decode errors.
Useful? React with 👍 / 👎.
| if state.openUntil.IsZero() { | ||
| // Closed: no open window pending. | ||
| return true |
There was a problem hiding this comment.
After the open window elapses, the first caller clears openUntil and sets halfOpen; any concurrent or subsequent Allow before that probe finishes re-enters here and returns true because openUntil is now zero. That defeats the new half-open invariant of allowing a single recovery probe, so a burst of requests can hit a still-unhealthy provider immediately after cooldown instead of being held until OnSuccess or OnFailure resolves the probe.
Useful? React with 👍 / 👎.
…ipe.Read Pipe.Read's select was chosen at random when both a value/closed channel and ctx.Done() were ready, so a producer that calls Interrupt() and then runs a deferred Close() (e.g. bytedance RecognizeStream's server-error and writer-error paths) could have its abnormal termination observed as a clean io.EOF. Check ctx.Done() first so "Interrupt returns context.Canceled immediately, skipping buffered values" is the deterministic guarantee the docs already promise. Co-authored-by: Cursor <cursoragent@cursor.com>
When the open window elapsed, the first Allow zeroed openUntil and set halfOpen, but any concurrent or subsequent Allow then saw a zero window and returned true, letting a post-cooldown burst hit a still-unhealthy provider. Distinguish a half-open probe in flight from a fully closed circuit and hold other callers until OnSuccess/OnFailure resolves the probe, restoring the single-probe invariant. Co-authored-by: Cursor <cursoragent@cursor.com>
…clean EOF A mid-stream failure (MiniMax base_resp API error or a response read/decode error) closed the output, so the terminal Read returned io.EOF and a direct SynthesizeStream caller could not distinguish a provider failure or a truncated utterance from a normally completed one. Interrupt on any mid-stream failure so the terminal Read reports an error, honoring the audio.Stream abnormal- termination contract. Co-authored-by: Cursor <cursoragent@cursor.com>
1 participant
Summary
Full-review pass over the
voicepackage, fixing concurrency hazards, goroutine leaks, stream-termination semantics, and provider-resilience bugs surfaced while reviewingmain. No public API changes; every fix ships with a regression test (goroutine-leak checks usego.uber.org/goleak, and each hardened test was mutation-verified to fail when its fix is reverted).Grouped into five scoped commits:
CodecFLAC; make theResampleStreamworker check its output context before each blockinginput.Readso it exits promptly on consumer interrupt.SleepWithContexthonors ctx for non-positive durations; recompute the clock per provider so a cooled-down breaker isn't skipped; add a half-open state; count provider-attributable failures toward the breaker and exclude client faults viaIsProviderFaultso bad client input can't open a healthy provider.input.Read(no leak when the reader returns early); surface genuine send failures viaInterrupt()instead of a clean EOF so a truncated transcript isn't mistaken for a complete one.bufio.Scannerwithbufio.Readerso a long utterance's cumulativestatus=2final line no longer fails withErrTooLong; centralize codec mapping inresolveOutputsoFrame.Formatmatches the bytes the API actually returns; drain the warmup body before closing.Failed/Closed, not on transientDisconnected(which flaps on packet loss and can recover); previously the latchedInterruptirrecoverably killed recoverable sessions.Test plan
go build ./...andgo vet ./...(voice module)go test -race ./...(voice module)Made with Cursor