backend/player/buffered_stall_test.go

@@ -0,0 +1,188 @@
+package player
+
+import (
+	"errors"
+	"testing"
+	"time"
+
+	"github.com/gopxl/beep/v2"
+)
+
+// errTestDecode stands in for a decoder blowing up mid-track.
+var errTestDecode = errors.New("decode blew up")
+
+// stalledStreamer never produces a sample and never reports
+// end-of-stream: (0, true), forever. A damaged file that decodes to
+// nothing looks like this, and so does any source whose producer has
+// quietly stopped.
+type stalledStreamer struct{}
+
+func (stalledStreamer) Stream(_ [][2]float64) (int, bool) { return 0, true }
+func (stalledStreamer) Err() error                        { return nil }
+
+// failingStreamer produces n good samples and then fails, which is
+// what a decode error mid-track looks like: the same (0, false) a
+// finished track returns, distinguishable only by Err.
+type failingStreamer struct {
+	remaining int
+	err       error
+}
+
+func (f *failingStreamer) Stream(samples [][2]float64) (int, bool) {
+	if f.remaining <= 0 {
+		return 0, false
+	}
+
+	n := min(len(samples), f.remaining)
+
+	for i := range n {
+		samples[i] = [2]float64{1, 1}
+	}
+
+	f.remaining -= n
+
+	return n, true
+}
+
+func (f *failingStreamer) Err() error { return f.err }
+
+// drainUntilEnd calls Stream until it reports end-of-stream, or gives
+// up. It returns whether the stream ended.
+//
+// The give-up bound is wall clock rather than a call count: the stall
+// budget is a duration, so a tight loop has to actually wait it out.
+func drainUntilEnd(bs *BufferedStreamer, within time.Duration) bool {
+	buf := make([][2]float64, 512)
+	deadline := time.Now().Add(within)
+
+	for time.Now().Before(deadline) {
+		if _, ok := bs.Stream(buf); !ok {
+			return true
+		}
+
+		time.Sleep(time.Millisecond)
+	}
+
+	return false
+}
+
+// A source that stops producing without ever ending is the fault this
+// whole file exists for: Stream used to answer with silence and ok
+// forever, so the chain never ended, the player stayed in Playing
+// with the button showing pause, and the decoder's position never
+// moved -- a frozen seek bar over a track that was not playing.
+func TestAStalledSourceEndsTheStream(t *testing.T) {
+	bs := NewBufferedStreamer(stalledStreamer{}, 2048)
+	defer bs.Close()
+
+	if !drainUntilEnd(bs, maxStarvedDuration+2*time.Second) {
+		t.Fatal(
+			"a stalled source never ended the stream: the player " +
+				"would sit in Playing with a frozen position",
+		)
+	}
+
+	if !errors.Is(bs.Err(), errSourceStalled) {
+		t.Fatalf(
+			"expected the stall to be reported, got %v", bs.Err(),
+		)
+	}
+}
+
+// Close is the other exit that used to leave `done` false, with the
+// same consequence: the ring drains and every call after it is
+// silence that claims to be audio.
+func TestClosingEndsTheStream(t *testing.T) {
+	bs := NewBufferedStreamer(finiteStreamer(1<<20), 2048)
+
+	// Let the read-ahead fill something, so this exercises the drain
+	// after Close rather than a buffer that was empty anyway.
+	time.Sleep(20 * time.Millisecond)
+	bs.Close()
+
+	if !drainUntilEnd(bs, 2*time.Second) {
+		t.Fatal("a closed streamer never reported end-of-stream")
+	}
+}
+
+// A source that fails is not a source that finished, and only Err
+// tells them apart. Before this, the player reported a mid-track
+// decode failure to the queue as a natural end, so the queue
+// auto-advanced in silence and counted the broken track as played.
+func TestAFailedSourceReportsItsError(t *testing.T) {
+	src := &failingStreamer{remaining: 4096, err: errTestDecode}
+
+	bs := NewBufferedStreamer(src, 2048)
+	defer bs.Close()
+
+	if !drainUntilEnd(bs, 2*time.Second) {
+		t.Fatal("a failing source never reported end-of-stream")
+	}
+
+	if !errors.Is(bs.Err(), errTestDecode) {
+		t.Fatalf(
+			"expected the source's error to survive, got %v",
+			bs.Err(),
+		)
+	}
+}
+
+// The ordinary case has to keep working: a source that ends cleanly
+// ends with no error, or every finished track would be reported as a
+// failure and skipped.
+func TestADrainedSourceReportsNoError(t *testing.T) {
+	bs := NewBufferedStreamer(finiteStreamer(4096), 2048)
+	defer bs.Close()
+
+	if !drainUntilEnd(bs, 2*time.Second) {
+		t.Fatal("a finite source never reported end-of-stream")
+	}
+
+	if bs.Err() != nil {
+		t.Fatalf(
+			"a track that finished normally reported %v", bs.Err(),
+		)
+	}
+}
+
+// A slow source is exactly what the read-ahead exists to absorb, so
+// underruns must not be charged cumulatively -- otherwise a file on a
+// slow disk ends itself partway through.
+func TestUnderrunsDoNotAccumulateAcrossASlowSource(t *testing.T) {
+	const total = 8192
+
+	src := &slowStreamer{
+		inner: finiteStreamer(total),
+		delay: 2 * time.Millisecond,
+	}
+
+	bs := NewBufferedStreamer(src, 1024)
+	defer bs.Close()
+
+	buf := make([][2]float64, 256)
+	got := 0
+
+	for {
+		n, ok := bs.Stream(buf)
+		if !ok {
+			break
+		}
+
+		for i := range n {
+			if buf[i][0] != 0 {
+				got++
+			}
+		}
+	}
+
+	if got != total {
+		t.Fatalf(
+			"a slow but healthy source was cut short: got %d of %d "+
+				"samples",
+			got, total,
+		)
+	}
+}
+
+// beep.Streamer is what the player wraps; keep the type honest.
+var _ beep.Streamer = (*BufferedStreamer)(nil)

backend/player/buffered_streamer.go

@@ -1,6 +1,7 @@
 package player
 
 import (
+	"errors"
 	"sync"
 	"time"
 
@@ -34,8 +35,45 @@ type BufferedStreamer struct {
 	done    bool
 	err     error
 	closed  chan struct{}
+
+	// starved counts consecutive Stream calls served with silence
+	// because the ring was empty, and starvedSince is when that run
+	// began.  An underrun is legitimate for a moment -- that is what
+	// the read-ahead exists to absorb -- but it is not legitimate
+	// forever, and "forever" is indistinguishable from healthy
+	// playback everywhere above this type: the chain never ends, so
+	// the player stays in Playing with the button showing pause, and
+	// the decoder's position never moves, so the 1 Hz report pins the
+	// seek bar and suppresses its interpolation.
+	starved      int
+	starvedSince time.Time
 }
 
+// The silence fill is bounded by both a duration and a run of calls,
+// and it needs both.
+//
+// Duration alone is the real measure -- the speaker paces itself, so
+// wall clock is what says whether the source has actually stopped --
+// but a caller draining in a tight loop (a test, a decode-to-buffer)
+// makes hundreds of calls in microseconds and would trip nothing.
+// A call count alone is the opposite failure: the same tight loop
+// spends the whole budget before the read-ahead goroutine has been
+// scheduled once, and ends a perfectly good stream at sample zero.
+//
+// The duration is longer than the 2 s read-ahead it is there to
+// outlast, and the count is short enough that the speaker (~200 ms a
+// call) reaches it well inside that.
+const (
+	maxStarvedDuration = 3 * time.Second
+	minStarvedCalls    = 8
+)
+
+// errSourceStalled is returned by Err when the source stopped
+// producing samples without ever reporting end-of-stream.
+var errSourceStalled = errors.New(
+	"audio source stopped producing samples",
+)
+
 // NewBufferedStreamer creates a BufferedStreamer that pre-fills
 // bufferSize samples from source via a background goroutine.
 // A typical bufferSize is 2× the sample rate (~2 seconds of audio).
@@ -54,8 +92,24 @@ func NewBufferedStreamer(
 	return bs
 }
 
+// finish marks the stream ended, recording err as the reason when
+// there is one.  Every exit from readAhead goes through it: an exit
+// that leaves done false strands Stream in its underrun branch,
+// where it returns silence and ok forever.
+func (bs *BufferedStreamer) finish(err error) {
+	bs.mu.Lock()
+	defer bs.mu.Unlock()
+
+	bs.done = true
+
+	if err != nil && bs.err == nil {
+		bs.err = err
+	}
+}
+
 // readAhead continuously reads from the source into the ring buffer
 // until the source is drained, an error occurs, or Close is called.
+// It always marks the stream done on the way out.
 func (bs *BufferedStreamer) readAhead() {
 	// Temporary buffer for reading from source outside the lock.
 	// 512 samples per chunk keeps the critical section short.
@@ -63,6 +117,13 @@ func (bs *BufferedStreamer) readAhead() {
 
 	tmp := make([][2]float64, chunkSize)
 
+	// Every exit marks the stream done.  An exit that does not is what
+	// stranded Stream in its underrun branch, returning silence and ok
+	// for the rest of the process's life.
+	var exitErr error
+
+	defer func() { bs.finish(exitErr) }()
+
 	for {
 		// Check if closed.
 		select {
@@ -72,6 +133,15 @@ func (bs *BufferedStreamer) readAhead() {
 		}
 
 		bs.mu.Lock()
+
+		// Stream gave up waiting for us.  Nothing downstream is
+		// listening any more, so filling the ring is work for nobody.
+		if bs.done {
+			bs.mu.Unlock()
+
+			return
+		}
+
 		space := len(bs.ring) - bs.count
 
 		if space == 0 {
@@ -115,14 +185,12 @@ func (bs *BufferedStreamer) readAhead() {
 		}
 
 		if !ok {
-			bs.mu.Lock()
+			// A drained source and a failed one both land here and are
-			bs.done = true
+			// not the same event: one is a track that ended, the other
-
+			// is a track that broke.  Err is what tells them apart, and
-			if srcErr := bs.source.Err(); srcErr != nil {
+			// it is why the player must ask before treating this as a
-				bs.err = srcErr
+			// natural finish.
-			}
+			exitErr = bs.source.Err()
-
-			bs.mu.Unlock()
 
 			return
 		}
@@ -154,7 +222,27 @@ func (bs *BufferedStreamer) Stream(
 	}
 
 	if bs.count == 0 {
-		// Buffer temporarily empty — fill with silence.
+		// The read-ahead has not caught up.  Silence buys it time --
+		// but only for a bounded stretch, because "forever" is
+		// reported upward as healthy playback and there is no watchdog
+		// above this to notice otherwise.
+		bs.starved++
+
+		if bs.starvedSince.IsZero() {
+			bs.starvedSince = time.Now()
+		}
+
+		if bs.starved >= minStarvedCalls &&
+			time.Since(bs.starvedSince) > maxStarvedDuration {
+			bs.done = true
+
+			if bs.err == nil {
+				bs.err = errSourceStalled
+			}
+
+			return 0, false
+		}
+
 		for i := range samples {
 			samples[i] = [2]float64{}
 		}
@@ -162,6 +250,9 @@ func (bs *BufferedStreamer) Stream(
 		return len(samples), true
 	}
 
+	// Samples arrived, so whatever the stall was, it is over.
+	bs.resetStarvationLocked()
+
 	// Copy available samples from ring buffer.
 	n := len(samples)
 	if n > bs.count {
@@ -197,6 +288,19 @@ func (bs *BufferedStreamer) Flush() {
 	bs.readPos = 0
 	bs.writPos = 0
 	bs.count = 0
+
+	// A seek empties the ring on purpose, and the refill that follows
+	// is exactly the stall the budget exists to tolerate.  Charging it
+	// against a budget the previous underrun already spent would end
+	// the track on a seek near the end of a slow file.
+	bs.resetStarvationLocked()
+}
+
+// resetStarvationLocked forgets an underrun run.  Must be called with
+// bs.mu held.
+func (bs *BufferedStreamer) resetStarvationLocked() {
+	bs.starved = 0
+	bs.starvedSince = time.Time{}
 }
 
 // LockSource blocks the read-ahead goroutine from touching the