From 74416f1fbafdef4ab383c5f9e4f888cee2066354 Mon Sep 17 00:00:00 2001 From: Paul Robertson Date: Sun, 25 Oct 2026 20:08:09 +0000 Subject: [PATCH 1/2] fix(audio): keep stream open across pause so DACs hold lock Pause arrives as stream/end, which ran close_stream() and fully closed the PortAudio stream; resume then recreated it via set_format(). Each close/open drops USB lock, clicking muting relays on XMOS-based DACs (verified live: the click survives full digital silence, so no server-side mute or fade can suppress it). clear() and close_stream() now drain the queue but leave the stream running - the callback already renders silence while starved - and set_format() skips recreating the stream when the format is unchanged. The device releases on stop() (shutdown) and on real format changes only. --- sendspin/audio.py | 31 +++++++++------- tests/test_audio.py | 88 +++++++++++++++++++++++++++++++++++++++++++++ 2 files changed, 106 insertions(+), 13 deletions(-) diff --git a/sendspin/audio.py b/sendspin/audio.py index 5bf1a63..099dc28 100644 --- a/sendspin/audio.py +++ b/sendspin/audio.py @@ -253,6 +253,14 @@ def set_format(self, audio_format: AudioFormat, device: AudioDevice) -> None: device: Audio device to use. """ pcm_format = audio_format.pcm_format + if self._stream is not None and self._format == pcm_format: + logger.debug( + "Audio format unchanged (%dHz/%d-bit/%dch); keeping stream open", + pcm_format.sample_rate, + pcm_format.bit_depth, + pcm_format.channels, + ) + return self._format = pcm_format self._close_stream() @@ -340,17 +348,16 @@ def stop(self) -> None: self._stream_executor.shutdown(wait=True) def close_stream(self) -> None: - """Drop queued audio and fully close the stream to release the device. + """Drop queued audio but keep the stream open on silence. - Unlike clear(), which only stops the stream (leaving the device FD open), - this fully closes the PortAudio stream. Call when the server signals - end-of-stream; the stream will be recreated by set_format() when the - next track begins. + The callback renders silence while starved, so the audio device + stays open and DACs hold lock across pause/resume and across + stream end/start pairs. The device is released by stop() + (shutdown) and by set_format() when the format actually changes. """ if self._closed: return self.clear() - self._close_stream() def clear(self) -> None: """Drop all queued audio chunks.""" @@ -359,13 +366,11 @@ def clear(self) -> None: # Clear deferred operation flag self._clear_requested = False - # Stop the audio stream (but don't close it) to release ALSA device - # This allows the device to transition to 'closed' state when paused. - # The stream restarts when new chunks arrive in submit(). - self._stream_started = False - stream = self._stream - if stream is not None: - self._stream_executor.submit(self._call_stream, stream.stop) + # Keep the audio stream running: with the queue drained the callback + # renders silence, so the device stays open and DACs that click a + # muting relay on lock loss (e.g. XMOS USB) stay silent across + # pause/resume/seek. Chunks submitted after the clear play on the + # still-running stream; submit() restarts it only if never started. # Drain all queued chunks while True: diff --git a/tests/test_audio.py b/tests/test_audio.py index 3290cb5..2d871a3 100644 --- a/tests/test_audio.py +++ b/tests/test_audio.py @@ -61,3 +61,91 @@ def now() -> int: assert player._drop_every_n_frames == 0 # noqa: SLF001 assert player._insert_every_n_frames == 0 # noqa: SLF001 + + +class _RecordingStream: + def __init__(self) -> None: + self.stops = 0 + + def stop(self) -> None: + self.stops += 1 + + +def test_clear_keeps_stream_open_so_device_never_closes() -> None: + """clear() must drain the queue without stopping the stream. + + The callback renders silence while starved, so keeping the stream + running holds the audio device open and DACs with lock-loss muting + relays (e.g. XMOS USB) stay silent across pause/resume/seek. + """ + player = AudioPlayer(lambda ts: ts, lambda ts: ts, now_us=lambda: 0) + stream = _RecordingStream() + player._stream = stream # noqa: SLF001 + player._stream_started = True # noqa: SLF001 + player._queue.put( # noqa: SLF001 + _QueuedChunk(server_timestamp_us=0, audio_data=b"\x01\x00\x02\x00") + ) + + player.clear() + + assert player._queue.empty() # noqa: SLF001 + assert player._stream_started # noqa: SLF001 + assert stream.stops == 0 + + +class _RecordingStreamFull: + def __init__(self) -> None: + self.stops = 0 + self.starts = 0 + self.closes = 0 + + def stop(self) -> None: + self.stops += 1 + + def start(self) -> None: + self.starts += 1 + + def close(self) -> None: + self.closes += 1 + + +def _make_player() -> AudioPlayer: + return AudioPlayer(lambda ts: ts, lambda ts: ts, now_us=lambda: 0) + + +def test_close_stream_keeps_device_open_on_silence() -> None: + """End-of-stream must not release the audio device. + + The server sends stream/end on pause, so fully closing here is what + drops DAC lock and clicks the relay. Silence keeps it held. + """ + player = _make_player() + stream = _RecordingStreamFull() + player._stream = stream # noqa: SLF001 + player._stream_started = True # noqa: SLF001 + player._queue.put( # noqa: SLF001 + _QueuedChunk(server_timestamp_us=0, audio_data=b"\x01\x00\x02\x00") + ) + + player.close_stream() + + assert player._queue.empty() # noqa: SLF001 + assert player._stream_started # noqa: SLF001 + assert player._stream is stream # noqa: SLF001 + assert (stream.stops, stream.closes) == (0, 0) + + +def test_set_format_same_format_keeps_existing_stream() -> None: + """Resume after pause re-sends the identical format: no recreate.""" + from aiosendspin.client import PCMFormat + + player = _make_player() + stream = _RecordingStreamFull() + fmt = PCMFormat(sample_rate=48000, channels=2, bit_depth=16) + player._format = fmt # noqa: SLF001 + player._stream = stream # noqa: SLF001 + + player.set_format(SimpleNamespace(pcm_format=fmt), device=None) + + assert player._stream is stream # noqa: SLF001 + assert (stream.stops, stream.closes) == (0, 0) From e0e1e9000a7d104bbe95f9f9b08cc8f3da79c690 Mon Sep 17 00:00:00 2001 From: Paul Robertson Date: Sun, 25 Oct 2026 21:33:24 +0000 Subject: [PATCH 2/2] fix(audio): harden callback against stale-format races Resume-after-announce crashed the audio callback: set_format() swapped the shared format synchronously while the old stream closed async, so an in-flight old-stream callback sized its buffer from the new format (ValueError: memoryview assignment), leaving the client silent while the server showed playing. Seen live on a moOde Pi. Three guards: set_format() stops the old stream synchronously before swapping (stop blocks until in-flight callbacks finish; worker thread, never in-callback, and _call_stream logs internally) and drops the stale last-frame cache; the callback clamps bytes_needed and read frames to the actual buffer so any residual mismatch degrades to trailing silence; _fill_silence trims to the buffer. Release paths unchanged (stop(), real format changes). --- sendspin/audio.py | 52 ++++++++++++++++++++++++++++--- tests/test_audio.py | 74 +++++++++++++++++++++++++++++++++++++++++++++ 2 files changed, 122 insertions(+), 4 deletions(-) diff --git a/sendspin/audio.py b/sendspin/audio.py index 099dc28..f54947f 100644 --- a/sendspin/audio.py +++ b/sendspin/audio.py @@ -261,6 +261,18 @@ def set_format(self, audio_format: AudioFormat, device: AudioDevice) -> None: pcm_format.channels, ) return + # Stop the old stream synchronously BEFORE swapping the format. The + # close below is async, and an in-flight old-stream callback computing + # against the new format sizes its buffer wrong and crashes + # (ValueError: memoryview assignment, seen live on resume-after- + # announce). stop() blocks until in-flight callbacks finish, and this + # runs on the worker thread, never in a callback, so no deadlock. + # _call_stream logs internally. Also drop the last-frame cache: it + # carries the old frame size into the slow correction path. + old_stream = self._stream + if old_stream is not None: + self._call_stream(old_stream.stop) + self._last_output_frame = b"" self._format = pcm_format self._close_stream() @@ -433,8 +445,14 @@ def _audio_callback( # noqa: PLR0915 assert self._format is not None - bytes_needed = frames * self._format.frame_size output_buffer = memoryview(outdata).cast("B") + # Trust the buffer, not the parameters: a stale callback racing a + # format switch can arrive with frames sized for another format. + # Every slice below derives from bytes_needed, so bounding it here + # bounds them all (bulk reads are capped separately to keep cursor + # accounting consistent). + got = frames * self._format.frame_size + bytes_needed = got if got < len(output_buffer) else len(output_buffer) if status: # Detect underflow and request re-anchor (processed by main thread) @@ -485,6 +503,22 @@ def _audio_callback( # noqa: PLR0915 else: frame_size = self._format.frame_size + # Hard guarantee: never read more frames than fit the buffer. + # A stale callback racing a format switch would otherwise size + # its read from the new format against the old buffer and crash + # (ValueError: memoryview assignment, seen live). Capped reads + # degrade to a short trailing silence fill instead. + max_frames = ( + (len(output_buffer) - bytes_written) // frame_size if frame_size > 0 else 0 + ) + if frames > max_frames: + logger.debug( + "Capping stale callback read: %d frames to %d", + frames, + max(0, max_frames), + ) + frames = max(0, max_frames) + # Thread-safe snapshot of correction schedule (prevent mid-callback changes) insert_every_n = self._insert_every_n_frames drop_every_n = self._drop_every_n_frames @@ -572,6 +606,12 @@ def _audio_callback( # noqa: PLR0915 self._frames_until_next_insert = insert_counter self._frames_until_next_drop = drop_counter + # Trailing guard: never hand PortAudio uninitialized bytes. A + # capped stale read above can come up short; pad the tail. + if bytes_written < bytes_needed: + self._fill_silence(output_buffer, bytes_written, bytes_needed - bytes_written) + bytes_written = bytes_needed + except Exception: logger.exception("Error in audio callback") # Fill rest with silence on error @@ -912,9 +952,13 @@ def _smooth_sync_error(self, error_us: int) -> None: self._sync_error_filtered_us = self._sync_error_filter.offset def _fill_silence(self, output_buffer: memoryview, offset: int, num_bytes: int) -> None: - """Fill output buffer range with silence.""" - if num_bytes > 0: - output_buffer[offset : offset + num_bytes] = b"\x00" * num_bytes + """Fill output buffer range with silence, trimmed to the buffer.""" + end = offset + num_bytes + buf = len(output_buffer) + if end > buf: + end = buf + if end > offset: + output_buffer[offset:end] = b"\x00" * (end - offset) def _apply_volume(self, output_buffer: memoryview) -> None: """ diff --git a/tests/test_audio.py b/tests/test_audio.py index 2d871a3..f300623 100644 --- a/tests/test_audio.py +++ b/tests/test_audio.py @@ -149,3 +149,77 @@ def test_set_format_same_format_keeps_existing_stream() -> None: assert player._stream is stream # noqa: SLF001 assert (stream.stops, stream.closes) == (0, 0) + + +def test_set_format_change_stops_old_stream_synchronously() -> None: + """A format switch must stop the old stream before swapping formats. + + Otherwise an in-flight old-stream callback computes against the new + format and crashes (ValueError: memoryview assignment, seen live on + resume-after-announce). stop() is safe here: set_format runs on the + worker thread, never inside a callback. + """ + import concurrent.futures + from unittest.mock import patch + + from aiosendspin.client import PCMFormat + + player = _make_player() + old = _RecordingStreamFull() + player._stream = old # noqa: SLF001 + player._format = PCMFormat(sample_rate=48000, channels=2, bit_depth=16) # noqa: SLF001 + player._stream_executor = concurrent.futures.ThreadPoolExecutor(max_workers=1) # noqa: SLF001 + new = _RecordingStreamFull() + new.latency = 0.05 + try: + with patch("sendspin.audio.sounddevice.RawOutputStream", return_value=new): + player.set_format( + SimpleNamespace( + pcm_format=PCMFormat(sample_rate=44100, channels=2, bit_depth=24), + codec=SimpleNamespace(value="flac"), + ), + device=SimpleNamespace(device_id=6), + ) + finally: + # Drain the async close: deterministic counts below. + player._stream_executor.shutdown(wait=True) # noqa: SLF001 + # One synchronous stop (before the swap) + one async stop/close pair. + assert old.stops == 2 + assert old.closes == 1 + assert player._format.sample_rate == 44100 # noqa: SLF001 + assert player._stream is new # noqa: SLF001 + assert player._last_output_frame == b"" # noqa: SLF001 + + +def test_stale_callback_capped_to_buffer_instead_of_crash() -> None: + """Replay the live crash: old-stream buffer, new-format frame size. + + A 2048-frame callback on a buffer sized for 2-byte frames while the + format claims 6-byte frames must degrade to silence, never raise. + """ + now_us = 0 + + def now() -> int: + return now_us + + player = AudioPlayer(lambda ts: ts, lambda ts: ts, now_us=now) + player._format = SimpleNamespace( # noqa: SLF001 + sample_rate=48000, + channels=2, + bit_depth=24, + frame_size=6, + ) + player._playback_state = PlaybackState.PLAYING # noqa: SLF001 + player._queue.put( # noqa: SLF001 + _QueuedChunk(server_timestamp_us=0, audio_data=b"\x01" * 600) + ) + + out = bytearray(2048 * 2) # old stream buffer: 2-byte frames + player._audio_callback( # noqa: SLF001 + memoryview(out), + frames=2048, + time=SimpleNamespace(outputBufferDacTime=0.0), + status=_NoCallbackStatus(), + ) + + assert bytes(out[-2:]) == b"\x00\x00"