From edc20d57982ff8c01692bbac89d36bc2bcd3e410 Mon Sep 17 00:00:00 2001 From: phyce Date: Mon, 6 Jul 2026 10:52:40 +0000 Subject: [PATCH] Make audio stream apply real backpressure instead of erroring StreamWriter.Write returned ErrStreamLimitReached once the elastic queue passed 64 MiB, which aborted valid long syntheses that buffer ahead via dispatchFutures. Write now blocks on the cond until the consumer drains space, and only returns an error if the stream is closed/errored while waiting. Tests updated for the new semantic. Co-Authored-By: Claude Opus 4.8 (1M context) --- app/common/audio/stream.go | 32 ++++++++++---- app/common/audio/stream_test.go | 75 +++++++++++++++++++++++++++++++-- 2 files changed, 94 insertions(+), 13 deletions(-) diff --git a/app/common/audio/stream.go b/app/common/audio/stream.go index 2a3e72a..3003b38 100644 --- a/app/common/audio/stream.go +++ b/app/common/audio/stream.go @@ -7,13 +7,14 @@ import ( "sync" ) -// Producers never block: the queue is elastic so generation can run ahead of -// playback. The limit only guards against a consumer that never reads. +// The queue is elastic so generation can run ahead of playback, but it is +// bounded by streamQueueLimit as a backpressure high-water mark: once the +// queued bytes reach the limit, Write blocks until the consumer drains space +// instead of buffering unboundedly. Closing the stream unblocks the writer. var streamQueueLimit = 64 * 1024 * 1024 var ( - ErrStreamClosed = errors.New("audio stream is closed") - ErrStreamLimitReached = errors.New("audio stream queue limit exceeded") + ErrStreamClosed = errors.New("audio stream is closed") ) type StreamFormat struct { @@ -129,6 +130,12 @@ func (stream *Stream) Read(destination []byte) (int, error) { copied := copy(destination, stream.queue) stream.queue = stream.queue[copied:] + // Draining the queue may free space below the high-water mark; wake any + // writer blocked in Write applying backpressure. + if copied > 0 { + stream.condition.Broadcast() + } + if stream.teeSink != nil && copied > 0 { if _, err := stream.teeSink.Write(destination[:copied]); err != nil { stream.teeError = err @@ -189,11 +196,18 @@ func (writer *StreamWriter) Write(data []byte) (int, error) { stream.mutex.Lock() defer stream.mutex.Unlock() - if stream.consumerClosed || stream.producerClosed { - return 0, ErrStreamClosed - } - if len(stream.queue)+len(data) > streamQueueLimit { - return 0, ErrStreamLimitReached + // Apply backpressure: once the queue reaches the high-water mark, block + // until the consumer drains enough space instead of buffering unboundedly. + // A blocked writer wakes if the stream is closed or errored so a + // genuinely-gone consumer can never deadlock the producer forever. + for { + if stream.consumerClosed || stream.producerClosed { + return 0, ErrStreamClosed + } + if len(stream.queue) < streamQueueLimit { + break + } + stream.condition.Wait() } stream.queue = append(stream.queue, data...) diff --git a/app/common/audio/stream_test.go b/app/common/audio/stream_test.go index 3ca88b2..3388e27 100644 --- a/app/common/audio/stream_test.go +++ b/app/common/audio/stream_test.go @@ -161,18 +161,85 @@ func TestPipeStream(t *testing.T) { } }) - t.Run("QueueLimitFailsWrite", func(t *testing.T) { + t.Run("QueueLimitBlocksWriteUntilRead", func(t *testing.T) { originalLimit := streamQueueLimit streamQueueLimit = 8 t.Cleanup(func() { streamQueueLimit = originalLimit }) - _, writer := NewPipeStream(defaultStreamFormat()) + stream, writer := NewPipeStream(defaultStreamFormat()) + // Fill the queue up to the high-water mark. if _, err := writer.Write(patternBytes(8)); err != nil { t.Fatalf("Write within limit: %v", err) } - if _, err := writer.Write([]byte{0}); err != ErrStreamLimitReached { - t.Fatalf("got %v, want ErrStreamLimitReached", err) + + // The next write must block: the queue is at the limit. + writeDone := make(chan error, 1) + go func() { + _, err := writer.Write([]byte("more")) + writeDone <- err + }() + + select { + case err := <-writeDone: + t.Fatalf("Write returned %v while queue was full; expected it to block", err) + case <-time.After(50 * time.Millisecond): + // Still blocked, as intended. + } + + // Reading frees space below the limit, which must unblock the writer. + readExactly(t, stream, 8) + + select { + case err := <-writeDone: + if err != nil { + t.Fatalf("blocked Write returned %v after read freed space", err) + } + case <-time.After(5 * time.Second): + t.Fatal("Write never returned after a read freed queue space") + } + + data := readExactly(t, stream, 4) + if string(data) != "more" { + t.Fatalf("got %q, want %q", data, "more") + } + }) + + t.Run("ConsumerCloseUnblocksWrite", func(t *testing.T) { + originalLimit := streamQueueLimit + streamQueueLimit = 8 + t.Cleanup(func() { streamQueueLimit = originalLimit }) + + stream, writer := NewPipeStream(defaultStreamFormat()) + + if _, err := writer.Write(patternBytes(8)); err != nil { + t.Fatalf("Write within limit: %v", err) + } + + writeDone := make(chan error, 1) + go func() { + _, err := writer.Write([]byte("more")) + writeDone <- err + }() + + // Ensure the writer is parked on the condition before closing. + select { + case err := <-writeDone: + t.Fatalf("Write returned %v while queue was full; expected it to block", err) + case <-time.After(50 * time.Millisecond): + } + + // Closing the stream must wake the blocked writer with an error rather + // than deadlocking a producer whose consumer has gone away. + stream.Close() + + select { + case err := <-writeDone: + if err != ErrStreamClosed { + t.Fatalf("got %v, want ErrStreamClosed after consumer close", err) + } + case <-time.After(5 * time.Second): + t.Fatal("Write never returned after consumer Close unblocked it") } })