Merge A6 playback handoff fix

build.rs

@@ -2,11 +2,11 @@ use std::process::Command;
 
 fn main() {
     println!("cargo:rerun-if-changed=.git/HEAD");
-    if let Ok(head) = std::fs::read_to_string(".git/HEAD") {
+    if let Ok(head) = std::fs::read_to_string(".git/HEAD")
-        if let Some(reference) = head.strip_prefix("ref: ") {
+        && let Some(reference) = head.strip_prefix("ref: ")
+    {
         println!("cargo:rerun-if-changed=.git/{}", reference.trim());
     }
-    }
 
     let short = Command::new("git")
         .args(["rev-parse", "--short=8", "HEAD"])

src/core/mod.rs

@@ -31,6 +31,12 @@ use tokio::sync::{Mutex, mpsc};
 
 type CoalesceStore = Arc<StdMutex<HashMap<CoalesceKey, CoreCommand>>>;
 
+// Mixer -> playback-worker handoff. The playback ring itself targets three
+// 20ms frames; allow at most two more in flight so worker lag applies
+// backpressure before the ring can overshoot to its 200ms cap (A6).
+const PLAYBACK_HANDOFF_QUEUE_FRAMES: usize = 2;
+const PLAYBACK_HANDOFF_RETRY: Duration = Duration::from_millis(1);
+
 pub struct CoreController {
     reliable_tx: mpsc::UnboundedSender<CoreCommand>,
     coalesce: CoalesceStore,
@@ -156,6 +162,22 @@ fn audio_datagram_len_ok(len: usize) -> bool {
     (4..=4 + MAX_OPUS_PAYLOAD).contains(&len)
 }
 
+async fn send_playback_frame(
+    tx: &std::sync::mpsc::SyncSender<Vec<i16>>,
+    mut frame: Vec<i16>,
+) -> bool {
+    loop {
+        match tx.try_send(frame) {
+            Ok(()) => return true,
+            Err(std::sync::mpsc::TrySendError::Full(returned)) => {
+                frame = returned;
+                tokio::time::sleep(PLAYBACK_HANDOFF_RETRY).await;
+            }
+            Err(std::sync::mpsc::TrySendError::Disconnected(_)) => return false,
+        }
+    }
+}
+
 /// The presence label to broadcast for a detected game: its display name,
 /// sanitized + length-capped, or `None` when there's no game or no broadcastable
 /// name (a Steam appid without a manifest name, or a label that sanitizes empty).
@@ -1654,7 +1676,8 @@ async fn run_core_loop(
 
                 // Setup raw audio channels
                 let (capture_tx, capture_rx) = std::sync::mpsc::channel();
-                let (playback_tx, playback_rx) = std::sync::mpsc::channel();
+                let (playback_tx, playback_rx) =
+                    std::sync::mpsc::sync_channel(PLAYBACK_HANDOFF_QUEUE_FRAMES);
 
                 // Echo cancellation: if enabled, load PipeWire's echo-cancel module
                 // bound to the chosen real devices and route capture/playback
@@ -2114,7 +2137,7 @@ async fn run_core_loop(
                             mixed
                         };
 
-                        if playback_tx.send(frame_to_send).is_err() {
+                        if !send_playback_frame(&playback_tx, frame_to_send).await {
                             break;
                         }
 
@@ -3222,12 +3245,15 @@ async fn run_core_loop(
 mod tests {
     use super::{
         KnownPeers, MAX_OPUS_PAYLOAD, MAX_RETAINED_PEERS, MIC_LEVEL_REPORT_SAMPLES, MicLevelMeter,
-        PeerSpeakTicket, admit_retained, apply_peer_volume, apply_volume, audio_datagram_len_ok,
+        PLAYBACK_HANDOFF_QUEUE_FRAMES, PeerSpeakTicket, admit_retained, apply_peer_volume,
-        coalesce_insert, coalesce_pop, frame_level, mix_frames, mix_stereo_frames,
+        apply_volume, audio_datagram_len_ok, coalesce_insert, coalesce_pop, frame_level,
-        next_game_change, should_auto_fetch, stereo_to_mono,
+        mix_frames, mix_stereo_frames, next_game_change, send_playback_frame, should_auto_fetch,
+        stereo_to_mono,
     };
     use crate::core::messages::{CoalesceKey, CoreCommand, coalesce_key};
     use std::collections::{HashMap, HashSet};
+    use std::sync::mpsc::sync_channel;
+    use std::time::Duration;
 
     fn endpoint_id() -> iroh::EndpointId {
         iroh::SecretKey::generate().public()
@@ -3461,6 +3487,33 @@ mod tests {
         assert!(!audio_datagram_len_ok(5 + MAX_OPUS_PAYLOAD));
     }
 
+    #[tokio::test]
+    async fn playback_handoff_waits_for_bounded_queue_space() {
+        let (tx, rx) = sync_channel::<Vec<i16>>(PLAYBACK_HANDOFF_QUEUE_FRAMES);
+        for n in 0..PLAYBACK_HANDOFF_QUEUE_FRAMES {
+            tx.try_send(vec![n as i16]).unwrap();
+        }
+
+        let worker = std::thread::spawn(move || {
+            std::thread::sleep(Duration::from_millis(20));
+            for n in 0..PLAYBACK_HANDOFF_QUEUE_FRAMES {
+                assert_eq!(rx.recv().unwrap(), vec![n as i16]);
+            }
+            assert_eq!(rx.recv().unwrap(), vec![99, 100]);
+        });
+
+        let sent = tokio::time::timeout(
+            Duration::from_secs(1),
+            send_playback_frame(&tx, vec![99, 100]),
+        )
+        .await
+        .expect("bounded handoff should unblock after the worker drains a frame");
+
+        assert!(sent);
+        drop(tx);
+        worker.join().unwrap();
+    }
+
     #[test]
     fn mic_meter_holds_the_peak_across_the_window() {
         let mut m = MicLevelMeter::new();