bound the per-source recording backlog
Each tick drains one fixed chunk per source, so if the encoder stalls the loop never makes the deficit up and the backlog only grows from there. The queues had no cap, so a long recording against a slow encoder grew for as long as it ran. Cap each queue and keep the newest audio; discarding the newest instead would only push the recording further behind. Co-Authored-By: Claude Opus 5 <noreply@anthropic.com>
This commit is contained in:
co-authored by
Claude Opus 5
parent
53894b33ae
commit
31ae03ad65
@@ -0,0 +1,46 @@
|
||||
package recording
|
||||
|
||||
import "testing"
|
||||
|
||||
// Regression: the per-source mix queues grew without bound. The tick drains a
|
||||
// fixed chunk per source, so a stalled encoder leaves a deficit the loop never
|
||||
// makes up, and the backlog only ever grew from there.
|
||||
func TestRecorderQueueIsCapped(t *testing.T) {
|
||||
t.Parallel()
|
||||
|
||||
var queue []int16
|
||||
frame := make([]int16, 960)
|
||||
// Far more audio than the encoder could have consumed.
|
||||
for i := 0; i < 2000; i++ {
|
||||
queue = appendCapped(queue, frame)
|
||||
}
|
||||
|
||||
if len(queue) > maxQueuedSamples {
|
||||
t.Fatalf("queue grew to %d samples, above the %d cap",
|
||||
len(queue), maxQueuedSamples)
|
||||
}
|
||||
}
|
||||
|
||||
// Capping must keep the newest audio: dropping the newest would make the
|
||||
// recording lag further behind with every overflow.
|
||||
func TestRecorderQueueKeepsNewestAudio(t *testing.T) {
|
||||
t.Parallel()
|
||||
|
||||
var queue []int16
|
||||
// Fill past the cap with a marker in the final frame.
|
||||
filler := make([]int16, maxQueuedSamples)
|
||||
queue = appendCapped(queue, filler)
|
||||
newest := []int16{1, 2, 3, 4}
|
||||
queue = appendCapped(queue, newest)
|
||||
|
||||
if len(queue) != maxQueuedSamples {
|
||||
t.Fatalf("expected the queue to sit at the %d cap, got %d",
|
||||
maxQueuedSamples, len(queue))
|
||||
}
|
||||
tail := queue[len(queue)-len(newest):]
|
||||
for i, want := range newest {
|
||||
if tail[i] != want {
|
||||
t.Fatalf("newest audio was dropped: tail %v, want %v", tail, newest)
|
||||
}
|
||||
}
|
||||
}
|
||||
+20
-1
@@ -20,6 +20,12 @@ const (
|
||||
FormatOpus = "opus"
|
||||
)
|
||||
|
||||
// maxQueuedSamples bounds the per-source mix backlog at roughly five seconds
|
||||
// of 48 kHz stereo audio. A source that runs further ahead than this is ahead
|
||||
// because the encoder stalled, and no amount of retained audio recovers the
|
||||
// timeline; keeping the newest is better than growing without bound.
|
||||
const maxQueuedSamples = 5 * gumble.AudioSampleRate * gumble.AudioChannels
|
||||
|
||||
type Recorder struct {
|
||||
path string
|
||||
format string
|
||||
@@ -188,6 +194,19 @@ func (r *Recorder) Stop() error {
|
||||
return r.err
|
||||
}
|
||||
|
||||
// appendCapped adds a source's incoming samples to its mix queue, bounded at
|
||||
// maxQueuedSamples. Each tick drains one fixed chunk per source, so a stalled
|
||||
// encoder leaves a deficit the loop never makes up and the backlog would
|
||||
// otherwise grow for as long as the recording ran. The newest audio is kept:
|
||||
// discarding it instead would only push the recording further behind.
|
||||
func appendCapped(queue []int16, incoming []int16) []int16 {
|
||||
queue = append(queue, incoming...)
|
||||
if len(queue) > maxQueuedSamples {
|
||||
queue = append(queue[:0], queue[len(queue)-maxQueuedSamples:]...)
|
||||
}
|
||||
return queue
|
||||
}
|
||||
|
||||
func (r *Recorder) run() {
|
||||
defer close(r.done)
|
||||
ticker := time.NewTicker(r.interval)
|
||||
@@ -203,7 +222,7 @@ func (r *Recorder) run() {
|
||||
r.closeEncoder()
|
||||
return
|
||||
case item := <-r.input:
|
||||
queues[item.source] = append(queues[item.source], item.samples...)
|
||||
queues[item.source] = appendCapped(queues[item.source], item.samples)
|
||||
case <-ticker.C:
|
||||
clear(chunk)
|
||||
for source, buffer := range queues {
|
||||
|
||||
Reference in New Issue
Block a user