Fix audio gaps in serve with multiple workers

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

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