diff --git a/backend/player/buffered_stall_test.go b/backend/player/buffered_stall_test.go new file mode 100644 index 0000000..3dbc7cc --- /dev/null +++ b/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) diff --git a/backend/player/buffered_streamer.go b/backend/player/buffered_streamer.go index f9e01cf..993286e 100644 --- a/backend/player/buffered_streamer.go +++ b/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() - bs.done = true - - if srcErr := bs.source.Err(); srcErr != nil { - bs.err = srcErr - } - - bs.mu.Unlock() + // A drained source and a failed one both land here and are + // not the same event: one is a track that ended, the other + // is a track that broke. Err is what tells them apart, and + // it is why the player must ask before treating this as a + // natural finish. + exitErr = bs.source.Err() 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