detached 8d647bb

Merge pull request #348 from bnema/perf/cef-audio-handoff

perf(audio): bound CEF packet handoff
commit
8d647bbb2c745851403061d0bac7f0e21fe3ee82
author
bnema · · 2026-07-14 17:29 UTC
committer
GitHub · · 2026-07-14 17:29 UTC
merge of
513887c + 5f3e0e8 · diff against the first parent
signature
GPG UnverifiedB5690EEEBB952194 The key is unknown
6 files · +489 -759
Unified
UnifiedSplit
M internal/application/port/audio_output.go +4 -3
@@ -18,9 +18,10 @@ type AudioStreamFormat struct {
1818 // AudioOutputStream represents an active audio output stream.
1919 // Implementations handle the low-level audio playback.
2020 type AudioOutputStream interface {
21- // Write sends audio samples to the output device.
22- // Samples are provided as a 2D slice: [channel][frame]float32.
23- // This matches CEF's planar float32 audio format.
21+ // Write sends callback-owned audio samples to the output device.
22+ // Samples are provided as [channel][frame]float32 matching CEF's planar
23+ // format. Implementations must not retain the slices and must be safe to
24+ // call concurrently with Close; audio callbacks must not block.
2425 Write(samples [][]float32) error
2526 // Close releases the audio stream and associated resources.
2627 Close() error
@@ -18,9 +18,10 @@ type AudioStreamFormat struct {
18 // AudioOutputStream represents an active audio output stream.18 // AudioOutputStream represents an active audio output stream.
19 // Implementations handle the low-level audio playback.19 // Implementations handle the low-level audio playback.
20 type AudioOutputStream interface {20 type AudioOutputStream interface {
21- // Write sends audio samples to the output device.21+ // Write sends callback-owned audio samples to the output device.
22- // Samples are provided as a 2D slice: [channel][frame]float32.22+ // Samples are provided as [channel][frame]float32 matching CEF's planar
23- // This matches CEF's planar float32 audio format.23+ // format. Implementations must not retain the slices and must be safe to
24+ // call concurrently with Close; audio callbacks must not block.
24 Write(samples [][]float32) error25 Write(samples [][]float32) error
25 // Close releases the audio stream and associated resources.26 // Close releases the audio stream and associated resources.
26 Close() error27 Close() error
M internal/infrastructure/audio/pipewire/output.go +123 -123
@@ -15,7 +15,6 @@ import (
1515 "github.com/bnema/dumber/internal/logging"
1616 )
1717
18-// Compile-time interface checks.
1918 var (
2019 _ port.AudioOutputFactory = (*Factory)(nil)
2120 _ port.AudioOutputStream = (*playerStreamAdapter)(nil)
@@ -27,6 +26,8 @@ var ErrInvalidFormat = errors.New("invalid audio stream format")
2726 // ErrStreamClosed is returned when writing to a closed stream.
2827 var ErrStreamClosed = errors.New("audio stream closed")
2928
29+const packetQueueCapacity = 4
30+
3031 // playerCreator is a function type that creates a Player from config and callbacks.
3132 // This exists to allow dependency injection in tests.
3233 type playerCreator func(config pwpipewire.PlayerConfig, callbacks pwpipewire.PlayerCallbacks) (pwpipewire.Player, error)
@@ -38,24 +39,15 @@ type Factory struct {
3839 }
3940
4041 // NewFactory creates a new PipeWire audio factory.
41-// The factory is stateless; players are created per-stream.
4242 func NewFactory() (*Factory, error) {
43- return &Factory{
44- createPlayer: pwpipewire.NewPlayer,
45- }, nil
43+ return &Factory{createPlayer: pwpipewire.NewPlayer}, nil
4644 }
4745
48-// NewStream creates a new audio output stream with the given format.
49-// It validates the Dumber format, creates a purego-pipewire Player configured
50-// from the format, and returns a push-to-pull adapter that bridges Write()
51-// calls into the Player's Fill callback.
46+// NewStream creates a push-to-pull adapter for a PipeWire player.
5247 func (f *Factory) NewStream(ctx context.Context, format port.AudioStreamFormat) (port.AudioOutputStream, error) {
53- // Check context cancellation
5448 if ctx.Err() != nil {
5549 return nil, ctx.Err()
5650 }
57-
58- // Validate format
5951 if err := validateFormat(format); err != nil {
6052 return nil, err
6153 }
@@ -67,182 +59,195 @@ func (f *Factory) NewStream(ctx context.Context, format port.AudioStreamFormat)
6759 Int("frames_per_buffer", format.FramesPerBuffer).
6860 Msg("pipewire: creating player")
6961
70- // Create the adapter with its ring buffer before creating the player,
71- // because the Fill callback captures the adapter.
72- adapter := &playerStreamAdapter{
73- ctx: ctx,
74- buf: make(chan [][]float32, 4), // small buffer for push→pull bridging
75- closed: make(chan struct{}),
76- }
77-
62+ // Every slot owns storage sized from the negotiated CEF format. Write copies
63+ // directly into an available slot; Fill copies that same slot into PipeWire.
64+ // Callback slices are therefore never queued or retained.
65+ adapter := newPlayerStreamAdapter(ctx, format.ChannelCount, format.FramesPerBuffer)
7866 config := pwpipewire.PlayerConfig{
79- SampleRate: format.SampleRate,
80- Channels: format.ChannelCount,
81- FramesPerBuffer: format.FramesPerBuffer,
82- SampleFormat: pwpipewire.SampleFormatF32,
83- UnderrunPolicy: pwpipewire.UnderrunFillSilence,
67+ SampleRate: format.SampleRate, Channels: format.ChannelCount, FramesPerBuffer: format.FramesPerBuffer,
68+ SampleFormat: pwpipewire.SampleFormatF32, UnderrunPolicy: pwpipewire.UnderrunFillSilence,
8469 }
85-
86- callbacks := pwpipewire.PlayerCallbacks{
87- Fill: adapter.fill,
88- }
89-
90- player, err := f.createPlayer(config, callbacks)
70+ player, err := f.createPlayer(config, pwpipewire.PlayerCallbacks{Fill: adapter.fill})
9171 if err != nil {
9272 log.Warn().Err(err).Msg("pipewire: failed to create player")
9373 return nil, fmt.Errorf("failed to create pipewire player: %w", err)
9474 }
9575 adapter.player = player
96-
97- // Start the player so the Fill callback begins being called.
9876 if err := player.Start(); err != nil {
9977 _ = player.Close()
10078 log.Warn().Err(err).Msg("pipewire: failed to start player")
10179 return nil, fmt.Errorf("failed to start pipewire player: %w", err)
10280 }
103-
10481 log.Info().Msg("pipewire: player started successfully")
105-
10682 return adapter, nil
10783 }
10884
109-// validateFormat checks if the audio format parameters are valid.
11085 func validateFormat(format port.AudioStreamFormat) error {
111- // Sample rate must be reasonable (8kHz to 384kHz)
11286 if format.SampleRate < 8000 || format.SampleRate > 384000 {
113- return fmt.Errorf("%w: sample rate %d Hz out of range [8000, 384000]",
114- ErrInvalidFormat, format.SampleRate)
87+ return fmt.Errorf("%w: sample rate %d Hz out of range [8000, 384kHz]", ErrInvalidFormat, format.SampleRate)
11588 }
116-
117- // Channel count must be 1 to 1024 (reasonable upper limit)
11889 if format.ChannelCount < 1 || format.ChannelCount > 1024 {
119- return fmt.Errorf("%w: channel count %d out of range [1, 1024]",
120- ErrInvalidFormat, format.ChannelCount)
90+ return fmt.Errorf("%w: channel count %d out of range [1, 1024]", ErrInvalidFormat, format.ChannelCount)
12191 }
122-
123- // Frames per buffer must be positive and reasonable
12492 if format.FramesPerBuffer < 1 || format.FramesPerBuffer > 8192 {
125- return fmt.Errorf("%w: frames per buffer %d out of range [1, 8192]",
126- ErrInvalidFormat, format.FramesPerBuffer)
93+ return fmt.Errorf("%w: frames per buffer %d out of range [1, 8192]", ErrInvalidFormat, format.FramesPerBuffer)
12794 }
128-
12995 return nil
13096 }
13197
132-// playerStreamAdapter bridges Dumber's push-style Write(samples) interface
133-// to purego-pipewire's pull-style Fill callback.
134-//
135-// Write() sends sample buffers into a channel. The Fill callback drains them
136-// into the PCMBuffer that purego-pipewire passes in. If no data is available,
137-// the player's UnderrunFillSilence policy handles it.
98+// audioPacket is a reusable, adapter-owned packet slot. Its backing arrays are
99+// allocated only when the stream is created and never alias CEF callback data.
100+type audioPacket struct {
101+ samples [][]float32
102+ channels int
103+ frames int
104+}
105+
106+// playerStreamAdapter bridges callback-owned CEF audio to PipeWire's Fill
107+// callback. available and ready form a bounded reusable packet ring.
138108 type playerStreamAdapter struct {
139109 player pwpipewire.Player
140110 ctx context.Context
141- buf chan [][]float32 // push→pull handoff channel
111+
112+ available chan *audioPacket
113+ ready chan *audioPacket
114+ // TryLock serializes only simultaneous producers. The CEF callback never
115+ // waits: contention and saturation both drop the packet. Fill never takes it.
116+ writeMu sync.Mutex
117+ closed atomic.Bool
142118
143119 closeOnce sync.Once
144- closed chan struct{} // closed when Close is called
145120
146- // Diagnostic counters (atomic, no mutex needed).
147- fillCount atomic.Uint64 // total Fill callbacks
148- underrunCount atomic.Uint64 // Fill callbacks with no data available
149- dropCount atomic.Uint64 // Write calls that dropped data (buffer full)
121+ fillCount atomic.Uint64
122+ underrunCount atomic.Uint64
123+ dropCount atomic.Uint64
124+}
125+
126+func newPlayerStreamAdapter(ctx context.Context, channels, frames int) *playerStreamAdapter {
127+ a := &playerStreamAdapter{
128+ ctx: ctx, available: make(chan *audioPacket, packetQueueCapacity), ready: make(chan *audioPacket, packetQueueCapacity),
129+ }
130+ for range packetQueueCapacity {
131+ packet := &audioPacket{samples: make([][]float32, channels)}
132+ for ch := range packet.samples {
133+ packet.samples[ch] = make([]float32, frames)
134+ }
135+ a.available <- packet
136+ }
137+ return a
150138 }
151139
152-// Write sends audio samples to the PipeWire output.
153-// Samples are provided as [channel][frame]float32 matching CEF's planar format.
154-// If the internal buffer is full the packet is dropped silently so the caller
155-// (CEF audio callback thread) is never blocked.
156-// Safe for concurrent use with Close.
140+// Write makes exactly one owned copy for each accepted packet. It never blocks:
141+// full queues or another producer cause a silent drop, preserving CEF timing.
142+// The supplied slices are callback-owned and are not retained after return.
157143 func (a *playerStreamAdapter) Write(samples [][]float32) error {
158- // Fast path: already closed.
159- select {
160- case <-a.closed:
144+ if a.closed.Load() {
161145 return ErrStreamClosed
162- default:
163146 }
164- // Non-blocking send: drop packet when buffer is full.
165- select {
166- case a.buf <- samples:
147+ if !a.writeMu.TryLock() {
148+ a.recordDrop()
167149 return nil
168- case <-a.closed:
150+ }
151+ defer a.writeMu.Unlock()
152+ if a.closed.Load() {
169153 return ErrStreamClosed
170- default:
171- // Buffer full — drop to avoid blocking the CEF audio thread.
172- n := a.dropCount.Add(1)
173- if n == 1 || n&(n-1) == 0 { // first drop, then powers of two
174- if a.ctx != nil {
175- logging.FromContext(a.ctx).Warn().
176- Uint64("total_drops", n).
177- Msg("pipewire: dropped audio packet (buffer full)")
178- }
154+ }
155+
156+ select {
157+ case packet := <-a.available:
158+ copyPacket(packet, samples)
159+ // A close racing after the first check must not publish a packet to a
160+ // stopped player. Returning the slot keeps the ring bounded.
161+ if a.closed.Load() {
162+ a.available <- packet
163+ return ErrStreamClosed
164+ }
165+ select {
166+ case a.ready <- packet:
167+ return nil
168+ default:
169+ // This cannot occur while ring invariants hold, but keep Write
170+ // non-blocking if an implementation change violates them.
171+ a.available <- packet
172+ a.recordDrop()
173+ return nil
179174 }
175+ default:
176+ a.recordDrop()
180177 return nil
181178 }
182179 }
183180
184-// fill is the PlayerCallbacks.Fill function called by purego-pipewire
185-// when it needs audio data. It pulls from the internal buffer channel.
181+func copyPacket(packet *audioPacket, samples [][]float32) {
182+ packet.channels = min(len(packet.samples), len(samples))
183+ packet.frames = 0
184+ if packet.channels == 0 {
185+ return
186+ }
187+ frames := len(packet.samples[0])
188+ for ch := 0; ch < packet.channels; ch++ {
189+ n := min(len(packet.samples[ch]), len(samples[ch]))
190+ if n < frames {
191+ frames = n
192+ }
193+ }
194+ packet.frames = frames
195+ for ch := 0; ch < packet.channels; ch++ {
196+ copy(packet.samples[ch][:frames], samples[ch][:frames])
197+ }
198+}
199+
200+func (a *playerStreamAdapter) recordDrop() {
201+ // Write runs on CEF's real-time callback thread. Diagnostics must stay
202+ // allocation-free and must not synchronously emit logs from that path.
203+ a.dropCount.Add(1)
204+}
205+
206+// fill is the PlayerCallbacks.Fill function. It performs the only post-handoff
207+// full-packet copy, directly into PipeWire's destination buffer.
186208 func (a *playerStreamAdapter) fill(pcm *pwpipewire.PCMBuffer) (int, error) {
187209 select {
188- case samples := <-a.buf:
210+ case packet := <-a.ready:
189211 fillNum := a.fillCount.Add(1)
190212 if fillNum == 1 && a.ctx != nil {
191- logging.FromContext(a.ctx).Info().
192- Msg("pipewire: first Fill callback with data")
213+ logging.FromContext(a.ctx).Info().Msg("pipewire: first Fill callback with data")
193214 }
194- // Copy available samples into the PCM buffer
195- copied := copyToPCM(pcm, samples)
215+ copied := copyToPCM(pcm, packet.samples[:packet.channels], packet.frames)
216+ a.available <- packet
196217 return copied, nil
197218 default:
198- // No data available — return 0 frames, let underrun policy handle it
199219 underruns := a.underrunCount.Add(1)
200220 if underruns == 1 && a.ctx != nil {
201- logging.FromContext(a.ctx).Debug().
202- Msg("pipewire: first Fill underrun (no data available)")
221+ logging.FromContext(a.ctx).Debug().Msg("pipewire: first Fill underrun (no data available)")
203222 }
204223 return 0, nil
205224 }
206225 }
207226
208-// copyToPCM copies planar samples into the PCMBuffer.
209-// Returns the number of frames actually copied.
210-func copyToPCM(pcm *pwpipewire.PCMBuffer, samples [][]float32) int {
211- if pcm == nil || len(samples) == 0 {
227+func copyToPCM(pcm *pwpipewire.PCMBuffer, samples [][]float32, packetFrames int) int {
228+ if pcm == nil || len(samples) == 0 || packetFrames == 0 {
212229 return 0
213230 }
214-
215- frames := pcm.Frames
216231 channels := min(min(len(samples), pcm.Channels), len(pcm.Samples))
217232 if channels == 0 {
218233 return 0
219234 }
220-
221- minFrames := frames
222-
223- for ch := 0; ch < channels && ch < len(samples); ch++ {
224- dst := pcm.Samples[ch]
225- src := samples[ch]
226- if len(dst) == 0 || len(src) == 0 {
227- minFrames = 0
228- continue
229- }
230- srcFrames := len(src)
231- n := min(len(dst), min(srcFrames, frames))
232- if n < minFrames {
233- minFrames = n
234- }
235- copy(dst[:n], src[:n])
235+ frames := min(pcm.Frames, packetFrames)
236+ for ch := 0; ch < channels; ch++ {
237+ frames = min(frames, min(len(pcm.Samples[ch]), len(samples[ch])))
236238 }
237-
238- return minFrames
239+ for ch := 0; ch < channels; ch++ {
240+ copy(pcm.Samples[ch][:frames], samples[ch][:frames])
241+ }
242+ return frames
239243 }
240244
241-// Close stops the player and releases resources.
242-// Safe to call multiple times.
245+// Close stops the player. It is safe with Write: Write observes closed before
246+// publishing, and no channel is closed while a callback may be selecting on it.
243247 func (a *playerStreamAdapter) Close() error {
244248 var err error
245249 a.closeOnce.Do(func() {
250+ a.closed.Store(true)
246251 if a.ctx != nil {
247252 logging.FromContext(a.ctx).Info().
248253 Uint64("fills", a.fillCount.Load()).
@@ -250,12 +255,7 @@ func (a *playerStreamAdapter) Close() error {
250255 Uint64("drops", a.dropCount.Load()).
251256 Msg("pipewire: closing player")
252257 }
253-
254- // Signal writers to stop
255- close(a.closed)
256-
257258 if a.player != nil {
258- // Stop playback then close the player
259259 if stopErr := a.player.Stop(); stopErr != nil {
260260 err = stopErr
261261 }
@@ -15,7 +15,6 @@ import (
15 "github.com/bnema/dumber/internal/logging"15 "github.com/bnema/dumber/internal/logging"
16 )16 )
17 17
18-// Compile-time interface checks.
19 var (18 var (
20 _ port.AudioOutputFactory = (*Factory)(nil)19 _ port.AudioOutputFactory = (*Factory)(nil)
21 _ port.AudioOutputStream = (*playerStreamAdapter)(nil)20 _ port.AudioOutputStream = (*playerStreamAdapter)(nil)
@@ -27,6 +26,8 @@ var ErrInvalidFormat = errors.New("invalid audio stream format")
27 // ErrStreamClosed is returned when writing to a closed stream.26 // ErrStreamClosed is returned when writing to a closed stream.
28 var ErrStreamClosed = errors.New("audio stream closed")27 var ErrStreamClosed = errors.New("audio stream closed")
29 28
29+const packetQueueCapacity = 4
30+
30 // playerCreator is a function type that creates a Player from config and callbacks.31 // playerCreator is a function type that creates a Player from config and callbacks.
31 // This exists to allow dependency injection in tests.32 // This exists to allow dependency injection in tests.
32 type playerCreator func(config pwpipewire.PlayerConfig, callbacks pwpipewire.PlayerCallbacks) (pwpipewire.Player, error)33 type playerCreator func(config pwpipewire.PlayerConfig, callbacks pwpipewire.PlayerCallbacks) (pwpipewire.Player, error)
@@ -38,24 +39,15 @@ type Factory struct {
38 }39 }
39 40
40 // NewFactory creates a new PipeWire audio factory.41 // NewFactory creates a new PipeWire audio factory.
41-// The factory is stateless; players are created per-stream.
42 func NewFactory() (*Factory, error) {42 func NewFactory() (*Factory, error) {
43- return &Factory{43+ return &Factory{createPlayer: pwpipewire.NewPlayer}, nil
44- createPlayer: pwpipewire.NewPlayer,
45- }, nil
46 }44 }
47 45
48-// NewStream creates a new audio output stream with the given format.46+// NewStream creates a push-to-pull adapter for a PipeWire player.
49-// It validates the Dumber format, creates a purego-pipewire Player configured
50-// from the format, and returns a push-to-pull adapter that bridges Write()
51-// calls into the Player's Fill callback.
52 func (f *Factory) NewStream(ctx context.Context, format port.AudioStreamFormat) (port.AudioOutputStream, error) {47 func (f *Factory) NewStream(ctx context.Context, format port.AudioStreamFormat) (port.AudioOutputStream, error) {
53- // Check context cancellation
54 if ctx.Err() != nil {48 if ctx.Err() != nil {
55 return nil, ctx.Err()49 return nil, ctx.Err()
56 }50 }
57-
58- // Validate format
59 if err := validateFormat(format); err != nil {51 if err := validateFormat(format); err != nil {
60 return nil, err52 return nil, err
61 }53 }
@@ -67,182 +59,195 @@ func (f *Factory) NewStream(ctx context.Context, format port.AudioStreamFormat)
67 Int("frames_per_buffer", format.FramesPerBuffer).59 Int("frames_per_buffer", format.FramesPerBuffer).
68 Msg("pipewire: creating player")60 Msg("pipewire: creating player")
69 61
70- // Create the adapter with its ring buffer before creating the player,62+ // Every slot owns storage sized from the negotiated CEF format. Write copies
71- // because the Fill callback captures the adapter.63+ // directly into an available slot; Fill copies that same slot into PipeWire.
72- adapter := &playerStreamAdapter{64+ // Callback slices are therefore never queued or retained.
73- ctx: ctx,65+ adapter := newPlayerStreamAdapter(ctx, format.ChannelCount, format.FramesPerBuffer)
74- buf: make(chan [][]float32, 4), // small buffer for push→pull bridging
75- closed: make(chan struct{}),
76- }
77-
78 config := pwpipewire.PlayerConfig{66 config := pwpipewire.PlayerConfig{
79- SampleRate: format.SampleRate,67+ SampleRate: format.SampleRate, Channels: format.ChannelCount, FramesPerBuffer: format.FramesPerBuffer,
80- Channels: format.ChannelCount,68+ SampleFormat: pwpipewire.SampleFormatF32, UnderrunPolicy: pwpipewire.UnderrunFillSilence,
81- FramesPerBuffer: format.FramesPerBuffer,
82- SampleFormat: pwpipewire.SampleFormatF32,
83- UnderrunPolicy: pwpipewire.UnderrunFillSilence,
84 }69 }
85-70+ player, err := f.createPlayer(config, pwpipewire.PlayerCallbacks{Fill: adapter.fill})
86- callbacks := pwpipewire.PlayerCallbacks{
87- Fill: adapter.fill,
88- }
89-
90- player, err := f.createPlayer(config, callbacks)
91 if err != nil {71 if err != nil {
92 log.Warn().Err(err).Msg("pipewire: failed to create player")72 log.Warn().Err(err).Msg("pipewire: failed to create player")
93 return nil, fmt.Errorf("failed to create pipewire player: %w", err)73 return nil, fmt.Errorf("failed to create pipewire player: %w", err)
94 }74 }
95 adapter.player = player75 adapter.player = player
96-
97- // Start the player so the Fill callback begins being called.
98 if err := player.Start(); err != nil {76 if err := player.Start(); err != nil {
99 _ = player.Close()77 _ = player.Close()
100 log.Warn().Err(err).Msg("pipewire: failed to start player")78 log.Warn().Err(err).Msg("pipewire: failed to start player")
101 return nil, fmt.Errorf("failed to start pipewire player: %w", err)79 return nil, fmt.Errorf("failed to start pipewire player: %w", err)
102 }80 }
103-
104 log.Info().Msg("pipewire: player started successfully")81 log.Info().Msg("pipewire: player started successfully")
105-
106 return adapter, nil82 return adapter, nil
107 }83 }
108 84
109-// validateFormat checks if the audio format parameters are valid.
110 func validateFormat(format port.AudioStreamFormat) error {85 func validateFormat(format port.AudioStreamFormat) error {
111- // Sample rate must be reasonable (8kHz to 384kHz)
112 if format.SampleRate < 8000 || format.SampleRate > 384000 {86 if format.SampleRate < 8000 || format.SampleRate > 384000 {
113- return fmt.Errorf("%w: sample rate %d Hz out of range [8000, 384000]",87+ return fmt.Errorf("%w: sample rate %d Hz out of range [8000, 384kHz]", ErrInvalidFormat, format.SampleRate)
114- ErrInvalidFormat, format.SampleRate)
115 }88 }
116-
117- // Channel count must be 1 to 1024 (reasonable upper limit)
118 if format.ChannelCount < 1 || format.ChannelCount > 1024 {89 if format.ChannelCount < 1 || format.ChannelCount > 1024 {
119- return fmt.Errorf("%w: channel count %d out of range [1, 1024]",90+ return fmt.Errorf("%w: channel count %d out of range [1, 1024]", ErrInvalidFormat, format.ChannelCount)
120- ErrInvalidFormat, format.ChannelCount)
121 }91 }
122-
123- // Frames per buffer must be positive and reasonable
124 if format.FramesPerBuffer < 1 || format.FramesPerBuffer > 8192 {92 if format.FramesPerBuffer < 1 || format.FramesPerBuffer > 8192 {
125- return fmt.Errorf("%w: frames per buffer %d out of range [1, 8192]",93+ return fmt.Errorf("%w: frames per buffer %d out of range [1, 8192]", ErrInvalidFormat, format.FramesPerBuffer)
126- ErrInvalidFormat, format.FramesPerBuffer)
127 }94 }
128-
129 return nil95 return nil
130 }96 }
131 97
132-// playerStreamAdapter bridges Dumber's push-style Write(samples) interface98+// audioPacket is a reusable, adapter-owned packet slot. Its backing arrays are
133-// to purego-pipewire's pull-style Fill callback.99+// allocated only when the stream is created and never alias CEF callback data.
134-//100+type audioPacket struct {
135-// Write() sends sample buffers into a channel. The Fill callback drains them101+ samples [][]float32
136-// into the PCMBuffer that purego-pipewire passes in. If no data is available,102+ channels int
137-// the player's UnderrunFillSilence policy handles it.103+ frames int
104+}
105+
106+// playerStreamAdapter bridges callback-owned CEF audio to PipeWire's Fill
107+// callback. available and ready form a bounded reusable packet ring.
138 type playerStreamAdapter struct {108 type playerStreamAdapter struct {
139 player pwpipewire.Player109 player pwpipewire.Player
140 ctx context.Context110 ctx context.Context
141- buf chan [][]float32 // push→pull handoff channel111+
112+ available chan *audioPacket
113+ ready chan *audioPacket
114+ // TryLock serializes only simultaneous producers. The CEF callback never
115+ // waits: contention and saturation both drop the packet. Fill never takes it.
116+ writeMu sync.Mutex
117+ closed atomic.Bool
142 118
143 closeOnce sync.Once119 closeOnce sync.Once
144- closed chan struct{} // closed when Close is called
145 120
146- // Diagnostic counters (atomic, no mutex needed).121+ fillCount atomic.Uint64
147- fillCount atomic.Uint64 // total Fill callbacks122+ underrunCount atomic.Uint64
148- underrunCount atomic.Uint64 // Fill callbacks with no data available123+ dropCount atomic.Uint64
149- dropCount atomic.Uint64 // Write calls that dropped data (buffer full)124+}
125+
126+func newPlayerStreamAdapter(ctx context.Context, channels, frames int) *playerStreamAdapter {
127+ a := &playerStreamAdapter{
128+ ctx: ctx, available: make(chan *audioPacket, packetQueueCapacity), ready: make(chan *audioPacket, packetQueueCapacity),
129+ }
130+ for range packetQueueCapacity {
131+ packet := &audioPacket{samples: make([][]float32, channels)}
132+ for ch := range packet.samples {
133+ packet.samples[ch] = make([]float32, frames)
134+ }
135+ a.available <- packet
136+ }
137+ return a
150 }138 }
151 139
152-// Write sends audio samples to the PipeWire output.140+// Write makes exactly one owned copy for each accepted packet. It never blocks:
153-// Samples are provided as [channel][frame]float32 matching CEF's planar format.141+// full queues or another producer cause a silent drop, preserving CEF timing.
154-// If the internal buffer is full the packet is dropped silently so the caller142+// The supplied slices are callback-owned and are not retained after return.
155-// (CEF audio callback thread) is never blocked.
156-// Safe for concurrent use with Close.
157 func (a *playerStreamAdapter) Write(samples [][]float32) error {143 func (a *playerStreamAdapter) Write(samples [][]float32) error {
158- // Fast path: already closed.144+ if a.closed.Load() {
159- select {
160- case <-a.closed:
161 return ErrStreamClosed145 return ErrStreamClosed
162- default:
163 }146 }
164- // Non-blocking send: drop packet when buffer is full.147+ if !a.writeMu.TryLock() {
165- select {148+ a.recordDrop()
166- case a.buf <- samples:
167 return nil149 return nil
168- case <-a.closed:150+ }
151+ defer a.writeMu.Unlock()
152+ if a.closed.Load() {
169 return ErrStreamClosed153 return ErrStreamClosed
170- default:154+ }
171- // Buffer full — drop to avoid blocking the CEF audio thread.155+
172- n := a.dropCount.Add(1)156+ select {
173- if n == 1 || n&(n-1) == 0 { // first drop, then powers of two157+ case packet := <-a.available:
174- if a.ctx != nil {158+ copyPacket(packet, samples)
175- logging.FromContext(a.ctx).Warn().159+ // A close racing after the first check must not publish a packet to a
176- Uint64("total_drops", n).160+ // stopped player. Returning the slot keeps the ring bounded.
177- Msg("pipewire: dropped audio packet (buffer full)")161+ if a.closed.Load() {
178- }162+ a.available <- packet
163+ return ErrStreamClosed
164+ }
165+ select {
166+ case a.ready <- packet:
167+ return nil
168+ default:
169+ // This cannot occur while ring invariants hold, but keep Write
170+ // non-blocking if an implementation change violates them.
171+ a.available <- packet
172+ a.recordDrop()
173+ return nil
179 }174 }
175+ default:
176+ a.recordDrop()
180 return nil177 return nil
181 }178 }
182 }179 }
183 180
184-// fill is the PlayerCallbacks.Fill function called by purego-pipewire181+func copyPacket(packet *audioPacket, samples [][]float32) {
185-// when it needs audio data. It pulls from the internal buffer channel.182+ packet.channels = min(len(packet.samples), len(samples))
183+ packet.frames = 0
184+ if packet.channels == 0 {
185+ return
186+ }
187+ frames := len(packet.samples[0])
188+ for ch := 0; ch < packet.channels; ch++ {
189+ n := min(len(packet.samples[ch]), len(samples[ch]))
190+ if n < frames {
191+ frames = n
192+ }
193+ }
194+ packet.frames = frames
195+ for ch := 0; ch < packet.channels; ch++ {
196+ copy(packet.samples[ch][:frames], samples[ch][:frames])
197+ }
198+}
199+
200+func (a *playerStreamAdapter) recordDrop() {
201+ // Write runs on CEF's real-time callback thread. Diagnostics must stay
202+ // allocation-free and must not synchronously emit logs from that path.
203+ a.dropCount.Add(1)
204+}
205+
206+// fill is the PlayerCallbacks.Fill function. It performs the only post-handoff
207+// full-packet copy, directly into PipeWire's destination buffer.
186 func (a *playerStreamAdapter) fill(pcm *pwpipewire.PCMBuffer) (int, error) {208 func (a *playerStreamAdapter) fill(pcm *pwpipewire.PCMBuffer) (int, error) {
187 select {209 select {
188- case samples := <-a.buf:210+ case packet := <-a.ready:
189 fillNum := a.fillCount.Add(1)211 fillNum := a.fillCount.Add(1)
190 if fillNum == 1 && a.ctx != nil {212 if fillNum == 1 && a.ctx != nil {
191- logging.FromContext(a.ctx).Info().213+ logging.FromContext(a.ctx).Info().Msg("pipewire: first Fill callback with data")
192- Msg("pipewire: first Fill callback with data")
193 }214 }
194- // Copy available samples into the PCM buffer215+ copied := copyToPCM(pcm, packet.samples[:packet.channels], packet.frames)
195- copied := copyToPCM(pcm, samples)216+ a.available <- packet
196 return copied, nil217 return copied, nil
197 default:218 default:
198- // No data available — return 0 frames, let underrun policy handle it
199 underruns := a.underrunCount.Add(1)219 underruns := a.underrunCount.Add(1)
200 if underruns == 1 && a.ctx != nil {220 if underruns == 1 && a.ctx != nil {
201- logging.FromContext(a.ctx).Debug().221+ logging.FromContext(a.ctx).Debug().Msg("pipewire: first Fill underrun (no data available)")
202- Msg("pipewire: first Fill underrun (no data available)")
203 }222 }
204 return 0, nil223 return 0, nil
205 }224 }
206 }225 }
207 226
208-// copyToPCM copies planar samples into the PCMBuffer.227+func copyToPCM(pcm *pwpipewire.PCMBuffer, samples [][]float32, packetFrames int) int {
209-// Returns the number of frames actually copied.228+ if pcm == nil || len(samples) == 0 || packetFrames == 0 {
210-func copyToPCM(pcm *pwpipewire.PCMBuffer, samples [][]float32) int {
211- if pcm == nil || len(samples) == 0 {
212 return 0229 return 0
213 }230 }
214-
215- frames := pcm.Frames
216 channels := min(min(len(samples), pcm.Channels), len(pcm.Samples))231 channels := min(min(len(samples), pcm.Channels), len(pcm.Samples))
217 if channels == 0 {232 if channels == 0 {
218 return 0233 return 0
219 }234 }
220-235+ frames := min(pcm.Frames, packetFrames)
221- minFrames := frames236+ for ch := 0; ch < channels; ch++ {
222-237+ frames = min(frames, min(len(pcm.Samples[ch]), len(samples[ch])))
223- for ch := 0; ch < channels && ch < len(samples); ch++ {
224- dst := pcm.Samples[ch]
225- src := samples[ch]
226- if len(dst) == 0 || len(src) == 0 {
227- minFrames = 0
228- continue
229- }
230- srcFrames := len(src)
231- n := min(len(dst), min(srcFrames, frames))
232- if n < minFrames {
233- minFrames = n
234- }
235- copy(dst[:n], src[:n])
236 }238 }
237-239+ for ch := 0; ch < channels; ch++ {
238- return minFrames240+ copy(pcm.Samples[ch][:frames], samples[ch][:frames])
241+ }
242+ return frames
239 }243 }
240 244
241-// Close stops the player and releases resources.245+// Close stops the player. It is safe with Write: Write observes closed before
242-// Safe to call multiple times.246+// publishing, and no channel is closed while a callback may be selecting on it.
243 func (a *playerStreamAdapter) Close() error {247 func (a *playerStreamAdapter) Close() error {
244 var err error248 var err error
245 a.closeOnce.Do(func() {249 a.closeOnce.Do(func() {
250+ a.closed.Store(true)
246 if a.ctx != nil {251 if a.ctx != nil {
247 logging.FromContext(a.ctx).Info().252 logging.FromContext(a.ctx).Info().
248 Uint64("fills", a.fillCount.Load()).253 Uint64("fills", a.fillCount.Load()).
@@ -250,12 +255,7 @@ func (a *playerStreamAdapter) Close() error {
250 Uint64("drops", a.dropCount.Load()).255 Uint64("drops", a.dropCount.Load()).
251 Msg("pipewire: closing player")256 Msg("pipewire: closing player")
252 }257 }
253-
254- // Signal writers to stop
255- close(a.closed)
256-
257 if a.player != nil {258 if a.player != nil {
258- // Stop playback then close the player
259 if stopErr := a.player.Stop(); stopErr != nil {259 if stopErr := a.player.Stop(); stopErr != nil {
260 err = stopErr260 err = stopErr
261 }261 }
M internal/infrastructure/audio/pipewire/output_test.go +139 -494
@@ -1,11 +1,11 @@
1-// Package pipewire provides a PipeWire-based audio output implementation.
2-// This wraps purego-pipewire's high-level Player API.
31 package pipewire
42
53 import (
64 "context"
75 "errors"
6+ "sync"
87 "testing"
8+ "time"
99
1010 "github.com/bnema/dumber/internal/application/port"
1111 pipewiremocks "github.com/bnema/dumber/internal/infrastructure/audio/pipewire/mocks"
@@ -16,548 +16,193 @@ import (
1616
1717 type playerRecorder struct {
1818 *pipewiremocks.MockPlayer
19-
20- started bool
21- paused bool
22- stopped bool
23- closedFlag bool
24- state pwpipewire.PlayerState
25-
26- startErr error
27- stopErr error
28- closeErr error
19+ started, stopped, closedFlag bool
20+ startErr, stopErr, closeErr error
2921 }
3022
3123 func newPlayerRecorder(t *testing.T) *playerRecorder {
3224 t.Helper()
33-
34- recorder := &playerRecorder{MockPlayer: pipewiremocks.NewMockPlayer(t)}
35- recorder.EXPECT().Start().RunAndReturn(recorder.start).Maybe()
36- recorder.EXPECT().Pause().RunAndReturn(recorder.pause).Maybe()
37- recorder.EXPECT().Stop().RunAndReturn(recorder.stop).Maybe()
38- recorder.EXPECT().Close().RunAndReturn(recorder.close).Maybe()
39- recorder.EXPECT().State().RunAndReturn(recorder.currentState).Maybe()
40- return recorder
41-}
42-
43-func (p *playerRecorder) start() error {
44- if p.startErr != nil {
45- return p.startErr
46- }
47- p.started = true
48- p.state = pwpipewire.PlayerStatePlaying
49- return nil
50-}
51-
52-func (p *playerRecorder) pause() error {
53- p.paused = true
54- p.state = pwpipewire.PlayerStatePaused
55- return nil
56-}
57-
58-func (p *playerRecorder) stop() error {
59- if p.stopErr != nil {
60- return p.stopErr
61- }
62- p.stopped = true
63- p.state = pwpipewire.PlayerStateStopped
64- return nil
25+ p := &playerRecorder{MockPlayer: pipewiremocks.NewMockPlayer(t)}
26+ p.EXPECT().Start().RunAndReturn(func() error {
27+ if p.startErr != nil {
28+ return p.startErr
29+ }
30+ p.started = true
31+ return nil
32+ }).Maybe()
33+ p.EXPECT().Stop().RunAndReturn(func() error {
34+ if p.stopErr != nil {
35+ return p.stopErr
36+ }
37+ p.stopped = true
38+ return nil
39+ }).Maybe()
40+ p.EXPECT().Close().RunAndReturn(func() error {
41+ if p.closeErr != nil {
42+ return p.closeErr
43+ }
44+ p.closedFlag = true
45+ return nil
46+ }).Maybe()
47+ return p
6548 }
6649
67-func (p *playerRecorder) close() error {
68- if p.closeErr != nil {
69- return p.closeErr
70- }
71- p.closedFlag = true
72- p.state = pwpipewire.PlayerStateClosed
73- return nil
50+func testAdapter(frames int) *playerStreamAdapter {
51+ return newPlayerStreamAdapter(context.Background(), 2, frames)
7452 }
7553
76-func (p *playerRecorder) currentState() pwpipewire.PlayerState {
77- return p.state
54+func stereoPCM(frames int) *pwpipewire.PCMBuffer {
55+ return &pwpipewire.PCMBuffer{Frames: frames, Channels: 2, Samples: [][]float32{make([]float32, frames), make([]float32, frames)}}
7856 }
7957
80-// --- Factory tests ---
81-
82-func TestFactory_NewStream_ValidFormat_CreatesStream(t *testing.T) {
83- var capturedConfig pwpipewire.PlayerConfig
58+func TestFactory_NewStreamValidatesAndStartsPlayer(t *testing.T) {
8459 player := newPlayerRecorder(t)
85-
86- factory := &Factory{
87- createPlayer: func(config pwpipewire.PlayerConfig, _ pwpipewire.PlayerCallbacks) (pwpipewire.Player, error) {
88- capturedConfig = config
89- return player, nil
90- },
91- }
92-
93- format := port.AudioStreamFormat{
94- SampleRate: 48000,
95- ChannelCount: 2,
96- FramesPerBuffer: 512,
97- }
98-
99- stream, err := factory.NewStream(context.Background(), format)
60+ factory := &Factory{createPlayer: func(config pwpipewire.PlayerConfig, callbacks pwpipewire.PlayerCallbacks) (pwpipewire.Player, error) {
61+ assert.Equal(t, 48000, config.SampleRate)
62+ assert.Equal(t, 2, config.Channels)
63+ assert.NotNil(t, callbacks.Fill)
64+ return player, nil
65+ }}
66+ stream, err := factory.NewStream(context.Background(), port.AudioStreamFormat{SampleRate: 48000, ChannelCount: 2, FramesPerBuffer: 512})
10067 require.NoError(t, err)
101- require.NotNil(t, stream)
102-
103- // Verify config mapping
104- assert.Equal(t, 48000, capturedConfig.SampleRate)
105- assert.Equal(t, 2, capturedConfig.Channels)
106- assert.Equal(t, 512, capturedConfig.FramesPerBuffer)
107- assert.Equal(t, pwpipewire.SampleFormatF32, capturedConfig.SampleFormat)
108- assert.Equal(t, pwpipewire.UnderrunFillSilence, capturedConfig.UnderrunPolicy)
109-
110- // Verify player was started
11168 assert.True(t, player.started)
112-
113- // Clean up
11469 require.NoError(t, stream.Close())
11570 }
11671
117-func TestFactory_NewStream_InvalidFormat_ReturnsError(t *testing.T) {
118- factory := &Factory{
119- createPlayer: func(_ pwpipewire.PlayerConfig, _ pwpipewire.PlayerCallbacks) (pwpipewire.Player, error) {
120- t.Fatal("createPlayer should not be called for invalid format")
121- return nil, nil
122- },
123- }
124-
125- tests := []struct {
126- name string
127- format port.AudioStreamFormat
128- }{
129- {
130- name: "zero sample rate",
131- format: port.AudioStreamFormat{SampleRate: 0, ChannelCount: 2, FramesPerBuffer: 512},
132- },
133- {
134- name: "zero channels",
135- format: port.AudioStreamFormat{SampleRate: 48000, ChannelCount: 0, FramesPerBuffer: 512},
136- },
137- {
138- name: "negative sample rate",
139- format: port.AudioStreamFormat{SampleRate: -48000, ChannelCount: 2, FramesPerBuffer: 512},
140- },
141- {
142- name: "negative channels",
143- format: port.AudioStreamFormat{SampleRate: 48000, ChannelCount: -2, FramesPerBuffer: 512},
144- },
145- {
146- name: "excessive sample rate",
147- format: port.AudioStreamFormat{SampleRate: 1000000, ChannelCount: 2, FramesPerBuffer: 512},
148- },
149- {
150- name: "excessive channels",
151- format: port.AudioStreamFormat{SampleRate: 48000, ChannelCount: 1025, FramesPerBuffer: 512},
152- },
153- {
154- name: "zero frames per buffer",
155- format: port.AudioStreamFormat{SampleRate: 48000, ChannelCount: 2, FramesPerBuffer: 0},
156- },
157- {
158- name: "excessive frames per buffer",
159- format: port.AudioStreamFormat{SampleRate: 48000, ChannelCount: 2, FramesPerBuffer: 10000},
160- },
161- }
162-
163- for _, tt := range tests {
164- t.Run(tt.name, func(t *testing.T) {
165- stream, err := factory.NewStream(context.Background(), tt.format)
166- require.Error(t, err)
167- assert.Nil(t, stream)
168- assert.ErrorIs(t, err, ErrInvalidFormat)
169- })
170- }
171-}
172-
173-func TestFactory_NewStream_PlayerCreationFails_ReturnsError(t *testing.T) {
174- expectedErr := errors.New("pipewire not available")
175- factory := &Factory{
176- createPlayer: func(_ pwpipewire.PlayerConfig, _ pwpipewire.PlayerCallbacks) (pwpipewire.Player, error) {
177- return nil, expectedErr
178- },
72+func TestFactory_NewStreamRejectsInvalidFormat(t *testing.T) {
73+ factory := &Factory{createPlayer: func(pwpipewire.PlayerConfig, pwpipewire.PlayerCallbacks) (pwpipewire.Player, error) {
74+ t.Fatal("unexpected player")
75+ return nil, nil
76+ }}
77+ for _, format := range []port.AudioStreamFormat{
78+ {SampleRate: 0, ChannelCount: 2, FramesPerBuffer: 512},
79+ {SampleRate: 48000, ChannelCount: 0, FramesPerBuffer: 512},
80+ {SampleRate: 48000, ChannelCount: 2, FramesPerBuffer: 0},
81+ } {
82+ stream, err := factory.NewStream(context.Background(), format)
83+ require.ErrorIs(t, err, ErrInvalidFormat)
84+ assert.Nil(t, stream)
17985 }
180-
181- format := port.AudioStreamFormat{SampleRate: 48000, ChannelCount: 2, FramesPerBuffer: 512}
182- stream, err := factory.NewStream(context.Background(), format)
183-
184- require.Error(t, err)
185- assert.Nil(t, stream)
186- assert.ErrorIs(t, err, expectedErr)
18786 }
18887
189-func TestFactory_NewStream_PlayerStartFails_ClosesAndReturnsError(t *testing.T) {
88+func TestFactory_NewStreamClosesPlayerWhenStartFails(t *testing.T) {
19089 player := newPlayerRecorder(t)
19190 player.startErr = errors.New("start failed")
192- factory := &Factory{
193- createPlayer: func(_ pwpipewire.PlayerConfig, _ pwpipewire.PlayerCallbacks) (pwpipewire.Player, error) {
194- return player, nil
195- },
196- }
197-
198- format := port.AudioStreamFormat{SampleRate: 48000, ChannelCount: 2, FramesPerBuffer: 512}
199- stream, err := factory.NewStream(context.Background(), format)
200-
201- require.Error(t, err)
202- assert.Nil(t, stream)
203- assert.True(t, player.closedFlag, "player should be closed after start failure")
204-}
205-
206-func TestFactory_NewStream_ContextCancelled_ReturnsError(t *testing.T) {
207- factory := &Factory{
208- createPlayer: func(_ pwpipewire.PlayerConfig, _ pwpipewire.PlayerCallbacks) (pwpipewire.Player, error) {
209- t.Fatal("createPlayer should not be called when context is canceled")
210- return nil, nil
211- },
212- }
213-
214- ctx, cancel := context.WithCancel(context.Background())
215- cancel()
216-
217- format := port.AudioStreamFormat{SampleRate: 48000, ChannelCount: 2, FramesPerBuffer: 512}
218- stream, err := factory.NewStream(ctx, format)
219-
91+ factory := &Factory{createPlayer: func(pwpipewire.PlayerConfig, pwpipewire.PlayerCallbacks) (pwpipewire.Player, error) {
92+ return player, nil
93+ }}
94+ stream, err := factory.NewStream(context.Background(), port.AudioStreamFormat{SampleRate: 48000, ChannelCount: 2, FramesPerBuffer: 512})
22095 require.Error(t, err)
22196 assert.Nil(t, stream)
222-}
223-
224-func TestFactory_NewStream_MultipleFormats_Accepted(t *testing.T) {
225- formats := []port.AudioStreamFormat{
226- {SampleRate: 48000, ChannelCount: 2, FramesPerBuffer: 512},
227- {SampleRate: 44100, ChannelCount: 2, FramesPerBuffer: 512},
228- {SampleRate: 48000, ChannelCount: 1, FramesPerBuffer: 256},
229- {SampleRate: 96000, ChannelCount: 2, FramesPerBuffer: 1024},
230- {SampleRate: 48000, ChannelCount: 6, FramesPerBuffer: 512},
231- }
232-
233- for _, fmt := range formats {
234- t.Run("", func(t *testing.T) {
235- factory := &Factory{
236- createPlayer: func(config pwpipewire.PlayerConfig, _ pwpipewire.PlayerCallbacks) (pwpipewire.Player, error) {
237- assert.Equal(t, fmt.SampleRate, config.SampleRate)
238- assert.Equal(t, fmt.ChannelCount, config.Channels)
239- assert.Equal(t, fmt.FramesPerBuffer, config.FramesPerBuffer)
240- return newPlayerRecorder(t), nil
241- },
242- }
243-
244- stream, err := factory.NewStream(context.Background(), fmt)
245- require.NoError(t, err)
246- assert.NotNil(t, stream)
247-
248- _ = stream.Close()
249- })
250- }
251-}
252-
253-func TestFactory_ImplementsPortInterface(_ *testing.T) {
254- var _ port.AudioOutputFactory = (*Factory)(nil)
255-}
256-
257-// --- Stream adapter tests ---
258-
259-func TestPlayerStreamAdapter_Write_SendsToBuffer(t *testing.T) {
260- player := newPlayerRecorder(t)
261- adapter := &playerStreamAdapter{
262- player: player,
263- buf: make(chan [][]float32, 4),
264- closed: make(chan struct{}),
265- }
266-
267- samples := [][]float32{
268- {0.1, 0.2},
269- {0.3, 0.4},
270- }
271-
272- err := adapter.Write(samples)
273- require.NoError(t, err)
274-
275- // Verify data is in the channel
276- received := <-adapter.buf
277- assert.Equal(t, samples, received)
278-}
279-
280-func TestPlayerStreamAdapter_Write_AfterClose_ReturnsError(t *testing.T) {
281- player := newPlayerRecorder(t)
282- adapter := &playerStreamAdapter{
283- player: player,
284- buf: make(chan [][]float32, 4),
285- closed: make(chan struct{}),
286- }
287-
288- require.NoError(t, adapter.Close())
289-
290- err := adapter.Write([][]float32{{0.1}})
291- assert.ErrorIs(t, err, ErrStreamClosed)
292-}
293-
294-func TestPlayerStreamAdapter_Close_StopsAndClosesPlayer(t *testing.T) {
295- player := newPlayerRecorder(t)
296- adapter := &playerStreamAdapter{
297- player: player,
298- buf: make(chan [][]float32, 4),
299- closed: make(chan struct{}),
300- }
301-
302- err := adapter.Close()
303- require.NoError(t, err)
304-
305- assert.True(t, player.stopped)
30697 assert.True(t, player.closedFlag)
30798 }
30899
309-func TestPlayerStreamAdapter_Close_IsIdempotent(t *testing.T) {
310- player := newPlayerRecorder(t)
311- adapter := &playerStreamAdapter{
312- player: player,
313- buf: make(chan [][]float32, 4),
314- closed: make(chan struct{}),
315- }
316-
317- require.NoError(t, adapter.Close())
318- require.NoError(t, adapter.Close())
319-}
320-
321-func TestPlayerStreamAdapter_Close_PropagatesStopError(t *testing.T) {
322- player := newPlayerRecorder(t)
323- player.stopErr = errors.New("stop error")
324- adapter := &playerStreamAdapter{
325- player: player,
326- buf: make(chan [][]float32, 4),
327- closed: make(chan struct{}),
328- }
329-
330- err := adapter.Close()
331- require.Error(t, err)
332- assert.Contains(t, err.Error(), "stop error")
333-}
334-
335-func TestPlayerStreamAdapter_Close_PropagatesCloseError(t *testing.T) {
336- player := newPlayerRecorder(t)
337- player.closeErr = errors.New("close error")
338- adapter := &playerStreamAdapter{
339- player: player,
340- buf: make(chan [][]float32, 4),
341- closed: make(chan struct{}),
342- }
343-
344- err := adapter.Close()
345- require.Error(t, err)
346- assert.Contains(t, err.Error(), "close error")
347-}
348-
349-// --- Fill callback tests ---
350-
351-func TestFill_PullsFromBuffer(t *testing.T) {
352- adapter := &playerStreamAdapter{
353- buf: make(chan [][]float32, 4),
354- }
355-
356- // Push samples
357- samples := [][]float32{
358- {0.1, 0.2, 0.3},
359- {0.4, 0.5, 0.6},
360- }
361- adapter.buf <- samples
362-
363- // Simulate Fill call
364- pcm := &pwpipewire.PCMBuffer{
365- Frames: 3,
366- Channels: 2,
367- Stride: 1,
368- Samples: [][]float32{
369- make([]float32, 3),
370- make([]float32, 3),
371- },
372- }
373-
100+func TestPlayerStreamAdapter_FillGetsOwnedPacketAfterCallbackMutation(t *testing.T) {
101+ adapter := testAdapter(3)
102+ samples := [][]float32{{1, 2, 3}, {4, 5, 6}}
103+ require.NoError(t, adapter.Write(samples))
104+ samples[0][0], samples[1][0] = 99, 99
105+ pcm := stereoPCM(3)
374106 frames, err := adapter.fill(pcm)
375107 require.NoError(t, err)
376108 assert.Equal(t, 3, frames)
377- assert.InEpsilon(t, 0.1, pcm.Samples[0][0], 0.0001)
378- assert.InEpsilon(t, 0.4, pcm.Samples[1][0], 0.0001)
379-}
380-
381-func TestFill_ReturnsZeroWhenEmpty(t *testing.T) {
382- adapter := &playerStreamAdapter{
383- buf: make(chan [][]float32, 4),
384- }
385-
386- pcm := &pwpipewire.PCMBuffer{
387- Frames: 256,
388- Channels: 2,
389- Stride: 1,
390- Samples: [][]float32{
391- make([]float32, 256),
392- make([]float32, 256),
393- },
394- }
395-
396- frames, err := adapter.fill(pcm)
397- require.NoError(t, err)
398- assert.Equal(t, 0, frames)
399-}
400-
401-func TestCopyToPCM_HandlesChannelMismatch(t *testing.T) {
402- // More source channels than PCM channels — truncates
403- pcm := &pwpipewire.PCMBuffer{
404- Frames: 2,
405- Channels: 1,
406- Samples: [][]float32{make([]float32, 2)},
407- }
408- samples := [][]float32{
409- {1.0, 2.0},
410- {3.0, 4.0}, // should be ignored
411- }
412- frames := copyToPCM(pcm, samples)
413- assert.Equal(t, 2, frames)
414- assert.InEpsilon(t, 1.0, pcm.Samples[0][0], 0.0001)
109+ assert.InEpsilon(t, 1, pcm.Samples[0][0], 0.0001)
110+ assert.InEpsilon(t, 4, pcm.Samples[1][0], 0.0001)
415111 }
416112
417-func TestCopyToPCM_HandlesFrameMismatch(t *testing.T) {
418- // Fewer source frames than PCM expects
419- pcm := &pwpipewire.PCMBuffer{
420- Frames: 4,
421- Channels: 1,
422- Samples: [][]float32{make([]float32, 4)},
113+func TestPlayerStreamAdapter_SaturationDropsWithoutBlocking(t *testing.T) {
114+ adapter := testAdapter(2)
115+ for i := 0; i < packetQueueCapacity; i++ {
116+ require.NoError(t, adapter.Write([][]float32{{float32(i)}, {float32(i)}}))
423117 }
424- samples := [][]float32{
425- {1.0, 2.0}, // only 2 frames
118+ done := make(chan error, 1)
119+ go func() { done <- adapter.Write([][]float32{{99}, {99}}) }()
120+ select {
121+ case err := <-done:
122+ require.NoError(t, err)
123+ case <-time.After(time.Second):
124+ t.Fatal("Write blocked while the bounded queue was saturated")
426125 }
427- frames := copyToPCM(pcm, samples)
428- assert.Equal(t, 2, frames) // reports partial fill
429-}
430-
431-func TestCopyToPCM_NilPCM_ReturnsZero(t *testing.T) {
432- assert.Equal(t, 0, copyToPCM(nil, [][]float32{{1.0}}))
126+ assert.Equal(t, uint64(1), adapter.dropCount.Load())
127+ assert.Len(t, adapter.ready, packetQueueCapacity)
433128 }
434129
435-func TestCopyToPCM_EmptySamples_ReturnsZero(t *testing.T) {
436- pcm := &pwpipewire.PCMBuffer{Frames: 1, Channels: 1, Samples: [][]float32{make([]float32, 1)}}
437- assert.Equal(t, 0, copyToPCM(pcm, nil))
438- assert.Equal(t, 0, copyToPCM(pcm, [][]float32{}))
130+func TestPlayerStreamAdapter_EmptyFillCountsUnderrun(t *testing.T) {
131+ adapter := testAdapter(2)
132+ frames, err := adapter.fill(stereoPCM(2))
133+ require.NoError(t, err)
134+ assert.Zero(t, frames)
135+ assert.Equal(t, uint64(1), adapter.underrunCount.Load())
439136 }
440137
441-// --- Non-blocking Write tests ---
442-
443-func TestPlayerStreamAdapter_Write_DropsWhenBufferFull(t *testing.T) {
444- adapter := &playerStreamAdapter{
445- player: newPlayerRecorder(t),
446- buf: make(chan [][]float32, 4),
447- closed: make(chan struct{}),
448- }
449-
450- // Fill the buffer to capacity
451- for i := range 4 {
452- require.NoError(t, adapter.Write([][]float32{{float32(i)}}))
453- }
454-
455- // The 5th write must NOT block — it should drop the packet and return nil.
456- err := adapter.Write([][]float32{{99.0}})
457- require.NoError(t, err, "Write must not return an error when dropping due to full buffer")
458-
459- // Verify the drop was counted
460- assert.Equal(t, uint64(1), adapter.dropCount.Load(), "one packet should have been dropped")
461-
462- // Verify buffer still contains the original 4 packets (not the dropped one)
463- assert.Len(t, adapter.buf, 4)
138+func TestPlayerStreamAdapter_CopyToPCMHandlesMismatches(t *testing.T) {
139+ pcm := &pwpipewire.PCMBuffer{Frames: 4, Channels: 1, Samples: [][]float32{make([]float32, 4)}}
140+ assert.Equal(t, 2, copyToPCM(pcm, [][]float32{{1, 2}, {3, 4}}, 2))
141+ assert.InEpsilon(t, 1, pcm.Samples[0][0], 0.0001)
142+ assert.Zero(t, copyToPCM(nil, [][]float32{{1}}, 1))
464143 }
465144
466-func TestPlayerStreamAdapter_Write_DropsLogOnlyOnFirstAndPowerOfTwo(t *testing.T) {
467- adapter := &playerStreamAdapter{
468- player: newPlayerRecorder(t),
469- buf: make(chan [][]float32, 1), // tiny buffer to force drops
470- closed: make(chan struct{}),
471- }
472-
473- // Fill buffer
474- require.NoError(t, adapter.Write([][]float32{{0.0}}))
475-
476- // Drop 10 packets
477- for i := range 10 {
478- _ = adapter.Write([][]float32{{float32(i)}})
479- }
480-
481- // Verify all drops were counted
482- assert.Equal(t, uint64(10), adapter.dropCount.Load())
145+func TestPlayerStreamAdapter_CloseVsWrite(t *testing.T) {
146+ adapter := testAdapter(2)
147+ player := newPlayerRecorder(t)
148+ adapter.player = player
149+ require.NoError(t, adapter.Close())
150+ require.ErrorIs(t, adapter.Write([][]float32{{1}, {2}}), ErrStreamClosed)
151+ assert.True(t, player.stopped)
152+ assert.True(t, player.closedFlag)
483153 }
484154
485-// --- Concurrent Write / Close tests ---
486-
487155 func TestPlayerStreamAdapter_ConcurrentWriteClose_NoRace(t *testing.T) {
488- player := newPlayerRecorder(t)
489- adapter := &playerStreamAdapter{
490- player: player,
491- buf: make(chan [][]float32, 4),
492- closed: make(chan struct{}),
493- }
494-
495- done := make(chan struct{})
496- go func() {
497- defer close(done)
498- for i := range 100 {
499- err := adapter.Write([][]float32{{float32(i)}})
500- if err != nil {
501- assert.ErrorIs(t, err, ErrStreamClosed)
502- return
156+ adapter := testAdapter(4)
157+ var writers sync.WaitGroup
158+ for range 4 {
159+ writers.Add(1)
160+ go func() {
161+ defer writers.Done()
162+ for range 1000 {
163+ err := adapter.Write([][]float32{{1, 2}, {3, 4}})
164+ if err != nil {
165+ assert.ErrorIs(t, err, ErrStreamClosed)
166+ return
167+ }
503168 }
504- }
505- }()
506-
507- // Let a few writes land, then close.
169+ }()
170+ }
508171 adapter.Close()
509- <-done
510-}
511-
512-// --- NewFactory tests ---
513-
514-func TestNewFactory_Succeeds(t *testing.T) {
515- factory, err := NewFactory()
516- require.NoError(t, err)
517- require.NotNil(t, factory)
518- assert.NotNil(t, factory.createPlayer)
172+ writers.Wait()
519173 }
520174
521-// --- Integration-style test: full write-through ---
522-
523175 func TestStream_WriteFlowsThroughToFill(t *testing.T) {
524- var capturedCallbacks pwpipewire.PlayerCallbacks
176+ var callbacks pwpipewire.PlayerCallbacks
525177 player := newPlayerRecorder(t)
526-
527- factory := &Factory{
528- createPlayer: func(_ pwpipewire.PlayerConfig, callbacks pwpipewire.PlayerCallbacks) (pwpipewire.Player, error) {
529- capturedCallbacks = callbacks
530- return player, nil
531- },
532- }
533-
534- format := port.AudioStreamFormat{SampleRate: 48000, ChannelCount: 2, FramesPerBuffer: 4}
535- stream, err := factory.NewStream(context.Background(), format)
178+ factory := &Factory{createPlayer: func(_ pwpipewire.PlayerConfig, got pwpipewire.PlayerCallbacks) (pwpipewire.Player, error) {
179+ callbacks = got
180+ return player, nil
181+ }}
182+ stream, err := factory.NewStream(context.Background(), port.AudioStreamFormat{SampleRate: 48000, ChannelCount: 2, FramesPerBuffer: 4})
536183 require.NoError(t, err)
537-
538- // Write samples through the stream
539- samples := [][]float32{
540- {0.1, 0.2, 0.3, 0.4},
541- {0.5, 0.6, 0.7, 0.8},
542- }
543- require.NoError(t, stream.Write(samples))
544-
545- // Simulate the Fill callback being called by the player
546- pcm := &pwpipewire.PCMBuffer{
547- Frames: 4,
548- Channels: 2,
549- Stride: 1,
550- Samples: [][]float32{
551- make([]float32, 4),
552- make([]float32, 4),
553- },
554- }
555-
556- frames, err := capturedCallbacks.Fill(pcm)
184+ require.NoError(t, stream.Write([][]float32{{.1, .2, .3, .4}, {.5, .6, .7, .8}}))
185+ pcm := stereoPCM(4)
186+ frames, err := callbacks.Fill(pcm)
557187 require.NoError(t, err)
558188 assert.Equal(t, 4, frames)
559- assert.InEpsilon(t, 0.1, pcm.Samples[0][0], 0.0001)
560- assert.InEpsilon(t, 0.5, pcm.Samples[1][0], 0.0001)
561-
189+ assert.InEpsilon(t, .1, pcm.Samples[0][0], .0001)
190+ assert.InEpsilon(t, .5, pcm.Samples[1][0], .0001)
562191 require.NoError(t, stream.Close())
563192 }
193+
194+func BenchmarkPlayerStreamAdapterStereo(b *testing.B) {
195+ adapter := testAdapter(512)
196+ samples := [][]float32{make([]float32, 512), make([]float32, 512)}
197+ pcm := stereoPCM(512)
198+ b.ReportAllocs()
199+ b.SetBytes(2 * 512 * 4)
200+ for i := 0; i < b.N; i++ {
201+ if err := adapter.Write(samples); err != nil {
202+ b.Fatal(err)
203+ }
204+ if _, err := adapter.fill(pcm); err != nil {
205+ b.Fatal(err)
206+ }
207+ }
208+}
@@ -1,11 +1,11 @@
1-// Package pipewire provides a PipeWire-based audio output implementation.
2-// This wraps purego-pipewire's high-level Player API.
3 package pipewire1 package pipewire
4 2
5 import (3 import (
6 "context"4 "context"
7 "errors"5 "errors"
6+ "sync"
8 "testing"7 "testing"
8+ "time"
9 9
10 "github.com/bnema/dumber/internal/application/port"10 "github.com/bnema/dumber/internal/application/port"
11 pipewiremocks "github.com/bnema/dumber/internal/infrastructure/audio/pipewire/mocks"11 pipewiremocks "github.com/bnema/dumber/internal/infrastructure/audio/pipewire/mocks"
@@ -16,548 +16,193 @@ import (
16 16
17 type playerRecorder struct {17 type playerRecorder struct {
18 *pipewiremocks.MockPlayer18 *pipewiremocks.MockPlayer
19-19+ started, stopped, closedFlag bool
20- started bool20+ startErr, stopErr, closeErr error
21- paused bool
22- stopped bool
23- closedFlag bool
24- state pwpipewire.PlayerState
25-
26- startErr error
27- stopErr error
28- closeErr error
29 }21 }
30 22
31 func newPlayerRecorder(t *testing.T) *playerRecorder {23 func newPlayerRecorder(t *testing.T) *playerRecorder {
32 t.Helper()24 t.Helper()
33-25+ p := &playerRecorder{MockPlayer: pipewiremocks.NewMockPlayer(t)}
34- recorder := &playerRecorder{MockPlayer: pipewiremocks.NewMockPlayer(t)}26+ p.EXPECT().Start().RunAndReturn(func() error {
35- recorder.EXPECT().Start().RunAndReturn(recorder.start).Maybe()27+ if p.startErr != nil {
36- recorder.EXPECT().Pause().RunAndReturn(recorder.pause).Maybe()28+ return p.startErr
37- recorder.EXPECT().Stop().RunAndReturn(recorder.stop).Maybe()29+ }
38- recorder.EXPECT().Close().RunAndReturn(recorder.close).Maybe()30+ p.started = true
39- recorder.EXPECT().State().RunAndReturn(recorder.currentState).Maybe()31+ return nil
40- return recorder32+ }).Maybe()
41-}33+ p.EXPECT().Stop().RunAndReturn(func() error {
42-34+ if p.stopErr != nil {
43-func (p *playerRecorder) start() error {35+ return p.stopErr
44- if p.startErr != nil {36+ }
45- return p.startErr37+ p.stopped = true
46- }38+ return nil
47- p.started = true39+ }).Maybe()
48- p.state = pwpipewire.PlayerStatePlaying40+ p.EXPECT().Close().RunAndReturn(func() error {
49- return nil41+ if p.closeErr != nil {
50-}42+ return p.closeErr
51-43+ }
52-func (p *playerRecorder) pause() error {44+ p.closedFlag = true
53- p.paused = true45+ return nil
54- p.state = pwpipewire.PlayerStatePaused46+ }).Maybe()
55- return nil47+ return p
56-}
57-
58-func (p *playerRecorder) stop() error {
59- if p.stopErr != nil {
60- return p.stopErr
61- }
62- p.stopped = true
63- p.state = pwpipewire.PlayerStateStopped
64- return nil
65 }48 }
66 49
67-func (p *playerRecorder) close() error {50+func testAdapter(frames int) *playerStreamAdapter {
68- if p.closeErr != nil {51+ return newPlayerStreamAdapter(context.Background(), 2, frames)
69- return p.closeErr
70- }
71- p.closedFlag = true
72- p.state = pwpipewire.PlayerStateClosed
73- return nil
74 }52 }
75 53
76-func (p *playerRecorder) currentState() pwpipewire.PlayerState {54+func stereoPCM(frames int) *pwpipewire.PCMBuffer {
77- return p.state55+ return &pwpipewire.PCMBuffer{Frames: frames, Channels: 2, Samples: [][]float32{make([]float32, frames), make([]float32, frames)}}
78 }56 }
79 57
80-// --- Factory tests ---58+func TestFactory_NewStreamValidatesAndStartsPlayer(t *testing.T) {
81-
82-func TestFactory_NewStream_ValidFormat_CreatesStream(t *testing.T) {
83- var capturedConfig pwpipewire.PlayerConfig
84 player := newPlayerRecorder(t)59 player := newPlayerRecorder(t)
85-60+ factory := &Factory{createPlayer: func(config pwpipewire.PlayerConfig, callbacks pwpipewire.PlayerCallbacks) (pwpipewire.Player, error) {
86- factory := &Factory{61+ assert.Equal(t, 48000, config.SampleRate)
87- createPlayer: func(config pwpipewire.PlayerConfig, _ pwpipewire.PlayerCallbacks) (pwpipewire.Player, error) {62+ assert.Equal(t, 2, config.Channels)
88- capturedConfig = config63+ assert.NotNil(t, callbacks.Fill)
89- return player, nil64+ return player, nil
90- },65+ }}
91- }66+ stream, err := factory.NewStream(context.Background(), port.AudioStreamFormat{SampleRate: 48000, ChannelCount: 2, FramesPerBuffer: 512})
92-
93- format := port.AudioStreamFormat{
94- SampleRate: 48000,
95- ChannelCount: 2,
96- FramesPerBuffer: 512,
97- }
98-
99- stream, err := factory.NewStream(context.Background(), format)
100 require.NoError(t, err)67 require.NoError(t, err)
101- require.NotNil(t, stream)
102-
103- // Verify config mapping
104- assert.Equal(t, 48000, capturedConfig.SampleRate)
105- assert.Equal(t, 2, capturedConfig.Channels)
106- assert.Equal(t, 512, capturedConfig.FramesPerBuffer)
107- assert.Equal(t, pwpipewire.SampleFormatF32, capturedConfig.SampleFormat)
108- assert.Equal(t, pwpipewire.UnderrunFillSilence, capturedConfig.UnderrunPolicy)
109-
110- // Verify player was started
111 assert.True(t, player.started)68 assert.True(t, player.started)
112-
113- // Clean up
114 require.NoError(t, stream.Close())69 require.NoError(t, stream.Close())
115 }70 }
116 71
117-func TestFactory_NewStream_InvalidFormat_ReturnsError(t *testing.T) {72+func TestFactory_NewStreamRejectsInvalidFormat(t *testing.T) {
118- factory := &Factory{73+ factory := &Factory{createPlayer: func(pwpipewire.PlayerConfig, pwpipewire.PlayerCallbacks) (pwpipewire.Player, error) {
119- createPlayer: func(_ pwpipewire.PlayerConfig, _ pwpipewire.PlayerCallbacks) (pwpipewire.Player, error) {74+ t.Fatal("unexpected player")
120- t.Fatal("createPlayer should not be called for invalid format")75+ return nil, nil
121- return nil, nil76+ }}
122- },77+ for _, format := range []port.AudioStreamFormat{
123- }78+ {SampleRate: 0, ChannelCount: 2, FramesPerBuffer: 512},
124-79+ {SampleRate: 48000, ChannelCount: 0, FramesPerBuffer: 512},
125- tests := []struct {80+ {SampleRate: 48000, ChannelCount: 2, FramesPerBuffer: 0},
126- name string81+ } {
127- format port.AudioStreamFormat82+ stream, err := factory.NewStream(context.Background(), format)
128- }{83+ require.ErrorIs(t, err, ErrInvalidFormat)
129- {84+ assert.Nil(t, stream)
130- name: "zero sample rate",
131- format: port.AudioStreamFormat{SampleRate: 0, ChannelCount: 2, FramesPerBuffer: 512},
132- },
133- {
134- name: "zero channels",
135- format: port.AudioStreamFormat{SampleRate: 48000, ChannelCount: 0, FramesPerBuffer: 512},
136- },
137- {
138- name: "negative sample rate",
139- format: port.AudioStreamFormat{SampleRate: -48000, ChannelCount: 2, FramesPerBuffer: 512},
140- },
141- {
142- name: "negative channels",
143- format: port.AudioStreamFormat{SampleRate: 48000, ChannelCount: -2, FramesPerBuffer: 512},
144- },
145- {
146- name: "excessive sample rate",
147- format: port.AudioStreamFormat{SampleRate: 1000000, ChannelCount: 2, FramesPerBuffer: 512},
148- },
149- {
150- name: "excessive channels",
151- format: port.AudioStreamFormat{SampleRate: 48000, ChannelCount: 1025, FramesPerBuffer: 512},
152- },
153- {
154- name: "zero frames per buffer",
155- format: port.AudioStreamFormat{SampleRate: 48000, ChannelCount: 2, FramesPerBuffer: 0},
156- },
157- {
158- name: "excessive frames per buffer",
159- format: port.AudioStreamFormat{SampleRate: 48000, ChannelCount: 2, FramesPerBuffer: 10000},
160- },
161- }
162-
163- for _, tt := range tests {
164- t.Run(tt.name, func(t *testing.T) {
165- stream, err := factory.NewStream(context.Background(), tt.format)
166- require.Error(t, err)
167- assert.Nil(t, stream)
168- assert.ErrorIs(t, err, ErrInvalidFormat)
169- })
170- }
171-}
172-
173-func TestFactory_NewStream_PlayerCreationFails_ReturnsError(t *testing.T) {
174- expectedErr := errors.New("pipewire not available")
175- factory := &Factory{
176- createPlayer: func(_ pwpipewire.PlayerConfig, _ pwpipewire.PlayerCallbacks) (pwpipewire.Player, error) {
177- return nil, expectedErr
178- },
179 }85 }
180-
181- format := port.AudioStreamFormat{SampleRate: 48000, ChannelCount: 2, FramesPerBuffer: 512}
182- stream, err := factory.NewStream(context.Background(), format)
183-
184- require.Error(t, err)
185- assert.Nil(t, stream)
186- assert.ErrorIs(t, err, expectedErr)
187 }86 }
188 87
189-func TestFactory_NewStream_PlayerStartFails_ClosesAndReturnsError(t *testing.T) {88+func TestFactory_NewStreamClosesPlayerWhenStartFails(t *testing.T) {
190 player := newPlayerRecorder(t)89 player := newPlayerRecorder(t)
191 player.startErr = errors.New("start failed")90 player.startErr = errors.New("start failed")
192- factory := &Factory{91+ factory := &Factory{createPlayer: func(pwpipewire.PlayerConfig, pwpipewire.PlayerCallbacks) (pwpipewire.Player, error) {
193- createPlayer: func(_ pwpipewire.PlayerConfig, _ pwpipewire.PlayerCallbacks) (pwpipewire.Player, error) {92+ return player, nil
194- return player, nil93+ }}
195- },94+ stream, err := factory.NewStream(context.Background(), port.AudioStreamFormat{SampleRate: 48000, ChannelCount: 2, FramesPerBuffer: 512})
196- }
197-
198- format := port.AudioStreamFormat{SampleRate: 48000, ChannelCount: 2, FramesPerBuffer: 512}
199- stream, err := factory.NewStream(context.Background(), format)
200-
201- require.Error(t, err)
202- assert.Nil(t, stream)
203- assert.True(t, player.closedFlag, "player should be closed after start failure")
204-}
205-
206-func TestFactory_NewStream_ContextCancelled_ReturnsError(t *testing.T) {
207- factory := &Factory{
208- createPlayer: func(_ pwpipewire.PlayerConfig, _ pwpipewire.PlayerCallbacks) (pwpipewire.Player, error) {
209- t.Fatal("createPlayer should not be called when context is canceled")
210- return nil, nil
211- },
212- }
213-
214- ctx, cancel := context.WithCancel(context.Background())
215- cancel()
216-
217- format := port.AudioStreamFormat{SampleRate: 48000, ChannelCount: 2, FramesPerBuffer: 512}
218- stream, err := factory.NewStream(ctx, format)
219-
220 require.Error(t, err)95 require.Error(t, err)
221 assert.Nil(t, stream)96 assert.Nil(t, stream)
222-}
223-
224-func TestFactory_NewStream_MultipleFormats_Accepted(t *testing.T) {
225- formats := []port.AudioStreamFormat{
226- {SampleRate: 48000, ChannelCount: 2, FramesPerBuffer: 512},
227- {SampleRate: 44100, ChannelCount: 2, FramesPerBuffer: 512},
228- {SampleRate: 48000, ChannelCount: 1, FramesPerBuffer: 256},
229- {SampleRate: 96000, ChannelCount: 2, FramesPerBuffer: 1024},
230- {SampleRate: 48000, ChannelCount: 6, FramesPerBuffer: 512},
231- }
232-
233- for _, fmt := range formats {
234- t.Run("", func(t *testing.T) {
235- factory := &Factory{
236- createPlayer: func(config pwpipewire.PlayerConfig, _ pwpipewire.PlayerCallbacks) (pwpipewire.Player, error) {
237- assert.Equal(t, fmt.SampleRate, config.SampleRate)
238- assert.Equal(t, fmt.ChannelCount, config.Channels)
239- assert.Equal(t, fmt.FramesPerBuffer, config.FramesPerBuffer)
240- return newPlayerRecorder(t), nil
241- },
242- }
243-
244- stream, err := factory.NewStream(context.Background(), fmt)
245- require.NoError(t, err)
246- assert.NotNil(t, stream)
247-
248- _ = stream.Close()
249- })
250- }
251-}
252-
253-func TestFactory_ImplementsPortInterface(_ *testing.T) {
254- var _ port.AudioOutputFactory = (*Factory)(nil)
255-}
256-
257-// --- Stream adapter tests ---
258-
259-func TestPlayerStreamAdapter_Write_SendsToBuffer(t *testing.T) {
260- player := newPlayerRecorder(t)
261- adapter := &playerStreamAdapter{
262- player: player,
263- buf: make(chan [][]float32, 4),
264- closed: make(chan struct{}),
265- }
266-
267- samples := [][]float32{
268- {0.1, 0.2},
269- {0.3, 0.4},
270- }
271-
272- err := adapter.Write(samples)
273- require.NoError(t, err)
274-
275- // Verify data is in the channel
276- received := <-adapter.buf
277- assert.Equal(t, samples, received)
278-}
279-
280-func TestPlayerStreamAdapter_Write_AfterClose_ReturnsError(t *testing.T) {
281- player := newPlayerRecorder(t)
282- adapter := &playerStreamAdapter{
283- player: player,
284- buf: make(chan [][]float32, 4),
285- closed: make(chan struct{}),
286- }
287-
288- require.NoError(t, adapter.Close())
289-
290- err := adapter.Write([][]float32{{0.1}})
291- assert.ErrorIs(t, err, ErrStreamClosed)
292-}
293-
294-func TestPlayerStreamAdapter_Close_StopsAndClosesPlayer(t *testing.T) {
295- player := newPlayerRecorder(t)
296- adapter := &playerStreamAdapter{
297- player: player,
298- buf: make(chan [][]float32, 4),
299- closed: make(chan struct{}),
300- }
301-
302- err := adapter.Close()
303- require.NoError(t, err)
304-
305- assert.True(t, player.stopped)
306 assert.True(t, player.closedFlag)97 assert.True(t, player.closedFlag)
307 }98 }
308 99
309-func TestPlayerStreamAdapter_Close_IsIdempotent(t *testing.T) {100+func TestPlayerStreamAdapter_FillGetsOwnedPacketAfterCallbackMutation(t *testing.T) {
310- player := newPlayerRecorder(t)101+ adapter := testAdapter(3)
311- adapter := &playerStreamAdapter{102+ samples := [][]float32{{1, 2, 3}, {4, 5, 6}}
312- player: player,103+ require.NoError(t, adapter.Write(samples))
313- buf: make(chan [][]float32, 4),104+ samples[0][0], samples[1][0] = 99, 99
314- closed: make(chan struct{}),105+ pcm := stereoPCM(3)
315- }
316-
317- require.NoError(t, adapter.Close())
318- require.NoError(t, adapter.Close())
319-}
320-
321-func TestPlayerStreamAdapter_Close_PropagatesStopError(t *testing.T) {
322- player := newPlayerRecorder(t)
323- player.stopErr = errors.New("stop error")
324- adapter := &playerStreamAdapter{
325- player: player,
326- buf: make(chan [][]float32, 4),
327- closed: make(chan struct{}),
328- }
329-
330- err := adapter.Close()
331- require.Error(t, err)
332- assert.Contains(t, err.Error(), "stop error")
333-}
334-
335-func TestPlayerStreamAdapter_Close_PropagatesCloseError(t *testing.T) {
336- player := newPlayerRecorder(t)
337- player.closeErr = errors.New("close error")
338- adapter := &playerStreamAdapter{
339- player: player,
340- buf: make(chan [][]float32, 4),
341- closed: make(chan struct{}),
342- }
343-
344- err := adapter.Close()
345- require.Error(t, err)
346- assert.Contains(t, err.Error(), "close error")
347-}
348-
349-// --- Fill callback tests ---
350-
351-func TestFill_PullsFromBuffer(t *testing.T) {
352- adapter := &playerStreamAdapter{
353- buf: make(chan [][]float32, 4),
354- }
355-
356- // Push samples
357- samples := [][]float32{
358- {0.1, 0.2, 0.3},
359- {0.4, 0.5, 0.6},
360- }
361- adapter.buf <- samples
362-
363- // Simulate Fill call
364- pcm := &pwpipewire.PCMBuffer{
365- Frames: 3,
366- Channels: 2,
367- Stride: 1,
368- Samples: [][]float32{
369- make([]float32, 3),
370- make([]float32, 3),
371- },
372- }
373-
374 frames, err := adapter.fill(pcm)106 frames, err := adapter.fill(pcm)
375 require.NoError(t, err)107 require.NoError(t, err)
376 assert.Equal(t, 3, frames)108 assert.Equal(t, 3, frames)
377- assert.InEpsilon(t, 0.1, pcm.Samples[0][0], 0.0001)109+ assert.InEpsilon(t, 1, pcm.Samples[0][0], 0.0001)
378- assert.InEpsilon(t, 0.4, pcm.Samples[1][0], 0.0001)110+ assert.InEpsilon(t, 4, pcm.Samples[1][0], 0.0001)
379-}
380-
381-func TestFill_ReturnsZeroWhenEmpty(t *testing.T) {
382- adapter := &playerStreamAdapter{
383- buf: make(chan [][]float32, 4),
384- }
385-
386- pcm := &pwpipewire.PCMBuffer{
387- Frames: 256,
388- Channels: 2,
389- Stride: 1,
390- Samples: [][]float32{
391- make([]float32, 256),
392- make([]float32, 256),
393- },
394- }
395-
396- frames, err := adapter.fill(pcm)
397- require.NoError(t, err)
398- assert.Equal(t, 0, frames)
399-}
400-
401-func TestCopyToPCM_HandlesChannelMismatch(t *testing.T) {
402- // More source channels than PCM channels — truncates
403- pcm := &pwpipewire.PCMBuffer{
404- Frames: 2,
405- Channels: 1,
406- Samples: [][]float32{make([]float32, 2)},
407- }
408- samples := [][]float32{
409- {1.0, 2.0},
410- {3.0, 4.0}, // should be ignored
411- }
412- frames := copyToPCM(pcm, samples)
413- assert.Equal(t, 2, frames)
414- assert.InEpsilon(t, 1.0, pcm.Samples[0][0], 0.0001)
415 }111 }
416 112
417-func TestCopyToPCM_HandlesFrameMismatch(t *testing.T) {113+func TestPlayerStreamAdapter_SaturationDropsWithoutBlocking(t *testing.T) {
418- // Fewer source frames than PCM expects114+ adapter := testAdapter(2)
419- pcm := &pwpipewire.PCMBuffer{115+ for i := 0; i < packetQueueCapacity; i++ {
420- Frames: 4,116+ require.NoError(t, adapter.Write([][]float32{{float32(i)}, {float32(i)}}))
421- Channels: 1,
422- Samples: [][]float32{make([]float32, 4)},
423 }117 }
424- samples := [][]float32{118+ done := make(chan error, 1)
425- {1.0, 2.0}, // only 2 frames119+ go func() { done <- adapter.Write([][]float32{{99}, {99}}) }()
120+ select {
121+ case err := <-done:
122+ require.NoError(t, err)
123+ case <-time.After(time.Second):
124+ t.Fatal("Write blocked while the bounded queue was saturated")
426 }125 }
427- frames := copyToPCM(pcm, samples)126+ assert.Equal(t, uint64(1), adapter.dropCount.Load())
428- assert.Equal(t, 2, frames) // reports partial fill127+ assert.Len(t, adapter.ready, packetQueueCapacity)
429-}
430-
431-func TestCopyToPCM_NilPCM_ReturnsZero(t *testing.T) {
432- assert.Equal(t, 0, copyToPCM(nil, [][]float32{{1.0}}))
433 }128 }
434 129
435-func TestCopyToPCM_EmptySamples_ReturnsZero(t *testing.T) {130+func TestPlayerStreamAdapter_EmptyFillCountsUnderrun(t *testing.T) {
436- pcm := &pwpipewire.PCMBuffer{Frames: 1, Channels: 1, Samples: [][]float32{make([]float32, 1)}}131+ adapter := testAdapter(2)
437- assert.Equal(t, 0, copyToPCM(pcm, nil))132+ frames, err := adapter.fill(stereoPCM(2))
438- assert.Equal(t, 0, copyToPCM(pcm, [][]float32{}))133+ require.NoError(t, err)
134+ assert.Zero(t, frames)
135+ assert.Equal(t, uint64(1), adapter.underrunCount.Load())
439 }136 }
440 137
441-// --- Non-blocking Write tests ---138+func TestPlayerStreamAdapter_CopyToPCMHandlesMismatches(t *testing.T) {
442-139+ pcm := &pwpipewire.PCMBuffer{Frames: 4, Channels: 1, Samples: [][]float32{make([]float32, 4)}}
443-func TestPlayerStreamAdapter_Write_DropsWhenBufferFull(t *testing.T) {140+ assert.Equal(t, 2, copyToPCM(pcm, [][]float32{{1, 2}, {3, 4}}, 2))
444- adapter := &playerStreamAdapter{141+ assert.InEpsilon(t, 1, pcm.Samples[0][0], 0.0001)
445- player: newPlayerRecorder(t),142+ assert.Zero(t, copyToPCM(nil, [][]float32{{1}}, 1))
446- buf: make(chan [][]float32, 4),
447- closed: make(chan struct{}),
448- }
449-
450- // Fill the buffer to capacity
451- for i := range 4 {
452- require.NoError(t, adapter.Write([][]float32{{float32(i)}}))
453- }
454-
455- // The 5th write must NOT block — it should drop the packet and return nil.
456- err := adapter.Write([][]float32{{99.0}})
457- require.NoError(t, err, "Write must not return an error when dropping due to full buffer")
458-
459- // Verify the drop was counted
460- assert.Equal(t, uint64(1), adapter.dropCount.Load(), "one packet should have been dropped")
461-
462- // Verify buffer still contains the original 4 packets (not the dropped one)
463- assert.Len(t, adapter.buf, 4)
464 }143 }
465 144
466-func TestPlayerStreamAdapter_Write_DropsLogOnlyOnFirstAndPowerOfTwo(t *testing.T) {145+func TestPlayerStreamAdapter_CloseVsWrite(t *testing.T) {
467- adapter := &playerStreamAdapter{146+ adapter := testAdapter(2)
468- player: newPlayerRecorder(t),147+ player := newPlayerRecorder(t)
469- buf: make(chan [][]float32, 1), // tiny buffer to force drops148+ adapter.player = player
470- closed: make(chan struct{}),149+ require.NoError(t, adapter.Close())
471- }150+ require.ErrorIs(t, adapter.Write([][]float32{{1}, {2}}), ErrStreamClosed)
472-151+ assert.True(t, player.stopped)
473- // Fill buffer152+ assert.True(t, player.closedFlag)
474- require.NoError(t, adapter.Write([][]float32{{0.0}}))
475-
476- // Drop 10 packets
477- for i := range 10 {
478- _ = adapter.Write([][]float32{{float32(i)}})
479- }
480-
481- // Verify all drops were counted
482- assert.Equal(t, uint64(10), adapter.dropCount.Load())
483 }153 }
484 154
485-// --- Concurrent Write / Close tests ---
486-
487 func TestPlayerStreamAdapter_ConcurrentWriteClose_NoRace(t *testing.T) {155 func TestPlayerStreamAdapter_ConcurrentWriteClose_NoRace(t *testing.T) {
488- player := newPlayerRecorder(t)156+ adapter := testAdapter(4)
489- adapter := &playerStreamAdapter{157+ var writers sync.WaitGroup
490- player: player,158+ for range 4 {
491- buf: make(chan [][]float32, 4),159+ writers.Add(1)
492- closed: make(chan struct{}),160+ go func() {
493- }161+ defer writers.Done()
494-162+ for range 1000 {
495- done := make(chan struct{})163+ err := adapter.Write([][]float32{{1, 2}, {3, 4}})
496- go func() {164+ if err != nil {
497- defer close(done)165+ assert.ErrorIs(t, err, ErrStreamClosed)
498- for i := range 100 {166+ return
499- err := adapter.Write([][]float32{{float32(i)}})167+ }
500- if err != nil {
501- assert.ErrorIs(t, err, ErrStreamClosed)
502- return
503 }168 }
504- }169+ }()
505- }()170+ }
506-
507- // Let a few writes land, then close.
508 adapter.Close()171 adapter.Close()
509- <-done172+ writers.Wait()
510-}
511-
512-// --- NewFactory tests ---
513-
514-func TestNewFactory_Succeeds(t *testing.T) {
515- factory, err := NewFactory()
516- require.NoError(t, err)
517- require.NotNil(t, factory)
518- assert.NotNil(t, factory.createPlayer)
519 }173 }
520 174
521-// --- Integration-style test: full write-through ---
522-
523 func TestStream_WriteFlowsThroughToFill(t *testing.T) {175 func TestStream_WriteFlowsThroughToFill(t *testing.T) {
524- var capturedCallbacks pwpipewire.PlayerCallbacks176+ var callbacks pwpipewire.PlayerCallbacks
525 player := newPlayerRecorder(t)177 player := newPlayerRecorder(t)
526-178+ factory := &Factory{createPlayer: func(_ pwpipewire.PlayerConfig, got pwpipewire.PlayerCallbacks) (pwpipewire.Player, error) {
527- factory := &Factory{179+ callbacks = got
528- createPlayer: func(_ pwpipewire.PlayerConfig, callbacks pwpipewire.PlayerCallbacks) (pwpipewire.Player, error) {180+ return player, nil
529- capturedCallbacks = callbacks181+ }}
530- return player, nil182+ stream, err := factory.NewStream(context.Background(), port.AudioStreamFormat{SampleRate: 48000, ChannelCount: 2, FramesPerBuffer: 4})
531- },
532- }
533-
534- format := port.AudioStreamFormat{SampleRate: 48000, ChannelCount: 2, FramesPerBuffer: 4}
535- stream, err := factory.NewStream(context.Background(), format)
536 require.NoError(t, err)183 require.NoError(t, err)
537-184+ require.NoError(t, stream.Write([][]float32{{.1, .2, .3, .4}, {.5, .6, .7, .8}}))
538- // Write samples through the stream185+ pcm := stereoPCM(4)
539- samples := [][]float32{186+ frames, err := callbacks.Fill(pcm)
540- {0.1, 0.2, 0.3, 0.4},
541- {0.5, 0.6, 0.7, 0.8},
542- }
543- require.NoError(t, stream.Write(samples))
544-
545- // Simulate the Fill callback being called by the player
546- pcm := &pwpipewire.PCMBuffer{
547- Frames: 4,
548- Channels: 2,
549- Stride: 1,
550- Samples: [][]float32{
551- make([]float32, 4),
552- make([]float32, 4),
553- },
554- }
555-
556- frames, err := capturedCallbacks.Fill(pcm)
557 require.NoError(t, err)187 require.NoError(t, err)
558 assert.Equal(t, 4, frames)188 assert.Equal(t, 4, frames)
559- assert.InEpsilon(t, 0.1, pcm.Samples[0][0], 0.0001)189+ assert.InEpsilon(t, .1, pcm.Samples[0][0], .0001)
560- assert.InEpsilon(t, 0.5, pcm.Samples[1][0], 0.0001)190+ assert.InEpsilon(t, .5, pcm.Samples[1][0], .0001)
561-
562 require.NoError(t, stream.Close())191 require.NoError(t, stream.Close())
563 }192 }
193+
194+func BenchmarkPlayerStreamAdapterStereo(b *testing.B) {
195+ adapter := testAdapter(512)
196+ samples := [][]float32{make([]float32, 512), make([]float32, 512)}
197+ pcm := stereoPCM(512)
198+ b.ReportAllocs()
199+ b.SetBytes(2 * 512 * 4)
200+ for i := 0; i < b.N; i++ {
201+ if err := adapter.Write(samples); err != nil {
202+ b.Fatal(err)
203+ }
204+ if _, err := adapter.fill(pcm); err != nil {
205+ b.Fatal(err)
206+ }
207+ }
208+}
M internal/infrastructure/cef/webview.go +55 -15
@@ -250,16 +250,25 @@ type WebView struct {
250250 inputDiagDeltaXSum int64
251251 inputDiagDeltaYSum int64
252252
253- // Audio output factory and active stream.
254- audioOutputFactory port.AudioOutputFactory
255- audioStreamMu sync.Mutex
256- activeAudioStream port.AudioOutputStream
253+ // Audio output lifecycle. Packet callbacks load audioStream directly: its
254+ // immutable handle is atomically published and never requires a mutex.
255+ audioOutputFactory port.AudioOutputFactory
256+ audioStreamMu sync.Mutex
257+ audioStreamGeneration atomic.Uint64
258+ audioStream atomic.Pointer[audioStreamHandle]
257259
258260 // Audio instrumentation counters (diagnostic only).
259261 audioPacketCount atomic.Uint64 // total OnAudioStreamPacket calls
260262 audioWriteCount atomic.Uint64 // successful Write calls to stream
261263 }
262264
265+// audioStreamHandle is immutable after publication so packet callbacks can
266+// safely snapshot it without coordinating with lifecycle operations.
267+type audioStreamHandle struct {
268+ stream port.AudioOutputStream
269+ generation uint64
270+}
271+
263272 // pendingBrowserCreate holds the parameters needed to create a CEF browser,
264273 // deferred until the GL area has a non-zero size.
265274 type pendingBrowserCreate struct {
@@ -2331,20 +2340,51 @@ func (wv *WebView) takePendingCreate() *pendingBrowserCreate {
23312340 return pc
23322341 }
23332342
2334-// closeAudioStream closes and clears the active audio output stream.
2335-// This is safe to call even if no stream is active.
2336-func (wv *WebView) closeAudioStream() {
2343+// beginAudioStreamStart invalidates any prior stream before creating a new
2344+// one. A stream created for the returned generation may only be published if
2345+// that generation is still current.
2346+func (wv *WebView) beginAudioStreamStart() (uint64, *audioStreamHandle, bool) {
23372347 wv.audioStreamMu.Lock()
23382348 defer wv.audioStreamMu.Unlock()
2349+ if wv.destroyed.Load() {
2350+ return 0, nil, false
2351+ }
2352+ generation := wv.audioStreamGeneration.Add(1)
2353+ previous := wv.audioStream.Swap(nil)
2354+ wv.audioPlaying.Store(false)
2355+ return generation, previous, true
2356+}
23392357
2340- if wv.activeAudioStream != nil {
2341- if err := wv.activeAudioStream.Close(); err != nil {
2342- if wv.ctx != nil {
2343- logging.FromContext(wv.ctx).Debug().
2344- Err(err).
2345- Msg("cef: error closing audio stream")
2346- }
2358+// publishAudioStream publishes stream only if no Stop, Destroy, or newer Start
2359+// invalidated its creation generation. A rejected stream is owned by the
2360+// caller and must be closed by it.
2361+func (wv *WebView) publishAudioStream(generation uint64, stream port.AudioOutputStream) (*audioStreamHandle, bool) {
2362+ wv.audioStreamMu.Lock()
2363+ defer wv.audioStreamMu.Unlock()
2364+ if wv.destroyed.Load() || wv.audioStreamGeneration.Load() != generation {
2365+ return nil, false
2366+ }
2367+ previous := wv.audioStream.Swap(&audioStreamHandle{stream: stream, generation: generation})
2368+ wv.audioPlaying.Store(true)
2369+ return previous, true
2370+}
2371+
2372+// closeAudioStream invalidates creation in progress, atomically detaches the
2373+// published handle, then closes it outside the lifecycle mutex. The port
2374+// contract permits Close concurrently with a packet Write.
2375+func (wv *WebView) closeAudioStream() {
2376+ wv.audioStreamMu.Lock()
2377+ wv.audioStreamGeneration.Add(1)
2378+ stream := wv.audioStream.Swap(nil)
2379+ wv.audioPlaying.Store(false)
2380+ wv.audioStreamMu.Unlock()
2381+ wv.closeDetachedAudioStream(stream)
2382+}
2383+
2384+func (wv *WebView) closeDetachedAudioStream(handle *audioStreamHandle) {
2385+ if handle != nil && handle.stream != nil {
2386+ if err := handle.stream.Close(); err != nil && wv.ctx != nil {
2387+ logging.FromContext(wv.ctx).Debug().Err(err).Msg("cef: error closing audio stream")
23472388 }
2348- wv.activeAudioStream = nil
23492389 }
23502390 }
@@ -250,16 +250,25 @@ type WebView struct {
250 inputDiagDeltaXSum int64250 inputDiagDeltaXSum int64
251 inputDiagDeltaYSum int64251 inputDiagDeltaYSum int64
252 252
253- // Audio output factory and active stream.253+ // Audio output lifecycle. Packet callbacks load audioStream directly: its
254- audioOutputFactory port.AudioOutputFactory254+ // immutable handle is atomically published and never requires a mutex.
255- audioStreamMu sync.Mutex255+ audioOutputFactory port.AudioOutputFactory
256- activeAudioStream port.AudioOutputStream256+ audioStreamMu sync.Mutex
257+ audioStreamGeneration atomic.Uint64
258+ audioStream atomic.Pointer[audioStreamHandle]
257 259
258 // Audio instrumentation counters (diagnostic only).260 // Audio instrumentation counters (diagnostic only).
259 audioPacketCount atomic.Uint64 // total OnAudioStreamPacket calls261 audioPacketCount atomic.Uint64 // total OnAudioStreamPacket calls
260 audioWriteCount atomic.Uint64 // successful Write calls to stream262 audioWriteCount atomic.Uint64 // successful Write calls to stream
261 }263 }
262 264
265+// audioStreamHandle is immutable after publication so packet callbacks can
266+// safely snapshot it without coordinating with lifecycle operations.
267+type audioStreamHandle struct {
268+ stream port.AudioOutputStream
269+ generation uint64
270+}
271+
263 // pendingBrowserCreate holds the parameters needed to create a CEF browser,272 // pendingBrowserCreate holds the parameters needed to create a CEF browser,
264 // deferred until the GL area has a non-zero size.273 // deferred until the GL area has a non-zero size.
265 type pendingBrowserCreate struct {274 type pendingBrowserCreate struct {
@@ -2331,20 +2340,51 @@ func (wv *WebView) takePendingCreate() *pendingBrowserCreate {
2331 return pc2340 return pc
2332 }2341 }
2333 2342
2334-// closeAudioStream closes and clears the active audio output stream.2343+// beginAudioStreamStart invalidates any prior stream before creating a new
2335-// This is safe to call even if no stream is active.2344+// one. A stream created for the returned generation may only be published if
2336-func (wv *WebView) closeAudioStream() {2345+// that generation is still current.
2346+func (wv *WebView) beginAudioStreamStart() (uint64, *audioStreamHandle, bool) {
2337 wv.audioStreamMu.Lock()2347 wv.audioStreamMu.Lock()
2338 defer wv.audioStreamMu.Unlock()2348 defer wv.audioStreamMu.Unlock()
2349+ if wv.destroyed.Load() {
2350+ return 0, nil, false
2351+ }
2352+ generation := wv.audioStreamGeneration.Add(1)
2353+ previous := wv.audioStream.Swap(nil)
2354+ wv.audioPlaying.Store(false)
2355+ return generation, previous, true
2356+}
2339 2357
2340- if wv.activeAudioStream != nil {2358+// publishAudioStream publishes stream only if no Stop, Destroy, or newer Start
2341- if err := wv.activeAudioStream.Close(); err != nil {2359+// invalidated its creation generation. A rejected stream is owned by the
2342- if wv.ctx != nil {2360+// caller and must be closed by it.
2343- logging.FromContext(wv.ctx).Debug().2361+func (wv *WebView) publishAudioStream(generation uint64, stream port.AudioOutputStream) (*audioStreamHandle, bool) {
2344- Err(err).2362+ wv.audioStreamMu.Lock()
2345- Msg("cef: error closing audio stream")2363+ defer wv.audioStreamMu.Unlock()
2346- }2364+ if wv.destroyed.Load() || wv.audioStreamGeneration.Load() != generation {
2365+ return nil, false
2366+ }
2367+ previous := wv.audioStream.Swap(&audioStreamHandle{stream: stream, generation: generation})
2368+ wv.audioPlaying.Store(true)
2369+ return previous, true
2370+}
2371+
2372+// closeAudioStream invalidates creation in progress, atomically detaches the
2373+// published handle, then closes it outside the lifecycle mutex. The port
2374+// contract permits Close concurrently with a packet Write.
2375+func (wv *WebView) closeAudioStream() {
2376+ wv.audioStreamMu.Lock()
2377+ wv.audioStreamGeneration.Add(1)
2378+ stream := wv.audioStream.Swap(nil)
2379+ wv.audioPlaying.Store(false)
2380+ wv.audioStreamMu.Unlock()
2381+ wv.closeDetachedAudioStream(stream)
2382+}
2383+
2384+func (wv *WebView) closeDetachedAudioStream(handle *audioStreamHandle) {
2385+ if handle != nil && handle.stream != nil {
2386+ if err := handle.stream.Close(); err != nil && wv.ctx != nil {
2387+ logging.FromContext(wv.ctx).Debug().Err(err).Msg("cef: error closing audio stream")
2347 }2388 }
2348- wv.activeAudioStream = nil
2349 }2389 }
2350 }2390 }
M internal/infrastructure/cef/webview_audio_test.go +141 -58
@@ -4,6 +4,7 @@ import (
44 "bytes"
55 "context"
66 "errors"
7+ "sync/atomic"
78 "testing"
89 "time"
910
@@ -83,10 +84,10 @@ func TestWebView_AudioStreamLifecycleFieldsExist(t *testing.T) {
8384 }
8485
8586 // Verify the fields exist and can be set
86- wv.activeAudioStream = &mockAudioStream{}
87+ setAudioStream(wv, &mockAudioStream{})
8788
8889 assert.NotNil(t, wv.audioOutputFactory)
89- assert.NotNil(t, wv.activeAudioStream)
90+ assert.NotNil(t, activeAudioStream(wv))
9091 }
9192
9293 // mockAudioStream is a test double for port.AudioOutputStream
@@ -110,6 +111,18 @@ func (m *mockAudioStream) Close() error {
110111 // Verify mock implements the interface
111112 var _ port.AudioOutputStream = (*mockAudioStream)(nil)
112113
114+func setAudioStream(wv *WebView, stream port.AudioOutputStream) {
115+ wv.audioStream.Store(&audioStreamHandle{stream: stream, generation: wv.audioStreamGeneration.Load()})
116+}
117+
118+func activeAudioStream(wv *WebView) port.AudioOutputStream {
119+ handle := wv.audioStream.Load()
120+ if handle == nil {
121+ return nil
122+ }
123+ return handle.stream
124+}
125+
113126 // racyMockAudioStream blocks inside Write until doneCh is closed, allowing
114127 // tests to verify that the caller holds a lock across the Write call.
115128 type racyMockAudioStream struct {
@@ -131,6 +144,26 @@ func (m *racyMockAudioStream) Close() error {
131144
132145 var _ port.AudioOutputStream = (*racyMockAudioStream)(nil)
133146
147+type countedAudioStream struct{ closes atomic.Int32 }
148+
149+func (*countedAudioStream) Write([][]float32) error { return nil }
150+func (s *countedAudioStream) Close() error {
151+ s.closes.Add(1)
152+ return nil
153+}
154+
155+type controlledAudioFactory struct {
156+ entered chan struct{}
157+ release chan struct{}
158+ stream port.AudioOutputStream
159+}
160+
161+func (f *controlledAudioFactory) NewStream(context.Context, port.AudioStreamFormat) (port.AudioOutputStream, error) {
162+ close(f.entered)
163+ <-f.release
164+ return f.stream, nil
165+}
166+
134167 // ============================================================================
135168 // Task 6: Audio Stream Lifecycle Handling Tests
136169 // ============================================================================
@@ -159,8 +192,8 @@ func TestHandlerSet_OnAudioStreamStarted_CreatesStream(t *testing.T) {
159192 handlers.OnAudioStreamStarted(nil, params, 2)
160193
161194 // Assert
162- require.NotNil(t, wv.activeAudioStream)
163- assert.Equal(t, mockStream, wv.activeAudioStream)
195+ require.NotNil(t, activeAudioStream(wv))
196+ assert.Equal(t, mockStream, activeAudioStream(wv))
164197 assert.True(t, wv.audioPlaying.Load())
165198
166199 // Verify factory was called with correct format
@@ -205,8 +238,8 @@ func TestHandlerSet_OnAudioStreamStarted_ClosesExistingStream(t *testing.T) {
205238 wv := &WebView{
206239 ctx: ctx,
207240 audioOutputFactory: mockFactory,
208- activeAudioStream: oldStream,
209241 }
242+ setAudioStream(wv, oldStream)
210243
211244 handlers := &handlerSet{wv: wv}
212245
@@ -221,7 +254,7 @@ func TestHandlerSet_OnAudioStreamStarted_ClosesExistingStream(t *testing.T) {
221254
222255 // Assert
223256 assert.True(t, oldStream.closeCalled)
224- assert.Equal(t, newStream, wv.activeAudioStream)
257+ assert.Equal(t, newStream, activeAudioStream(wv))
225258 }
226259
227260 // TestHandlerSet_OnAudioStreamStarted_FactoryError_HandlesGracefully verifies
@@ -251,23 +284,23 @@ func TestHandlerSet_OnAudioStreamStarted_FactoryError_HandlesGracefully(t *testi
251284 // audioPlaying must be false when NewStream fails — if it stays true,
252285 // the UI/state layer incorrectly believes audio is active.
253286 assert.False(t, wv.audioPlaying.Load(), "audioPlaying must be false after NewStream failure")
254- assert.Nil(t, wv.activeAudioStream, "activeAudioStream must be nil after NewStream failure")
287+ assert.Nil(t, activeAudioStream(wv), "activeAudioStream must be nil after NewStream failure")
255288 }
256289
257290 // ============================================================================
258291 // Task 7: PCM Packet Forwarding Tests
259292 // ============================================================================
260293
261-// TestHandlerSet_OnAudioStreamPacket_CopiesAndWrites verifies that
262-// OnAudioStreamPacket copies samples before writing to avoid data races.
263-func TestHandlerSet_OnAudioStreamPacket_CopiesAndWrites(t *testing.T) {
294+// TestHandlerSet_OnAudioStreamPacket_ForwardsCallbackData verifies that CEF
295+// does not allocate a staging copy; the audio output port owns that handoff.
296+func TestHandlerSet_OnAudioStreamPacket_ForwardsCallbackData(t *testing.T) {
264297 ctx := context.Background()
265298 mockStream := &mockAudioStream{}
266299
267300 wv := &WebView{
268- ctx: ctx,
269- activeAudioStream: mockStream,
301+ ctx: ctx,
270302 }
303+ setAudioStream(wv, mockStream)
271304
272305 handlers := &handlerSet{wv: wv}
273306
@@ -287,20 +320,16 @@ func TestHandlerSet_OnAudioStreamPacket_CopiesAndWrites(t *testing.T) {
287320 require.Len(t, mockStream.writeSamples, 2)
288321 require.Len(t, mockStream.writeSamples[0], 512)
289322
290- // Verify data was copied (values match)
291323 assert.InDelta(t, 0.5, mockStream.writeSamples[0][0], 0.000001)
292324 assert.InDelta(t, -0.5, mockStream.writeSamples[1][0], 0.000001)
293-
294- // Verify it's a copy (different underlying array)
295- assert.NotSame(t, &originalData[0][0], &mockStream.writeSamples[0][0])
325+ assert.Same(t, &originalData[0][0], &mockStream.writeSamples[0][0])
296326 }
297327
298328 // TestHandlerSet_OnAudioStreamPacket_NoActiveStream_DoesNothing verifies
299329 // that packets are silently discarded when no stream is active.
300330 func TestHandlerSet_OnAudioStreamPacket_NoActiveStream_DoesNothing(t *testing.T) {
301331 wv := &WebView{
302- ctx: context.Background(),
303- activeAudioStream: nil,
332+ ctx: context.Background(),
304333 }
305334
306335 handlers := &handlerSet{wv: wv}
@@ -319,10 +348,8 @@ func TestHandlerSet_OnAudioStreamPacket_NoActiveStream_DoesNothing(t *testing.T)
319348 // TestHandlerSet_OnAudioStreamPacket_NilData_DoesNotPanic verifies
320349 // that nil data is handled gracefully.
321350 func TestHandlerSet_OnAudioStreamPacket_NilData_DoesNotPanic(t *testing.T) {
322- mockStream := &mockAudioStream{}
323351 wv := &WebView{
324- ctx: context.Background(),
325- activeAudioStream: mockStream,
352+ ctx: context.Background(),
326353 }
327354
328355 handlers := &handlerSet{wv: wv}
@@ -333,6 +360,19 @@ func TestHandlerSet_OnAudioStreamPacket_NilData_DoesNotPanic(t *testing.T) {
333360 })
334361 }
335362
363+func BenchmarkHandlerSet_OnAudioStreamPacketStereo(b *testing.B) {
364+ stream := &mockAudioStream{}
365+ wv := &WebView{ctx: context.Background()}
366+ setAudioStream(wv, stream)
367+ handlers := &handlerSet{wv: wv}
368+ data := [][]float32{make([]float32, 512), make([]float32, 512)}
369+ b.ReportAllocs()
370+ b.SetBytes(2 * 512 * 4)
371+ for i := 0; i < b.N; i++ {
372+ handlers.OnAudioStreamPacket(nil, data, 512, 0)
373+ }
374+}
375+
336376 // ============================================================================
337377 // Task 8: Stream Cleanup Tests
338378 // ============================================================================
@@ -342,9 +382,9 @@ func TestHandlerSet_OnAudioStreamPacket_NilData_DoesNotPanic(t *testing.T) {
342382 func TestHandlerSet_OnAudioStreamStopped_ClosesStream(t *testing.T) {
343383 mockStream := &mockAudioStream{}
344384 wv := &WebView{
345- ctx: context.Background(),
346- activeAudioStream: mockStream,
385+ ctx: context.Background(),
347386 }
387+ setAudioStream(wv, mockStream)
348388 wv.audioPlaying.Store(true)
349389
350390 handlers := &handlerSet{wv: wv}
@@ -354,7 +394,7 @@ func TestHandlerSet_OnAudioStreamStopped_ClosesStream(t *testing.T) {
354394
355395 // Assert
356396 assert.True(t, mockStream.closeCalled)
357- assert.Nil(t, wv.activeAudioStream)
397+ assert.Nil(t, activeAudioStream(wv))
358398 assert.False(t, wv.audioPlaying.Load())
359399 }
360400
@@ -362,8 +402,7 @@ func TestHandlerSet_OnAudioStreamStopped_ClosesStream(t *testing.T) {
362402 // that stopping when no stream exists doesn't panic.
363403 func TestHandlerSet_OnAudioStreamStopped_NoActiveStream_DoesNotPanic(t *testing.T) {
364404 wv := &WebView{
365- ctx: context.Background(),
366- activeAudioStream: nil,
405+ ctx: context.Background(),
367406 }
368407 wv.audioPlaying.Store(true)
369408
@@ -382,9 +421,9 @@ func TestHandlerSet_OnAudioStreamStopped_NoActiveStream_DoesNotPanic(t *testing.
382421 func TestHandlerSet_OnAudioStreamError_ClosesStream(t *testing.T) {
383422 mockStream := &mockAudioStream{}
384423 wv := &WebView{
385- ctx: context.Background(),
386- activeAudioStream: mockStream,
424+ ctx: context.Background(),
387425 }
426+ setAudioStream(wv, mockStream)
388427 wv.audioPlaying.Store(true)
389428
390429 handlers := &handlerSet{wv: wv}
@@ -394,7 +433,7 @@ func TestHandlerSet_OnAudioStreamError_ClosesStream(t *testing.T) {
394433
395434 // Assert
396435 assert.True(t, mockStream.closeCalled)
397- assert.Nil(t, wv.activeAudioStream)
436+ assert.Nil(t, activeAudioStream(wv))
398437 assert.False(t, wv.audioPlaying.Load())
399438 }
400439
@@ -402,8 +441,7 @@ func TestHandlerSet_OnAudioStreamError_ClosesStream(t *testing.T) {
402441 // that error handling when no stream exists doesn't panic.
403442 func TestHandlerSet_OnAudioStreamError_NoActiveStream_DoesNotPanic(t *testing.T) {
404443 wv := &WebView{
405- ctx: context.Background(),
406- activeAudioStream: nil,
444+ ctx: context.Background(),
407445 }
408446 wv.audioPlaying.Store(true)
409447
@@ -456,9 +494,9 @@ func TestHandlerSet_OnAudioStreamError_ShutdownSocketCloseLogsDebug(t *testing.T
456494 func TestWebView_Destroy_ClosesAudioStream(t *testing.T) {
457495 mockStream := &mockAudioStream{}
458496 wv := &WebView{
459- ctx: context.Background(),
460- activeAudioStream: mockStream,
497+ ctx: context.Background(),
461498 }
499+ setAudioStream(wv, mockStream)
462500 wv.audioPlaying.Store(true)
463501
464502 // Act — Destroy must close the audio stream as part of teardown.
@@ -466,7 +504,7 @@ func TestWebView_Destroy_ClosesAudioStream(t *testing.T) {
466504
467505 // Assert
468506 assert.True(t, mockStream.closeCalled, "Destroy must close the active audio stream")
469- assert.Nil(t, wv.activeAudioStream)
507+ assert.Nil(t, activeAudioStream(wv))
470508 }
471509
472510 // TestWebView_Destroy_NoAudioStream_DoesNotPanic verifies that Destroy
@@ -558,28 +596,81 @@ func TestOnAudioStreamStarted_CorrectParameterMapping(t *testing.T) {
558596 assert.Equal(t, 1024, mockFactory.lastFormat.FramesPerBuffer, "frames_per_buffer should be 1024 (from params.FramesPerBuffer)")
559597 }
560598
599+func TestHandlerSet_OnAudioStreamStarted_StopRejectsInFlightCreation(t *testing.T) {
600+ stream := &countedAudioStream{}
601+ factory := &controlledAudioFactory{entered: make(chan struct{}), release: make(chan struct{}), stream: stream}
602+ wv := &WebView{ctx: context.Background(), audioOutputFactory: factory}
603+ handlers := &handlerSet{wv: wv}
604+ params := &purecef.AudioParameters{SampleRate: 48000, FramesPerBuffer: 512}
605+ done := make(chan struct{})
606+ go func() { handlers.OnAudioStreamStarted(nil, params, 2); close(done) }()
607+ <-factory.entered
608+ handlers.OnAudioStreamStopped(nil)
609+ close(factory.release)
610+ <-done
611+
612+ assert.Nil(t, activeAudioStream(wv))
613+ assert.False(t, wv.audioPlaying.Load())
614+ assert.Equal(t, int32(1), stream.closes.Load())
615+}
616+
617+func TestHandlerSet_OnAudioStreamStarted_DestroyRejectsInFlightCreation(t *testing.T) {
618+ stream := &countedAudioStream{}
619+ factory := &controlledAudioFactory{entered: make(chan struct{}), release: make(chan struct{}), stream: stream}
620+ wv := &WebView{ctx: context.Background(), audioOutputFactory: factory}
621+ handlers := &handlerSet{wv: wv}
622+ params := &purecef.AudioParameters{SampleRate: 48000, FramesPerBuffer: 512}
623+ done := make(chan struct{})
624+ go func() { handlers.OnAudioStreamStarted(nil, params, 2); close(done) }()
625+ <-factory.entered
626+ wv.Destroy()
627+ close(factory.release)
628+ <-done
629+
630+ assert.Nil(t, activeAudioStream(wv))
631+ assert.True(t, wv.destroyed.Load())
632+ assert.Equal(t, int32(1), stream.closes.Load())
633+}
634+
635+func TestHandlerSet_OnAudioStreamStarted_ReplacesAndClosesEachStreamOnce(t *testing.T) {
636+ old, first, second := &countedAudioStream{}, &countedAudioStream{}, &countedAudioStream{}
637+ factory := &mockAudioFactory{stream: first}
638+ wv := &WebView{ctx: context.Background(), audioOutputFactory: factory}
639+ setAudioStream(wv, old)
640+ handlers := &handlerSet{wv: wv}
641+ params := &purecef.AudioParameters{SampleRate: 48000, FramesPerBuffer: 512}
642+
643+ handlers.OnAudioStreamStarted(nil, params, 2)
644+ factory.stream = second
645+ handlers.OnAudioStreamStarted(nil, params, 2)
646+ wv.closeAudioStream()
647+
648+ assert.Equal(t, int32(1), old.closes.Load())
649+ assert.Equal(t, int32(1), first.closes.Load())
650+ assert.Equal(t, int32(1), second.closes.Load())
651+ assert.Nil(t, activeAudioStream(wv))
652+}
653+
561654 // ============================================================================
562655 // Race-window tests (write/close interleave)
563656 // ============================================================================
564657
565-// TestOnAudioStreamPacket_WriteHoldsLock verifies that closeAudioStream cannot
566-// interleave between the stream snapshot and Write. If the lock is NOT held
567-// across Write, the goroutine calling closeAudioStream will Close the stream
568-// while Write is in-flight, producing a write-after-close.
569-func TestOnAudioStreamPacket_WriteHoldsLock(t *testing.T) {
658+// TestOnAudioStreamPacket_CloseDoesNotWaitForWrite verifies that lifecycle
659+// shutdown is not serialized behind a potentially slow callback write. The
660+// port contract makes Close/Write concurrency safe.
661+func TestOnAudioStreamPacket_CloseDoesNotWaitForWrite(t *testing.T) {
570662 ctx := context.Background()
571663
572- // racyStream blocks inside Write until we signal, giving closeAudioStream
573- // a window to race if the mutex isn't held.
664+ // racyStream blocks inside Write until we signal.
574665 stream := &racyMockAudioStream{
575666 writeCh: make(chan struct{}),
576667 doneCh: make(chan struct{}),
577668 }
578669
579670 wv := &WebView{
580- ctx: ctx,
581- activeAudioStream: stream,
671+ ctx: ctx,
582672 }
673+ setAudioStream(wv, stream)
583674 handlers := &handlerSet{wv: wv}
584675
585676 data := [][]float32{{0.1, 0.2}, {0.3, 0.4}}
@@ -590,32 +681,24 @@ func TestOnAudioStreamPacket_WriteHoldsLock(t *testing.T) {
590681 // Wait for Write to be entered.
591682 <-stream.writeCh
592683
593- // Now try to close. If the implementation holds the lock across Write,
594- // this will block until Write returns. If not, Close races with Write.
595684 closeDone := make(chan struct{})
596685 go func() {
597686 wv.closeAudioStream()
598687 close(closeDone)
599688 }()
600689
601- // Close should NOT have completed yet because Write holds the lock.
602- // Give the goroutine time to acquire the lock if it can.
603- time.Sleep(50 * time.Millisecond)
604690 select {
605691 case <-closeDone:
606- t.Fatal("closeAudioStream completed while Write was in-flight — lock not held across Write")
607- default:
608- // expected: closeAudioStream is blocked
692+ // expected: only the short stream snapshot takes the mutex
693+ case <-time.After(time.Second):
694+ t.Fatal("closeAudioStream waited for an in-flight Write")
609695 }
610696
611- // Unblock Write.
697+ // The write can finish after shutdown without holding the lifecycle mutex.
612698 close(stream.doneCh)
613699
614- // Now closeAudioStream should complete.
615- <-closeDone
616-
617700 assert.True(t, stream.closeCalled)
618- assert.Nil(t, wv.activeAudioStream)
701+ assert.Nil(t, activeAudioStream(wv))
619702 }
620703
621704 // TestOnAudioStreamPacket_StreamClosedBetweenPackets verifies that when a
@@ -626,9 +709,9 @@ func TestOnAudioStreamPacket_StreamClosedBetweenPackets(t *testing.T) {
626709 stream := &mockAudioStream{}
627710
628711 wv := &WebView{
629- ctx: ctx,
630- activeAudioStream: stream,
712+ ctx: ctx,
631713 }
714+ setAudioStream(wv, stream)
632715 handlers := &handlerSet{wv: wv}
633716 data := [][]float32{{0.1, 0.2}, {0.3, 0.4}}
634717
@@ -4,6 +4,7 @@ import (
4 "bytes"4 "bytes"
5 "context"5 "context"
6 "errors"6 "errors"
7+ "sync/atomic"
7 "testing"8 "testing"
8 "time"9 "time"
9 10
@@ -83,10 +84,10 @@ func TestWebView_AudioStreamLifecycleFieldsExist(t *testing.T) {
83 }84 }
84 85
85 // Verify the fields exist and can be set86 // Verify the fields exist and can be set
86- wv.activeAudioStream = &mockAudioStream{}87+ setAudioStream(wv, &mockAudioStream{})
87 88
88 assert.NotNil(t, wv.audioOutputFactory)89 assert.NotNil(t, wv.audioOutputFactory)
89- assert.NotNil(t, wv.activeAudioStream)90+ assert.NotNil(t, activeAudioStream(wv))
90 }91 }
91 92
92 // mockAudioStream is a test double for port.AudioOutputStream93 // mockAudioStream is a test double for port.AudioOutputStream
@@ -110,6 +111,18 @@ func (m *mockAudioStream) Close() error {
110 // Verify mock implements the interface111 // Verify mock implements the interface
111 var _ port.AudioOutputStream = (*mockAudioStream)(nil)112 var _ port.AudioOutputStream = (*mockAudioStream)(nil)
112 113
114+func setAudioStream(wv *WebView, stream port.AudioOutputStream) {
115+ wv.audioStream.Store(&audioStreamHandle{stream: stream, generation: wv.audioStreamGeneration.Load()})
116+}
117+
118+func activeAudioStream(wv *WebView) port.AudioOutputStream {
119+ handle := wv.audioStream.Load()
120+ if handle == nil {
121+ return nil
122+ }
123+ return handle.stream
124+}
125+
113 // racyMockAudioStream blocks inside Write until doneCh is closed, allowing126 // racyMockAudioStream blocks inside Write until doneCh is closed, allowing
114 // tests to verify that the caller holds a lock across the Write call.127 // tests to verify that the caller holds a lock across the Write call.
115 type racyMockAudioStream struct {128 type racyMockAudioStream struct {
@@ -131,6 +144,26 @@ func (m *racyMockAudioStream) Close() error {
131 144
132 var _ port.AudioOutputStream = (*racyMockAudioStream)(nil)145 var _ port.AudioOutputStream = (*racyMockAudioStream)(nil)
133 146
147+type countedAudioStream struct{ closes atomic.Int32 }
148+
149+func (*countedAudioStream) Write([][]float32) error { return nil }
150+func (s *countedAudioStream) Close() error {
151+ s.closes.Add(1)
152+ return nil
153+}
154+
155+type controlledAudioFactory struct {
156+ entered chan struct{}
157+ release chan struct{}
158+ stream port.AudioOutputStream
159+}
160+
161+func (f *controlledAudioFactory) NewStream(context.Context, port.AudioStreamFormat) (port.AudioOutputStream, error) {
162+ close(f.entered)
163+ <-f.release
164+ return f.stream, nil
165+}
166+
134 // ============================================================================167 // ============================================================================
135 // Task 6: Audio Stream Lifecycle Handling Tests168 // Task 6: Audio Stream Lifecycle Handling Tests
136 // ============================================================================169 // ============================================================================
@@ -159,8 +192,8 @@ func TestHandlerSet_OnAudioStreamStarted_CreatesStream(t *testing.T) {
159 handlers.OnAudioStreamStarted(nil, params, 2)192 handlers.OnAudioStreamStarted(nil, params, 2)
160 193
161 // Assert194 // Assert
162- require.NotNil(t, wv.activeAudioStream)195+ require.NotNil(t, activeAudioStream(wv))
163- assert.Equal(t, mockStream, wv.activeAudioStream)196+ assert.Equal(t, mockStream, activeAudioStream(wv))
164 assert.True(t, wv.audioPlaying.Load())197 assert.True(t, wv.audioPlaying.Load())
165 198
166 // Verify factory was called with correct format199 // Verify factory was called with correct format
@@ -205,8 +238,8 @@ func TestHandlerSet_OnAudioStreamStarted_ClosesExistingStream(t *testing.T) {
205 wv := &WebView{238 wv := &WebView{
206 ctx: ctx,239 ctx: ctx,
207 audioOutputFactory: mockFactory,240 audioOutputFactory: mockFactory,
208- activeAudioStream: oldStream,
209 }241 }
242+ setAudioStream(wv, oldStream)
210 243
211 handlers := &handlerSet{wv: wv}244 handlers := &handlerSet{wv: wv}
212 245
@@ -221,7 +254,7 @@ func TestHandlerSet_OnAudioStreamStarted_ClosesExistingStream(t *testing.T) {
221 254
222 // Assert255 // Assert
223 assert.True(t, oldStream.closeCalled)256 assert.True(t, oldStream.closeCalled)
224- assert.Equal(t, newStream, wv.activeAudioStream)257+ assert.Equal(t, newStream, activeAudioStream(wv))
225 }258 }
226 259
227 // TestHandlerSet_OnAudioStreamStarted_FactoryError_HandlesGracefully verifies260 // TestHandlerSet_OnAudioStreamStarted_FactoryError_HandlesGracefully verifies
@@ -251,23 +284,23 @@ func TestHandlerSet_OnAudioStreamStarted_FactoryError_HandlesGracefully(t *testi
251 // audioPlaying must be false when NewStream fails — if it stays true,284 // audioPlaying must be false when NewStream fails — if it stays true,
252 // the UI/state layer incorrectly believes audio is active.285 // the UI/state layer incorrectly believes audio is active.
253 assert.False(t, wv.audioPlaying.Load(), "audioPlaying must be false after NewStream failure")286 assert.False(t, wv.audioPlaying.Load(), "audioPlaying must be false after NewStream failure")
254- assert.Nil(t, wv.activeAudioStream, "activeAudioStream must be nil after NewStream failure")287+ assert.Nil(t, activeAudioStream(wv), "activeAudioStream must be nil after NewStream failure")
255 }288 }
256 289
257 // ============================================================================290 // ============================================================================
258 // Task 7: PCM Packet Forwarding Tests291 // Task 7: PCM Packet Forwarding Tests
259 // ============================================================================292 // ============================================================================
260 293
261-// TestHandlerSet_OnAudioStreamPacket_CopiesAndWrites verifies that294+// TestHandlerSet_OnAudioStreamPacket_ForwardsCallbackData verifies that CEF
262-// OnAudioStreamPacket copies samples before writing to avoid data races.295+// does not allocate a staging copy; the audio output port owns that handoff.
263-func TestHandlerSet_OnAudioStreamPacket_CopiesAndWrites(t *testing.T) {296+func TestHandlerSet_OnAudioStreamPacket_ForwardsCallbackData(t *testing.T) {
264 ctx := context.Background()297 ctx := context.Background()
265 mockStream := &mockAudioStream{}298 mockStream := &mockAudioStream{}
266 299
267 wv := &WebView{300 wv := &WebView{
268- ctx: ctx,301+ ctx: ctx,
269- activeAudioStream: mockStream,
270 }302 }
303+ setAudioStream(wv, mockStream)
271 304
272 handlers := &handlerSet{wv: wv}305 handlers := &handlerSet{wv: wv}
273 306
@@ -287,20 +320,16 @@ func TestHandlerSet_OnAudioStreamPacket_CopiesAndWrites(t *testing.T) {
287 require.Len(t, mockStream.writeSamples, 2)320 require.Len(t, mockStream.writeSamples, 2)
288 require.Len(t, mockStream.writeSamples[0], 512)321 require.Len(t, mockStream.writeSamples[0], 512)
289 322
290- // Verify data was copied (values match)
291 assert.InDelta(t, 0.5, mockStream.writeSamples[0][0], 0.000001)323 assert.InDelta(t, 0.5, mockStream.writeSamples[0][0], 0.000001)
292 assert.InDelta(t, -0.5, mockStream.writeSamples[1][0], 0.000001)324 assert.InDelta(t, -0.5, mockStream.writeSamples[1][0], 0.000001)
293-325+ assert.Same(t, &originalData[0][0], &mockStream.writeSamples[0][0])
294- // Verify it's a copy (different underlying array)
295- assert.NotSame(t, &originalData[0][0], &mockStream.writeSamples[0][0])
296 }326 }
297 327
298 // TestHandlerSet_OnAudioStreamPacket_NoActiveStream_DoesNothing verifies328 // TestHandlerSet_OnAudioStreamPacket_NoActiveStream_DoesNothing verifies
299 // that packets are silently discarded when no stream is active.329 // that packets are silently discarded when no stream is active.
300 func TestHandlerSet_OnAudioStreamPacket_NoActiveStream_DoesNothing(t *testing.T) {330 func TestHandlerSet_OnAudioStreamPacket_NoActiveStream_DoesNothing(t *testing.T) {
301 wv := &WebView{331 wv := &WebView{
302- ctx: context.Background(),332+ ctx: context.Background(),
303- activeAudioStream: nil,
304 }333 }
305 334
306 handlers := &handlerSet{wv: wv}335 handlers := &handlerSet{wv: wv}
@@ -319,10 +348,8 @@ func TestHandlerSet_OnAudioStreamPacket_NoActiveStream_DoesNothing(t *testing.T)
319 // TestHandlerSet_OnAudioStreamPacket_NilData_DoesNotPanic verifies348 // TestHandlerSet_OnAudioStreamPacket_NilData_DoesNotPanic verifies
320 // that nil data is handled gracefully.349 // that nil data is handled gracefully.
321 func TestHandlerSet_OnAudioStreamPacket_NilData_DoesNotPanic(t *testing.T) {350 func TestHandlerSet_OnAudioStreamPacket_NilData_DoesNotPanic(t *testing.T) {
322- mockStream := &mockAudioStream{}
323 wv := &WebView{351 wv := &WebView{
324- ctx: context.Background(),352+ ctx: context.Background(),
325- activeAudioStream: mockStream,
326 }353 }
327 354
328 handlers := &handlerSet{wv: wv}355 handlers := &handlerSet{wv: wv}
@@ -333,6 +360,19 @@ func TestHandlerSet_OnAudioStreamPacket_NilData_DoesNotPanic(t *testing.T) {
333 })360 })
334 }361 }
335 362
363+func BenchmarkHandlerSet_OnAudioStreamPacketStereo(b *testing.B) {
364+ stream := &mockAudioStream{}
365+ wv := &WebView{ctx: context.Background()}
366+ setAudioStream(wv, stream)
367+ handlers := &handlerSet{wv: wv}
368+ data := [][]float32{make([]float32, 512), make([]float32, 512)}
369+ b.ReportAllocs()
370+ b.SetBytes(2 * 512 * 4)
371+ for i := 0; i < b.N; i++ {
372+ handlers.OnAudioStreamPacket(nil, data, 512, 0)
373+ }
374+}
375+
336 // ============================================================================376 // ============================================================================
337 // Task 8: Stream Cleanup Tests377 // Task 8: Stream Cleanup Tests
338 // ============================================================================378 // ============================================================================
@@ -342,9 +382,9 @@ func TestHandlerSet_OnAudioStreamPacket_NilData_DoesNotPanic(t *testing.T) {
342 func TestHandlerSet_OnAudioStreamStopped_ClosesStream(t *testing.T) {382 func TestHandlerSet_OnAudioStreamStopped_ClosesStream(t *testing.T) {
343 mockStream := &mockAudioStream{}383 mockStream := &mockAudioStream{}
344 wv := &WebView{384 wv := &WebView{
345- ctx: context.Background(),385+ ctx: context.Background(),
346- activeAudioStream: mockStream,
347 }386 }
387+ setAudioStream(wv, mockStream)
348 wv.audioPlaying.Store(true)388 wv.audioPlaying.Store(true)
349 389
350 handlers := &handlerSet{wv: wv}390 handlers := &handlerSet{wv: wv}
@@ -354,7 +394,7 @@ func TestHandlerSet_OnAudioStreamStopped_ClosesStream(t *testing.T) {
354 394
355 // Assert395 // Assert
356 assert.True(t, mockStream.closeCalled)396 assert.True(t, mockStream.closeCalled)
357- assert.Nil(t, wv.activeAudioStream)397+ assert.Nil(t, activeAudioStream(wv))
358 assert.False(t, wv.audioPlaying.Load())398 assert.False(t, wv.audioPlaying.Load())
359 }399 }
360 400
@@ -362,8 +402,7 @@ func TestHandlerSet_OnAudioStreamStopped_ClosesStream(t *testing.T) {
362 // that stopping when no stream exists doesn't panic.402 // that stopping when no stream exists doesn't panic.
363 func TestHandlerSet_OnAudioStreamStopped_NoActiveStream_DoesNotPanic(t *testing.T) {403 func TestHandlerSet_OnAudioStreamStopped_NoActiveStream_DoesNotPanic(t *testing.T) {
364 wv := &WebView{404 wv := &WebView{
365- ctx: context.Background(),405+ ctx: context.Background(),
366- activeAudioStream: nil,
367 }406 }
368 wv.audioPlaying.Store(true)407 wv.audioPlaying.Store(true)
369 408
@@ -382,9 +421,9 @@ func TestHandlerSet_OnAudioStreamStopped_NoActiveStream_DoesNotPanic(t *testing.
382 func TestHandlerSet_OnAudioStreamError_ClosesStream(t *testing.T) {421 func TestHandlerSet_OnAudioStreamError_ClosesStream(t *testing.T) {
383 mockStream := &mockAudioStream{}422 mockStream := &mockAudioStream{}
384 wv := &WebView{423 wv := &WebView{
385- ctx: context.Background(),424+ ctx: context.Background(),
386- activeAudioStream: mockStream,
387 }425 }
426+ setAudioStream(wv, mockStream)
388 wv.audioPlaying.Store(true)427 wv.audioPlaying.Store(true)
389 428
390 handlers := &handlerSet{wv: wv}429 handlers := &handlerSet{wv: wv}
@@ -394,7 +433,7 @@ func TestHandlerSet_OnAudioStreamError_ClosesStream(t *testing.T) {
394 433
395 // Assert434 // Assert
396 assert.True(t, mockStream.closeCalled)435 assert.True(t, mockStream.closeCalled)
397- assert.Nil(t, wv.activeAudioStream)436+ assert.Nil(t, activeAudioStream(wv))
398 assert.False(t, wv.audioPlaying.Load())437 assert.False(t, wv.audioPlaying.Load())
399 }438 }
400 439
@@ -402,8 +441,7 @@ func TestHandlerSet_OnAudioStreamError_ClosesStream(t *testing.T) {
402 // that error handling when no stream exists doesn't panic.441 // that error handling when no stream exists doesn't panic.
403 func TestHandlerSet_OnAudioStreamError_NoActiveStream_DoesNotPanic(t *testing.T) {442 func TestHandlerSet_OnAudioStreamError_NoActiveStream_DoesNotPanic(t *testing.T) {
404 wv := &WebView{443 wv := &WebView{
405- ctx: context.Background(),444+ ctx: context.Background(),
406- activeAudioStream: nil,
407 }445 }
408 wv.audioPlaying.Store(true)446 wv.audioPlaying.Store(true)
409 447
@@ -456,9 +494,9 @@ func TestHandlerSet_OnAudioStreamError_ShutdownSocketCloseLogsDebug(t *testing.T
456 func TestWebView_Destroy_ClosesAudioStream(t *testing.T) {494 func TestWebView_Destroy_ClosesAudioStream(t *testing.T) {
457 mockStream := &mockAudioStream{}495 mockStream := &mockAudioStream{}
458 wv := &WebView{496 wv := &WebView{
459- ctx: context.Background(),497+ ctx: context.Background(),
460- activeAudioStream: mockStream,
461 }498 }
499+ setAudioStream(wv, mockStream)
462 wv.audioPlaying.Store(true)500 wv.audioPlaying.Store(true)
463 501
464 // Act — Destroy must close the audio stream as part of teardown.502 // Act — Destroy must close the audio stream as part of teardown.
@@ -466,7 +504,7 @@ func TestWebView_Destroy_ClosesAudioStream(t *testing.T) {
466 504
467 // Assert505 // Assert
468 assert.True(t, mockStream.closeCalled, "Destroy must close the active audio stream")506 assert.True(t, mockStream.closeCalled, "Destroy must close the active audio stream")
469- assert.Nil(t, wv.activeAudioStream)507+ assert.Nil(t, activeAudioStream(wv))
470 }508 }
471 509
472 // TestWebView_Destroy_NoAudioStream_DoesNotPanic verifies that Destroy510 // TestWebView_Destroy_NoAudioStream_DoesNotPanic verifies that Destroy
@@ -558,28 +596,81 @@ func TestOnAudioStreamStarted_CorrectParameterMapping(t *testing.T) {
558 assert.Equal(t, 1024, mockFactory.lastFormat.FramesPerBuffer, "frames_per_buffer should be 1024 (from params.FramesPerBuffer)")596 assert.Equal(t, 1024, mockFactory.lastFormat.FramesPerBuffer, "frames_per_buffer should be 1024 (from params.FramesPerBuffer)")
559 }597 }
560 598
599+func TestHandlerSet_OnAudioStreamStarted_StopRejectsInFlightCreation(t *testing.T) {
600+ stream := &countedAudioStream{}
601+ factory := &controlledAudioFactory{entered: make(chan struct{}), release: make(chan struct{}), stream: stream}
602+ wv := &WebView{ctx: context.Background(), audioOutputFactory: factory}
603+ handlers := &handlerSet{wv: wv}
604+ params := &purecef.AudioParameters{SampleRate: 48000, FramesPerBuffer: 512}
605+ done := make(chan struct{})
606+ go func() { handlers.OnAudioStreamStarted(nil, params, 2); close(done) }()
607+ <-factory.entered
608+ handlers.OnAudioStreamStopped(nil)
609+ close(factory.release)
610+ <-done
611+
612+ assert.Nil(t, activeAudioStream(wv))
613+ assert.False(t, wv.audioPlaying.Load())
614+ assert.Equal(t, int32(1), stream.closes.Load())
615+}
616+
617+func TestHandlerSet_OnAudioStreamStarted_DestroyRejectsInFlightCreation(t *testing.T) {
618+ stream := &countedAudioStream{}
619+ factory := &controlledAudioFactory{entered: make(chan struct{}), release: make(chan struct{}), stream: stream}
620+ wv := &WebView{ctx: context.Background(), audioOutputFactory: factory}
621+ handlers := &handlerSet{wv: wv}
622+ params := &purecef.AudioParameters{SampleRate: 48000, FramesPerBuffer: 512}
623+ done := make(chan struct{})
624+ go func() { handlers.OnAudioStreamStarted(nil, params, 2); close(done) }()
625+ <-factory.entered
626+ wv.Destroy()
627+ close(factory.release)
628+ <-done
629+
630+ assert.Nil(t, activeAudioStream(wv))
631+ assert.True(t, wv.destroyed.Load())
632+ assert.Equal(t, int32(1), stream.closes.Load())
633+}
634+
635+func TestHandlerSet_OnAudioStreamStarted_ReplacesAndClosesEachStreamOnce(t *testing.T) {
636+ old, first, second := &countedAudioStream{}, &countedAudioStream{}, &countedAudioStream{}
637+ factory := &mockAudioFactory{stream: first}
638+ wv := &WebView{ctx: context.Background(), audioOutputFactory: factory}
639+ setAudioStream(wv, old)
640+ handlers := &handlerSet{wv: wv}
641+ params := &purecef.AudioParameters{SampleRate: 48000, FramesPerBuffer: 512}
642+
643+ handlers.OnAudioStreamStarted(nil, params, 2)
644+ factory.stream = second
645+ handlers.OnAudioStreamStarted(nil, params, 2)
646+ wv.closeAudioStream()
647+
648+ assert.Equal(t, int32(1), old.closes.Load())
649+ assert.Equal(t, int32(1), first.closes.Load())
650+ assert.Equal(t, int32(1), second.closes.Load())
651+ assert.Nil(t, activeAudioStream(wv))
652+}
653+
561 // ============================================================================654 // ============================================================================
562 // Race-window tests (write/close interleave)655 // Race-window tests (write/close interleave)
563 // ============================================================================656 // ============================================================================
564 657
565-// TestOnAudioStreamPacket_WriteHoldsLock verifies that closeAudioStream cannot658+// TestOnAudioStreamPacket_CloseDoesNotWaitForWrite verifies that lifecycle
566-// interleave between the stream snapshot and Write. If the lock is NOT held659+// shutdown is not serialized behind a potentially slow callback write. The
567-// across Write, the goroutine calling closeAudioStream will Close the stream660+// port contract makes Close/Write concurrency safe.
568-// while Write is in-flight, producing a write-after-close.661+func TestOnAudioStreamPacket_CloseDoesNotWaitForWrite(t *testing.T) {
569-func TestOnAudioStreamPacket_WriteHoldsLock(t *testing.T) {
570 ctx := context.Background()662 ctx := context.Background()
571 663
572- // racyStream blocks inside Write until we signal, giving closeAudioStream664+ // racyStream blocks inside Write until we signal.
573- // a window to race if the mutex isn't held.
574 stream := &racyMockAudioStream{665 stream := &racyMockAudioStream{
575 writeCh: make(chan struct{}),666 writeCh: make(chan struct{}),
576 doneCh: make(chan struct{}),667 doneCh: make(chan struct{}),
577 }668 }
578 669
579 wv := &WebView{670 wv := &WebView{
580- ctx: ctx,671+ ctx: ctx,
581- activeAudioStream: stream,
582 }672 }
673+ setAudioStream(wv, stream)
583 handlers := &handlerSet{wv: wv}674 handlers := &handlerSet{wv: wv}
584 675
585 data := [][]float32{{0.1, 0.2}, {0.3, 0.4}}676 data := [][]float32{{0.1, 0.2}, {0.3, 0.4}}
@@ -590,32 +681,24 @@ func TestOnAudioStreamPacket_WriteHoldsLock(t *testing.T) {
590 // Wait for Write to be entered.681 // Wait for Write to be entered.
591 <-stream.writeCh682 <-stream.writeCh
592 683
593- // Now try to close. If the implementation holds the lock across Write,
594- // this will block until Write returns. If not, Close races with Write.
595 closeDone := make(chan struct{})684 closeDone := make(chan struct{})
596 go func() {685 go func() {
597 wv.closeAudioStream()686 wv.closeAudioStream()
598 close(closeDone)687 close(closeDone)
599 }()688 }()
600 689
601- // Close should NOT have completed yet because Write holds the lock.
602- // Give the goroutine time to acquire the lock if it can.
603- time.Sleep(50 * time.Millisecond)
604 select {690 select {
605 case <-closeDone:691 case <-closeDone:
606- t.Fatal("closeAudioStream completed while Write was in-flight — lock not held across Write")692+ // expected: only the short stream snapshot takes the mutex
607- default:693+ case <-time.After(time.Second):
608- // expected: closeAudioStream is blocked694+ t.Fatal("closeAudioStream waited for an in-flight Write")
609 }695 }
610 696
611- // Unblock Write.697+ // The write can finish after shutdown without holding the lifecycle mutex.
612 close(stream.doneCh)698 close(stream.doneCh)
613 699
614- // Now closeAudioStream should complete.
615- <-closeDone
616-
617 assert.True(t, stream.closeCalled)700 assert.True(t, stream.closeCalled)
618- assert.Nil(t, wv.activeAudioStream)701+ assert.Nil(t, activeAudioStream(wv))
619 }702 }
620 703
621 // TestOnAudioStreamPacket_StreamClosedBetweenPackets verifies that when a704 // TestOnAudioStreamPacket_StreamClosedBetweenPackets verifies that when a
@@ -626,9 +709,9 @@ func TestOnAudioStreamPacket_StreamClosedBetweenPackets(t *testing.T) {
626 stream := &mockAudioStream{}709 stream := &mockAudioStream{}
627 710
628 wv := &WebView{711 wv := &WebView{
629- ctx: ctx,712+ ctx: ctx,
630- activeAudioStream: stream,
631 }713 }
714+ setAudioStream(wv, stream)
632 handlers := &handlerSet{wv: wv}715 handlers := &handlerSet{wv: wv}
633 data := [][]float32{{0.1, 0.2}, {0.3, 0.4}}716 data := [][]float32{{0.1, 0.2}, {0.3, 0.4}}
634 717
M internal/infrastructure/cef/webview_handlers.go +27 -66
@@ -1134,48 +1134,44 @@ func (h *handlerSet) OnAudioStreamStarted(_ purecef.Browser, params *purecef.Aud
11341134 Msg("cef: OnAudioStreamStarted")
11351135 }
11361136
1137- // Reset packet counters for the new stream
1137+ // Starting a stream invalidates both a previous publication and any older
1138+ // creation already in flight. Close detached streams outside the mutex.
1139+ generation, previous, accepted := h.wv.beginAudioStreamStart()
1140+ if !accepted {
1141+ return
1142+ }
1143+ h.wv.closeDetachedAudioStream(previous)
11381144 h.wv.audioPacketCount.Store(0)
11391145 h.wv.audioWriteCount.Store(0)
11401146
1141- // Close any existing stream first
1142- h.wv.closeAudioStream()
1143-
1144- // Create new stream
11451147 ctx := h.wv.ctx
11461148 if ctx == nil {
11471149 ctx = context.Background()
11481150 }
1149-
11501151 stream, err := h.wv.audioOutputFactory.NewStream(ctx, format)
11511152 if err != nil {
1152- // Log error but don't panic - audio is non-critical.
1153- // Do NOT set audioPlaying — the stream was never created.
11541153 if h.wv.ctx != nil {
1155- logging.FromContext(h.wv.ctx).Warn().
1156- Err(err).
1157- Int("sample_rate", format.SampleRate).
1158- Int("channels", format.ChannelCount).
1154+ logging.FromContext(h.wv.ctx).Warn().Err(err).
1155+ Int("sample_rate", format.SampleRate).Int("channels", format.ChannelCount).
11591156 Msg("cef: failed to create audio output stream")
11601157 }
11611158 return
11621159 }
11631160
1164- // Set playing state only after stream creation succeeds
1165- h.wv.setAudioPlaying(true)
1166-
1161+ replaced, published := h.wv.publishAudioStream(generation, stream)
1162+ if !published {
1163+ // Stop, Destroy, or a newer Start won the race. This callback owns the
1164+ // unpublishable stream and closes it exactly once.
1165+ h.wv.closeDetachedAudioStream(&audioStreamHandle{stream: stream, generation: generation})
1166+ return
1167+ }
1168+ h.wv.closeDetachedAudioStream(replaced)
11671169 if h.wv.ctx != nil {
11681170 logging.FromContext(h.wv.ctx).Info().
1169- Int("sample_rate", format.SampleRate).
1170- Int("channels", format.ChannelCount).
1171+ Int("sample_rate", format.SampleRate).Int("channels", format.ChannelCount).
11711172 Int("frames_per_buffer", format.FramesPerBuffer).
11721173 Msg("cef: audio output stream created successfully")
11731174 }
1174-
1175- // Store the new stream
1176- h.wv.audioStreamMu.Lock()
1177- h.wv.activeAudioStream = stream
1178- h.wv.audioStreamMu.Unlock()
11791175 }
11801176
11811177 // buildAudioStreamFormat converts CEF AudioParameters to port.AudioStreamFormat.
@@ -1189,56 +1185,21 @@ func (h *handlerSet) buildAudioStreamFormat(params *purecef.AudioParameters, cha
11891185 }
11901186 }
11911187
1192-// OnAudioStreamPacket receives audio packets and forwards them to the output stream.
1193-// The mutex is held across both the stream snapshot and Write to prevent
1194-// closeAudioStream from closing the stream mid-write.
1195-func (h *handlerSet) OnAudioStreamPacket(_ purecef.Browser, data [][]float32, frames int32, pts int64) {
1188+// OnAudioStreamPacket is a real-time callback: it atomically snapshots an
1189+// immutable handle and performs only allocation-free atomic diagnostics. It
1190+// never takes a lifecycle mutex or emits synchronous logs.
1191+func (h *handlerSet) OnAudioStreamPacket(_ purecef.Browser, data [][]float32, frames int32, _ int64) {
11961192 if len(data) == 0 || frames <= 0 {
11971193 return
11981194 }
1199-
1200- // Copy the data before acquiring the stream lock because CEF can reuse the
1201- // buffer as soon as this callback returns.
1202- copiedData := make([][]float32, len(data))
1203- for i, channel := range data {
1204- if len(channel) < int(frames) {
1205- continue
1206- }
1207- copiedData[i] = make([]float32, frames)
1208- copy(copiedData[i], channel[:frames])
1209- }
1210-
1211- // Hold the lock across the stream read and Write so closeAudioStream
1212- // cannot close the stream between snapshot and write.
1213- h.wv.audioStreamMu.Lock()
1214- stream := h.wv.activeAudioStream
1215- if stream == nil {
1216- h.wv.audioStreamMu.Unlock()
1195+ handle := h.wv.audioStream.Load()
1196+ if handle == nil || handle.stream == nil {
12171197 return
12181198 }
1219-
1220- pktNum := h.wv.audioPacketCount.Add(1)
1221- if pktNum == 1 && h.wv.ctx != nil {
1222- logging.FromContext(h.wv.ctx).Info().
1223- Int("channels", len(data)).
1224- Int32("frames", frames).
1225- Int64("pts", pts).
1226- Msg("cef: first audio packet received")
1227- }
1228-
1229- err := stream.Write(copiedData)
1230- h.wv.audioStreamMu.Unlock()
1231-
1232- if err != nil {
1233- if h.wv.ctx != nil {
1234- logging.FromContext(h.wv.ctx).Debug().
1235- Err(err).
1236- Uint64("packets_received", pktNum).
1237- Msg("cef: audio stream write failed")
1238- }
1239- return
1199+ h.wv.audioPacketCount.Add(1)
1200+ if handle.stream.Write(data) == nil {
1201+ h.wv.audioWriteCount.Add(1)
12401202 }
1241- h.wv.audioWriteCount.Add(1)
12421203 }
12431204
12441205 // OnAudioStreamStopped handles stream stop.
@@ -1134,48 +1134,44 @@ func (h *handlerSet) OnAudioStreamStarted(_ purecef.Browser, params *purecef.Aud
1134 Msg("cef: OnAudioStreamStarted")1134 Msg("cef: OnAudioStreamStarted")
1135 }1135 }
1136 1136
1137- // Reset packet counters for the new stream1137+ // Starting a stream invalidates both a previous publication and any older
1138+ // creation already in flight. Close detached streams outside the mutex.
1139+ generation, previous, accepted := h.wv.beginAudioStreamStart()
1140+ if !accepted {
1141+ return
1142+ }
1143+ h.wv.closeDetachedAudioStream(previous)
1138 h.wv.audioPacketCount.Store(0)1144 h.wv.audioPacketCount.Store(0)
1139 h.wv.audioWriteCount.Store(0)1145 h.wv.audioWriteCount.Store(0)
1140 1146
1141- // Close any existing stream first
1142- h.wv.closeAudioStream()
1143-
1144- // Create new stream
1145 ctx := h.wv.ctx1147 ctx := h.wv.ctx
1146 if ctx == nil {1148 if ctx == nil {
1147 ctx = context.Background()1149 ctx = context.Background()
1148 }1150 }
1149-
1150 stream, err := h.wv.audioOutputFactory.NewStream(ctx, format)1151 stream, err := h.wv.audioOutputFactory.NewStream(ctx, format)
1151 if err != nil {1152 if err != nil {
1152- // Log error but don't panic - audio is non-critical.
1153- // Do NOT set audioPlaying — the stream was never created.
1154 if h.wv.ctx != nil {1153 if h.wv.ctx != nil {
1155- logging.FromContext(h.wv.ctx).Warn().1154+ logging.FromContext(h.wv.ctx).Warn().Err(err).
1156- Err(err).1155+ Int("sample_rate", format.SampleRate).Int("channels", format.ChannelCount).
1157- Int("sample_rate", format.SampleRate).
1158- Int("channels", format.ChannelCount).
1159 Msg("cef: failed to create audio output stream")1156 Msg("cef: failed to create audio output stream")
1160 }1157 }
1161 return1158 return
1162 }1159 }
1163 1160
1164- // Set playing state only after stream creation succeeds1161+ replaced, published := h.wv.publishAudioStream(generation, stream)
1165- h.wv.setAudioPlaying(true)1162+ if !published {
1166-1163+ // Stop, Destroy, or a newer Start won the race. This callback owns the
1164+ // unpublishable stream and closes it exactly once.
1165+ h.wv.closeDetachedAudioStream(&audioStreamHandle{stream: stream, generation: generation})
1166+ return
1167+ }
1168+ h.wv.closeDetachedAudioStream(replaced)
1167 if h.wv.ctx != nil {1169 if h.wv.ctx != nil {
1168 logging.FromContext(h.wv.ctx).Info().1170 logging.FromContext(h.wv.ctx).Info().
1169- Int("sample_rate", format.SampleRate).1171+ Int("sample_rate", format.SampleRate).Int("channels", format.ChannelCount).
1170- Int("channels", format.ChannelCount).
1171 Int("frames_per_buffer", format.FramesPerBuffer).1172 Int("frames_per_buffer", format.FramesPerBuffer).
1172 Msg("cef: audio output stream created successfully")1173 Msg("cef: audio output stream created successfully")
1173 }1174 }
1174-
1175- // Store the new stream
1176- h.wv.audioStreamMu.Lock()
1177- h.wv.activeAudioStream = stream
1178- h.wv.audioStreamMu.Unlock()
1179 }1175 }
1180 1176
1181 // buildAudioStreamFormat converts CEF AudioParameters to port.AudioStreamFormat.1177 // buildAudioStreamFormat converts CEF AudioParameters to port.AudioStreamFormat.
@@ -1189,56 +1185,21 @@ func (h *handlerSet) buildAudioStreamFormat(params *purecef.AudioParameters, cha
1189 }1185 }
1190 }1186 }
1191 1187
1192-// OnAudioStreamPacket receives audio packets and forwards them to the output stream.1188+// OnAudioStreamPacket is a real-time callback: it atomically snapshots an
1193-// The mutex is held across both the stream snapshot and Write to prevent1189+// immutable handle and performs only allocation-free atomic diagnostics. It
1194-// closeAudioStream from closing the stream mid-write.1190+// never takes a lifecycle mutex or emits synchronous logs.
1195-func (h *handlerSet) OnAudioStreamPacket(_ purecef.Browser, data [][]float32, frames int32, pts int64) {1191+func (h *handlerSet) OnAudioStreamPacket(_ purecef.Browser, data [][]float32, frames int32, _ int64) {
1196 if len(data) == 0 || frames <= 0 {1192 if len(data) == 0 || frames <= 0 {
1197 return1193 return
1198 }1194 }
1199-1195+ handle := h.wv.audioStream.Load()
1200- // Copy the data before acquiring the stream lock because CEF can reuse the1196+ if handle == nil || handle.stream == nil {
1201- // buffer as soon as this callback returns.
1202- copiedData := make([][]float32, len(data))
1203- for i, channel := range data {
1204- if len(channel) < int(frames) {
1205- continue
1206- }
1207- copiedData[i] = make([]float32, frames)
1208- copy(copiedData[i], channel[:frames])
1209- }
1210-
1211- // Hold the lock across the stream read and Write so closeAudioStream
1212- // cannot close the stream between snapshot and write.
1213- h.wv.audioStreamMu.Lock()
1214- stream := h.wv.activeAudioStream
1215- if stream == nil {
1216- h.wv.audioStreamMu.Unlock()
1217 return1197 return
1218 }1198 }
1219-1199+ h.wv.audioPacketCount.Add(1)
1220- pktNum := h.wv.audioPacketCount.Add(1)1200+ if handle.stream.Write(data) == nil {
1221- if pktNum == 1 && h.wv.ctx != nil {1201+ h.wv.audioWriteCount.Add(1)
1222- logging.FromContext(h.wv.ctx).Info().
1223- Int("channels", len(data)).
1224- Int32("frames", frames).
1225- Int64("pts", pts).
1226- Msg("cef: first audio packet received")
1227- }
1228-
1229- err := stream.Write(copiedData)
1230- h.wv.audioStreamMu.Unlock()
1231-
1232- if err != nil {
1233- if h.wv.ctx != nil {
1234- logging.FromContext(h.wv.ctx).Debug().
1235- Err(err).
1236- Uint64("packets_received", pktNum).
1237- Msg("cef: audio stream write failed")
1238- }
1239- return
1240 }1202 }
1241- h.wv.audioWriteCount.Add(1)
1242 }1203 }
1243 1204
1244 // OnAudioStreamStopped handles stream stop.1205 // OnAudioStreamStopped handles stream stop.

Keyboard shortcuts

Single-key shortcuts are on. ctrl k and esc work whatever this says.

ctrl k :
go to a section, branch, tag, file or repository
?
this list
j k
move the row cursor down / up
enter
open the row under the cursor; on Files, a directory opens or closes in place
↑ ↓ ← → home end
in the file tree (focus on a row): previous / next row, close / open a directory or go to its parent / first entry, first / last row. They never open a file
l h
on Files: open a directory or a file / close a directory or go up to its parent, then to the parent page. Elsewhere: into the side panel (on a merge request, its files) / back to the content; there j k move between its controls
g g G
first / last row
/
filter the list; on Files, find a file anywhere in the repository (esc clears, enter opens the best match)
b
switch branch or tag
g s g f g c g b g m g i g r g a
tabs: summary, files, commits, branches & tags, merge requests, issues, releases, actions (when available)
g h
home
w
wide mode: use the whole width of the screen, or not (remembered)
y
copy a permalink pinned to the commit
d u x
diff: split / unified / collapse or expand all
e E
diff: the whole file in view / every file, with the lines the hunks hide
n
merge request: next note in the diff
esc
close a dialog, leave a field

Branches 8

Tags 42

↑↓ moveenter openesc close