Make audio stream apply real backpressure instead of erroring

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...)

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")
 		}
 	})