diff --git a/fix.txt b/fix.txt index 64fbb04..df341a9 100644 --- a/fix.txt +++ b/fix.txt @@ -77,7 +77,7 @@ Priority 1: transport, lifecycle, and correctness failed. Mark UDP active only after an authenticated UDP response/audio packet (or retain TCP until confirmed), and define fallback/recovery rules. -9. Stream capture shutdown/startup races OpenAL +[x] 9. Stream capture shutdown/startup races OpenAL File: gumble/gumbleopenal/stream.go StopSource closes a channel but does not wait for sourceRoutine. Destroy immediately closes the capture device, so the routine can access a closed diff --git a/gumble/gumbleopenal/stream.go b/gumble/gumbleopenal/stream.go index 884040f..b08357c 100644 --- a/gumble/gumbleopenal/stream.go +++ b/gumble/gumbleopenal/stream.go @@ -59,7 +59,9 @@ type Stream struct { sourceChannels int sourceFrameSize int micVolume atomic.Uint32 // float32 stored as bits + sourceMu sync.Mutex sourceStop chan bool + sourceDone chan struct{} deviceSink *openal.Device contextSink *openal.Context @@ -323,6 +325,8 @@ func (s *Stream) Destroy() { } func (s *Stream) StartSource(inputDevice *string) error { + s.sourceMu.Lock() + defer s.sourceMu.Unlock() if s.sourceStop != nil { return ErrState } @@ -330,21 +334,29 @@ func (s *Stream) StartSource(inputDevice *string) error { return ErrMic } s.deviceSource.CaptureStart() - s.sourceStop = make(chan bool) - go s.sourceRoutine(inputDevice) + stop := make(chan bool) + done := make(chan struct{}) + s.sourceStop, s.sourceDone = stop, done + go s.sourceRoutine(inputDevice, stop, done) return nil } func (s *Stream) StopSource() error { + s.sourceMu.Lock() + if s.sourceStop == nil { + s.sourceMu.Unlock() + return ErrState + } + stop, done := s.sourceStop, s.sourceDone + s.sourceStop, s.sourceDone = nil, nil + close(stop) + s.sourceMu.Unlock() + // The routine owns capture access; wait for it before closing/reusing it. + <-done if s.deviceSource == nil { return ErrMic } s.deviceSource.CaptureStop() - if s.sourceStop == nil { - return ErrState - } - close(s.sourceStop) - s.sourceStop = nil return nil } @@ -644,7 +656,8 @@ func (s *Stream) processAudioPacket(packet *gumble.AudioPacket, user *gumble.Use return emptyBufs } -func (s *Stream) sourceRoutine(inputDevice *string) { +func (s *Stream) sourceRoutine(inputDevice *string, stop chan bool, done chan struct{}) { + defer close(done) log.Info("source routine started: interval=%v frameSize=%d channels=%d", s.client.Config.AudioInterval, s.client.Config.AudioFrameSize(), s.sourceChannels) interval := s.client.Config.AudioInterval @@ -679,8 +692,6 @@ func (s *Stream) sourceRoutine(inputDevice *string) { ticker := time.NewTicker(interval) defer ticker.Stop() - stop := s.sourceStop - outgoing := s.client.AudioOutgoing() defer close(outgoing) diff --git a/gumble/gumbleopenal/stream_regression_test.go b/gumble/gumbleopenal/stream_regression_test.go index 2c353b6..4fc4660 100644 --- a/gumble/gumbleopenal/stream_regression_test.go +++ b/gumble/gumbleopenal/stream_regression_test.go @@ -4,6 +4,22 @@ import "testing" // Regression: audio cleanup could send a final render command after Destroy // had closed renderCh, panicking instead of safely discarding that work. +// Regression: StopSource returned before the capture worker ended, allowing +// Destroy to close the device while that worker still used it. +func TestStopSourceWaitsForWorker(t *testing.T) { + stop, done := make(chan bool), make(chan struct{}) + s := &Stream{sourceStop: stop, sourceDone: done} + returned := make(chan struct{}) + go func() { _ = s.StopSource(); close(returned) }() + select { + case <-returned: + t.Fatal("StopSource returned before worker") + default: + } + close(done) + <-returned +} + func TestRenderRejectsWorkAfterShutdown(t *testing.T) { s := &Stream{renderClosed: true} called := false