fix(audio): prevent stale PCM buildup under A2DP backpressure

guest/test_udp_to_sink.py

@@ -0,0 +1,85 @@
+#!/usr/bin/env python3
+"""Regression tests for the live UDP -> pw-play relay."""
+
+import socket
+import unittest
+
+import udp_to_sink as bridge
+
+
+class FakeSocket:
+    def __init__(self, events):
+        self.events = list(events)
+        self.recv_calls = 0
+
+    def recvfrom(self, _size):
+        self.recv_calls += 1
+        event = self.events.pop(0)
+        if event == "timeout":
+            raise socket.timeout
+        return event, ("127.0.0.1", 4010)
+
+
+class RelayTests(unittest.TestCase):
+    def test_backpressure_drops_frames_but_keeps_draining_udp(self):
+        sock = FakeSocket([b"x" * 1400 for _ in range(100)])
+
+        def blocked(_fd, _data):
+            raise BlockingIOError
+
+        misses = 0
+        for _ in range(100):
+            misses, written = bridge._relay_once(sock, 1, misses, blocked)
+            self.assertFalse(written)
+        self.assertEqual(100, sock.recv_calls)
+        self.assertEqual(0, misses)
+
+    def test_eagain_does_not_escape_or_crash(self):
+        def blocked(_fd, _data):
+            raise BlockingIOError
+
+        self.assertFalse(bridge._write_live(1, b"pcm", blocked))
+
+    def test_protocol_oversize_is_dropped_without_partial_write(self):
+        calls = []
+        self.assertFalse(bridge._write_live(
+            1, b"x" * (bridge.PIPE_BUF + 1),
+            lambda fd, data: calls.append((fd, data))))
+        self.assertEqual([], calls)
+
+    def test_current_packet_is_written_intact(self):
+        calls = []
+        frame = b"x" * 1400
+
+        def capture(fd, data):
+            calls.append((fd, data))
+            return len(data)
+
+        self.assertTrue(bridge._write_live(1, frame, capture))
+        self.assertEqual([(1, frame)], calls)
+
+    def test_partial_write_is_not_reported_as_success(self):
+        self.assertFalse(bridge._write_live(
+            1, b"pcm", lambda _fd, _data: 1))
+
+    def test_silence_keepalive_starts_after_existing_debounce(self):
+        sock = FakeSocket(["timeout"] * bridge.MISS_LIMIT)
+        writes = []
+        misses = 0
+
+        def capture(fd, data):
+            writes.append((fd, data))
+            return len(data)
+
+        for _ in range(bridge.MISS_LIMIT - 1):
+            misses, written = bridge._relay_once(
+                sock, 1, misses, capture)
+            self.assertFalse(written)
+        misses, written = bridge._relay_once(
+            sock, 1, misses, capture)
+        self.assertTrue(written)
+        self.assertEqual([(1, bridge.SILENCE)], writes)
+
+
+if __name__ == "__main__":
+    unittest.main()

guest/udp_to_sink.py

@@ -17,8 +17,10 @@
 
 LATENCY (2026-07-10): two mechanisms silently added delay on top of the 250ms.
 (1) The stdin pipe to pw-play is a 64KB Linux pipe = a hidden 341ms of audio;
-    it is now shrunk to 16KB (~85ms) via F_SETPIPE_SZ, and the Popen is
-    unbuffered so Python adds no batching of its own.
+    it is now shrunk to 16KB (~85ms) via F_SETPIPE_SZ, Popen is unbuffered,
+    and writes are nonblocking. If the real-time sink is full, the current
+    frame is dropped so the receiver keeps draining UDP instead of retaining
+    stale PCM in the large socket receive buffer.
 (2) The old gap-filler injected silence on EVERY 20ms socket timeout. When
     packets were merely LATE (BLE airtime burst, VM scheduling) the late data
     still arrived after the silence -- net queued bytes grew, playback consumes
@@ -52,37 +54,66 @@
 GAP = 0.02
 SILENCE = b"\x00" * int(48000 * 2 * 2 * GAP)  # 20ms -> matches the timeout below
 MISS_LIMIT = 5  # inject silence only after 5*GAP = 100ms of true no-data
+PIPE_BUF = 4096  # Linux atomic pipe-write limit; sender packets are <=1400B
 
-sock = socket.socket(socket.AF_INET, socket.SOCK_DGRAM)
-sock.setsockopt(socket.SOL_SOCKET, socket.SO_RCVBUF, 1 << 20)
-sock.bind(("0.0.0.0", 4010))
-sock.settimeout(GAP)
 
-while True:
-    player = subprocess.Popen(
-        ["pw-play", "--format=s16", "--rate=48000", "--channels=2",
-         f"--latency={LATENCY_MS}ms", "-"],
-        stdin=subprocess.PIPE, bufsize=0, env=env)
+def _write_live(pipe_fd, data, writer=os.write):
+    """Write one current PCM frame or drop it if the sink is backpressured.
+
+    OpenSpan sender datagrams (<=1400B) and SILENCE (3840B) fit PIPE_BUF, so
+    nonblocking writes are atomic: the whole frame is accepted or EAGAIN.
+    Keeping the UDP receiver live is more important than replaying stale PCM.
+    """
+    if len(data) > PIPE_BUF:
+        return False
     try:
-        fcntl.fcntl(player.stdin.fileno(), F_SETPIPE_SZ, PIPE_BYTES)
-    except OSError:
-        pass  # kernel refuses -> default 64KB pipe; higher latency, still works
-    misses = 0
+        return writer(pipe_fd, data) == len(data)
+    except BlockingIOError:
+        return False
+
+
+def _relay_once(sock, pipe_fd, misses, writer=os.write):
+    """Drain one UDP frame (or keepalive timeout) without blocking on sink."""
     try:
-        while player.poll() is None:
-            try:
-                data, _ = sock.recvfrom(8192)
-                misses = 0
-            except socket.timeout:
-                misses += 1
-                if misses < MISS_LIMIT:
-                    continue  # brief jitter: let the 250ms cushion ride it
-                data = SILENCE  # real pause: keep the A2DP stream alive
-            player.stdin.write(data)
-    except (BrokenPipeError, OSError):
-        pass
-    finally:
+        data, _ = sock.recvfrom(8192)
+        misses = 0
+    except socket.timeout:
+        misses += 1
+        if misses < MISS_LIMIT:
+            return misses, False
+        data = SILENCE  # real pause: keep the A2DP stream alive
+    return misses, _write_live(pipe_fd, data, writer)
+
+
+def main():
+    sock = socket.socket(socket.AF_INET, socket.SOCK_DGRAM)
+    sock.setsockopt(socket.SOL_SOCKET, socket.SO_RCVBUF, 1 << 20)
+    sock.bind(("0.0.0.0", 4010))
+    sock.settimeout(GAP)
+
+    while True:
+        player = subprocess.Popen(
+            ["pw-play", "--format=s16", "--rate=48000", "--channels=2",
+             f"--latency={LATENCY_MS}ms", "-"],
+            stdin=subprocess.PIPE, bufsize=0, env=env)
         try:
-            player.terminate()
-        except Exception:
+            pipe_fd = player.stdin.fileno()
+            try:
+                fcntl.fcntl(pipe_fd, F_SETPIPE_SZ, PIPE_BYTES)
+            except OSError:
+                pass  # kernel refuses -> default pipe; still nonblocking
+            os.set_blocking(pipe_fd, False)
+            misses = 0
+            while player.poll() is None:
+                misses, _ = _relay_once(sock, pipe_fd, misses)
+        except (BrokenPipeError, OSError):
             pass
+        finally:
+            try:
+                player.terminate()
+            except Exception:
+                pass
+
+
+if __name__ == "__main__":
+    main()