From 67aad047e5df2e7d9cb9b13b83b2485dd14af074 Mon Sep 17 00:00:00 2001 From: Maxim Raznatovski Date: Tue, 27 Oct 2026 15:52:38 +0100 Subject: [PATCH] fix(serve): stop multi-worker serve from skipping audio MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit With --workers, the workers regularly dropped a whole 4096-sample block, leaving an 85 ms gap. Each block's timestamp was truncated to whole microseconds, so the timeline fell 1 µs behind every three blocks and the workers dropped the next block as already played. --- sendspin/serve/coordinator.py | 10 ++++++---- tests/serve/test_coordinator.py | 26 +++++++++++++++++++++++--- 2 files changed, 29 insertions(+), 7 deletions(-) diff --git a/sendspin/serve/coordinator.py b/sendspin/serve/coordinator.py index f8ca5ca..6340b25 100644 --- a/sendspin/serve/coordinator.py +++ b/sendspin/serve/coordinator.py @@ -261,7 +261,9 @@ async def _stream_audio_loop(self) -> None: audio_source = await decode_audio(self.source, source_format=self.source_format) fmt = audio_source.format - play_start_us = server_now_us() + DEFAULT_INITIAL_DELAY_US + stream_start_us = server_now_us() + DEFAULT_INITIAL_DELAY_US + play_start_us = stream_start_us + samples_sent = 0 async for pcm_chunk in audio_source.generator: if self._shutdown_requested: @@ -281,8 +283,6 @@ async def _stream_audio_loop(self) -> None: last_stats_time = now frame_stride = (fmt.bit_depth // 8) * fmt.channels - sample_count = len(pcm_chunk) // frame_stride - chunk_duration_us = int(sample_count * 1_000_000 / fmt.sample_rate) chunk_msg = AudioChunk( pcm_bytes=pcm_chunk, @@ -294,7 +294,9 @@ async def _stream_audio_loop(self) -> None: for queue in self._audio_queues: queue.put(chunk_msg) - play_start_us += chunk_duration_us + # Stamp from the total sample count so rounding never puts a chunk before the previous one's end. + samples_sent += len(pcm_chunk) // frame_stride + play_start_us = stream_start_us + samples_sent * 1_000_000 // fmt.sample_rate now_us = server_now_us() ahead_us = play_start_us - DEFAULT_INITIAL_DELAY_US - now_us diff --git a/tests/serve/test_coordinator.py b/tests/serve/test_coordinator.py index 773691b..47639a2 100644 --- a/tests/serve/test_coordinator.py +++ b/tests/serve/test_coordinator.py @@ -2,13 +2,14 @@ from __future__ import annotations - -from unittest.mock import MagicMock +from collections.abc import AsyncGenerator +from unittest.mock import MagicMock, patch import pytest +from aiosendspin.server import AudioFormat from sendspin.serve.coordinator import ServeCoordinator -from sendspin.serve.ipc import WorkerClientCount, WorkerError, WorkerListening +from sendspin.serve.ipc import AudioChunk, WorkerClientCount, WorkerError, WorkerListening @pytest.fixture @@ -104,3 +105,22 @@ def test_check_worker_health_shuts_down_on_unexpected_worker_crash( assert coordinator._shutdown_requested is True assert coordinator._reported_crashed == {0} + + +@pytest.mark.asyncio +async def test_chunk_timestamps_follow_sample_count(coordinator: ServeCoordinator) -> None: + """Chunk timestamps must not fall behind the audio they follow.""" + sent: list[AudioChunk] = [] + coordinator._audio_queues = [MagicMock(put=sent.append)] + coordinator._total_listeners.value = 1 + + async def chunks() -> AsyncGenerator[bytes, None]: + for _ in range(4): + yield bytes(4096 * 4) + coordinator._shutdown_requested = True + + source = MagicMock(generator=chunks(), format=AudioFormat(48000, 16, 2)) + with patch("sendspin.serve.coordinator.decode_audio", return_value=source): + await coordinator._stream_audio_loop() + + assert sent[3].play_start_us - sent[0].play_start_us == 256_000