src/audio/cpal_impl.rs

@@ -7,6 +7,9 @@
 //!
 //! - **Capture**: mono, 48 kHz, S16 PCM, emitted as `Vec<i16>` frames of
 //!   [`CAPTURE_FRAME`] (960 = 20 ms) samples — matching the encoder/jitter frame.
+//!   The RT capture callback only downmixes and pushes samples into a lock-free
+//!   ring; the owning thread drains that ring, frames it, and sends — so the
+//!   callback never allocates, locks, or touches an mpsc channel.
 //! - **Playback**: stereo interleaved ([`PLAYBACK_CHANNELS`]) S16 PCM at 48 kHz,
 //!   drained from a ring buffer that is paced to the device's hardware clock via
 //!   `ring_fill` exactly as the PipeWire backend does.
@@ -20,7 +23,12 @@
 //! stream, plays it, and keeps it alive until the per-worker `running` flag flips
 //! (set by `stop`). The struct holds only `Send` handles (the flag + the join
 //! handle). The stream's RT callback does the actual audio work; the owning
-//! thread additionally feeds the playback ring from the network mixer.
+//! thread additionally feeds the playback ring (or drains the capture ring).
+//!
+//! `start_*` does not return until the owning thread reports back over a readiness
+//! channel that the device resolved and the stream is built and playing — so a
+//! device/format/WASAPI failure surfaces as a real `Err` to the caller instead of
+//! leaving the UI in a joined-but-silent room.
 //!
 //! ## Sample rate
 //!
@@ -30,7 +38,7 @@
 //! sample-rate support (resampling) is a Phase 1.1 follow-up.
 
 use std::sync::atomic::{AtomicBool, AtomicU64, AtomicUsize, Ordering};
-use std::sync::mpsc::{Receiver, RecvTimeoutError, Sender};
+use std::sync::mpsc::{channel, Receiver, RecvTimeoutError, Sender};
 use std::sync::{Arc, Mutex};
 use std::thread::{self, JoinHandle};
 use std::time::Duration;
@@ -49,6 +57,13 @@ const SAMPLE_RATE: u32 = 48_000;
 /// Mono capture frame: 960 samples = 20 ms @ 48 kHz. Matches the PipeWire backend
 /// and `core::jitter::FRAME_SAMPLES`.
 const CAPTURE_FRAME: usize = 960;
+/// Lock-free capture ring capacity (mono samples) between the RT callback and the
+/// owning drain thread: 8 frames = 160 ms of headroom, so a scheduling hiccup on
+/// the drain thread doesn't immediately overrun the RT producer.
+const CAPTURE_RING_CAPACITY: usize = CAPTURE_FRAME * 8;
+/// How long the capture drain thread sleeps when the ring is momentarily empty,
+/// before polling again. Small enough to stay well under the 20 ms frame cadence.
+const CAPTURE_POLL: Duration = Duration::from_millis(5);
 /// Playback ring capacity in interleaved samples: 200 ms of stereo @ 48 kHz.
 /// Comfortably above [`PLAYBACK_TARGET_SAMPLES`] so the clock-paced producer has
 /// headroom and never has to drop frames in steady state.
@@ -90,22 +105,20 @@ impl AudioBackend for CpalBackend {
         tx: Sender<Vec<i16>>,
         target_node: Option<String>,
     ) -> Result<(), AudioError> {
-        let mut guard = self.capture.lock().unwrap();
+        let guard = self.capture.lock().unwrap();
         if guard.is_some() {
             return Err(AudioError::Stream("Capture already started".to_string()));
         }
         let running = Arc::new(AtomicBool::new(true));
         let running_thread = running.clone();
+        let (ready_tx, ready_rx) = channel::<Result<(), AudioError>>();
         let thread = thread::Builder::new()
             .name("peerspeak-cpal-capture".to_string())
             .spawn(move || {
-                if let Err(e) = run_capture(tx, target_node, running_thread) {
+                run_capture(tx, target_node, running_thread, ready_tx);
-                    crate::log_msg(&format!("cpal capture error: {e}"));
-                }
             })
             .map_err(|e| AudioError::Init(e.to_string()))?;
-        *guard = Some(StreamWorker { running, thread });
+        finish_start(guard, StreamWorker { running, thread }, ready_rx, "capture")
-        Ok(())
     }
 
     fn start_playback(
@@ -114,22 +127,20 @@ impl AudioBackend for CpalBackend {
         target_node: Option<String>,
         ring_fill: Arc<AtomicUsize>,
     ) -> Result<(), AudioError> {
-        let mut guard = self.playback.lock().unwrap();
+        let guard = self.playback.lock().unwrap();
         if guard.is_some() {
             return Err(AudioError::Stream("Playback already started".to_string()));
         }
         let running = Arc::new(AtomicBool::new(true));
         let running_thread = running.clone();
+        let (ready_tx, ready_rx) = channel::<Result<(), AudioError>>();
         let thread = thread::Builder::new()
             .name("peerspeak-cpal-playback".to_string())
             .spawn(move || {
-                if let Err(e) = run_playback(rx, target_node, ring_fill, running_thread) {
+                run_playback(rx, target_node, ring_fill, running_thread, ready_tx);
-                    crate::log_msg(&format!("cpal playback error: {e}"));
-                }
             })
             .map_err(|e| AudioError::Init(e.to_string()))?;
-        *guard = Some(StreamWorker { running, thread });
+        finish_start(guard, StreamWorker { running, thread }, ready_rx, "playback")
-        Ok(())
     }
 
     fn stop(&self) -> Result<(), AudioError> {
@@ -143,6 +154,38 @@ impl AudioBackend for CpalBackend {
     }
 }
 
+/// Block until the just-spawned worker reports (over `ready_rx`) that its stream
+/// is built and playing, then either install it (`Ok`) or join it and surface the
+/// real error. This is what makes `start_capture`/`start_playback` fail loudly
+/// instead of returning `Ok` into a joined-but-silent room (Codex review W1).
+fn finish_start(
+    mut guard: std::sync::MutexGuard<'_, Option<StreamWorker>>,
+    worker: StreamWorker,
+    ready_rx: Receiver<Result<(), AudioError>>,
+    what: &str,
+) -> Result<(), AudioError> {
+    match ready_rx.recv() {
+        Ok(Ok(())) => {
+            *guard = Some(worker);
+            Ok(())
+        }
+        // Setup failed (Err) or the worker exited before reporting (recv Err):
+        // either way it has stopped, so reap it and surface the error.
+        Ok(Err(e)) => {
+            worker.running.store(false, Ordering::Relaxed);
+            let _ = worker.thread.join();
+            Err(e)
+        }
+        Err(_) => {
+            worker.running.store(false, Ordering::Relaxed);
+            let _ = worker.thread.join();
+            Err(AudioError::Init(format!(
+                "cpal {what} worker exited before reporting readiness"
+            )))
+        }
+    }
+}
+
 // ---------------------------------------------------------------------------
 // Device enumeration (for the settings device pickers)
 // ---------------------------------------------------------------------------
@@ -285,53 +328,108 @@ fn run_capture(
     tx: Sender<Vec<i16>>,
     target: Option<String>,
     running: Arc<AtomicBool>,
-) -> Result<(), AudioError> {
+    ready: Sender<Result<(), AudioError>>,
-    let (device, config, sample_format) = resolve(false, target)?;
+) {
-    let channels = config.channels as usize;
+    // The RT callback pushes mono samples into this lock-free ring; we drain it on
+    // this (non-RT) thread, so the callback never allocates or sends on a channel.
+    let rb = HeapRb::<i16>::new(CAPTURE_RING_CAPACITY);
+    let (producer, mut consumer) = rb.split();
+    let overrun = Arc::new(AtomicU64::new(0));
 
-    let stream = match sample_format {
+    // Fallible device/stream setup. We report the real error to `start_capture`
-        SampleFormat::F32 => build_input::<f32>(&device, &config, tx, channels),
+    // before doing any work, so a join never lands in a silent room.
-        SampleFormat::I16 => build_input::<i16>(&device, &config, tx, channels),
+    let setup = || -> Result<(Stream, String, SampleFormat, usize), AudioError> {
-        SampleFormat::U16 => build_input::<u16>(&device, &config, tx, channels),
+        let (device, config, sample_format) = resolve(false, target)?;
-        other => Err(AudioError::Stream(format!(
+        let channels = config.channels as usize;
-            "unsupported capture sample format: {other:?}"
+        let stream = match sample_format {
-        ))),
+            SampleFormat::F32 => {
-    }?;
+                build_input::<f32, _>(&device, &config, producer, channels, overrun.clone())
+            }
+            SampleFormat::I16 => {
+                build_input::<i16, _>(&device, &config, producer, channels, overrun.clone())
+            }
+            SampleFormat::U16 => {
+                build_input::<u16, _>(&device, &config, producer, channels, overrun.clone())
+            }
+            other => Err(AudioError::Stream(format!(
+                "unsupported capture sample format: {other:?}"
+            ))),
+        }?;
+        stream.play().map_err(|e| AudioError::Stream(e.to_string()))?;
+        let name = device.name().unwrap_or_else(|_| "<unknown>".to_string());
+        Ok((stream, name, sample_format, channels))
+    };
 
-    stream.play().map_err(|e| AudioError::Stream(e.to_string()))?;
+    let (stream, dev_name, sample_format, channels) = match setup() {
+        Ok(v) => {
+            let _ = ready.send(Ok(()));
+            v
+        }
+        Err(e) => {
+            let _ = ready.send(Err(e));
+            return;
+        }
+    };
+    crate::log_msg(&format!(
+        "cpal capture started: device='{dev_name}' format={sample_format:?} channels={channels} rate={SAMPLE_RATE} Hz"
+    ));
 
-    // The RT callback does the work; this thread just keeps `stream` alive until
+    // Drain the RT ring on this thread: pop mono samples, frame them (the `Vec`
-    // `stop()` flips the flag, at which point the stream is dropped (= stopped).
+    // allocation lives here, off the RT path), and send completed frames. Keep
+    // `stream` alive until `stop()` flips the flag.
+    let mut acc = FrameAccumulator::new(CAPTURE_FRAME);
+    let mut last_overrun = 0u64;
     while running.load(Ordering::Relaxed) {
-        thread::sleep(WORKER_POLL);
+        let mut drained = false;
+        while let Some(sample) = consumer.try_pop() {
+            drained = true;
+            if let Some(frame) = acc.push(sample) {
+                // Consumer gone (call ended) → stop feeding; the stream is
+                // dropped below on the way out.
+                if tx.send(frame).is_err() {
+                    drop(stream);
+                    return;
+                }
+            }
+        }
+        let o = overrun.load(Ordering::Relaxed);
+        if o != last_overrun {
+            crate::log_msg(&format!(
+                "cpal capture overrun: dropped {} samples (drain thread fell behind)",
+                o - last_overrun
+            ));
+            last_overrun = o;
+        }
+        if !drained {
+            thread::sleep(CAPTURE_POLL);
+        }
     }
-    Ok(())
+    drop(stream);
 }
 
-fn build_input<T>(
+fn build_input<T, P>(
     device: &Device,
     config: &StreamConfig,
-    tx: Sender<Vec<i16>>,
+    mut producer: P,
     channels: usize,
+    overrun: Arc<AtomicU64>,
 ) -> Result<Stream, AudioError>
 where
     T: SizedSample + Send + 'static,
     i16: FromSample<T>,
+    P: Producer<Item = i16> + Send + 'static,
 {
-    let mut acc = FrameAccumulator::new(CAPTURE_FRAME);
     let err_fn = |e| crate::log_msg(&format!("cpal capture stream error: {e}"));
     device
         .build_input_stream::<T, _, _>(
             config,
             move |data: &[T], _| {
+                // RT-safe: downmix + wait-free push only. A full ring means the
+                // drain thread stalled; count the drop and keep going.
                 for frame in data.chunks_exact(channels) {
                     let mono = downmix_to_mono(frame);
-                    if let Some(full) = acc.push(mono) {
+                    if producer.try_push(mono).is_err() {
-                        // Consumer gone (call ended) → stop feeding; the owning
+                        overrun.fetch_add(1, Ordering::Relaxed);
-                        // thread will drop the stream on `stop()`.
-                        if tx.send(full).is_err() {
-                            return;
-                        }
                     }
                 }
             },
@@ -393,9 +491,8 @@ fn run_playback(
     target: Option<String>,
     ring_fill: Arc<AtomicUsize>,
     running: Arc<AtomicBool>,
-) -> Result<(), AudioError> {
+    ready: Sender<Result<(), AudioError>>,
-    let (device, config, sample_format) = resolve(true, target)?;
+) {
-
     let rb = HeapRb::<i16>::new(RING_CAPACITY);
     let (mut producer, consumer) = rb.split();
 
@@ -413,22 +510,43 @@ fn run_playback(
     let underrun = Arc::new(AtomicU64::new(0));
     let dropped = Arc::new(AtomicU64::new(0));
 
-    let stream = match sample_format {
+    // Fallible device/stream setup; report the real error to `start_playback`
-        SampleFormat::F32 => {
+    // before any work so a failure surfaces instead of a silent room. `consumer`
-            build_output::<f32, _>(&device, &config, consumer, ring_fill.clone(), underrun.clone())
+    // is moved into the output callback here.
-        }
+    let setup = || -> Result<(Stream, String, SampleFormat), AudioError> {
-        SampleFormat::I16 => {
+        let (device, config, sample_format) = resolve(true, target)?;
-            build_output::<i16, _>(&device, &config, consumer, ring_fill.clone(), underrun.clone())
+        let stream = match sample_format {
-        }
+            SampleFormat::F32 => {
-        SampleFormat::U16 => {
+                build_output::<f32, _>(&device, &config, consumer, ring_fill.clone(), underrun.clone())
-            build_output::<u16, _>(&device, &config, consumer, ring_fill.clone(), underrun.clone())
+            }
-        }
+            SampleFormat::I16 => {
-        other => Err(AudioError::Stream(format!(
+                build_output::<i16, _>(&device, &config, consumer, ring_fill.clone(), underrun.clone())
-            "unsupported playback sample format: {other:?}"
+            }
-        ))),
+            SampleFormat::U16 => {
-    }?;
+                build_output::<u16, _>(&device, &config, consumer, ring_fill.clone(), underrun.clone())
+            }
+            other => Err(AudioError::Stream(format!(
+                "unsupported playback sample format: {other:?}"
+            ))),
+        }?;
+        stream.play().map_err(|e| AudioError::Stream(e.to_string()))?;
+        let name = device.name().unwrap_or_else(|_| "<unknown>".to_string());
+        Ok((stream, name, sample_format))
+    };
 
-    stream.play().map_err(|e| AudioError::Stream(e.to_string()))?;
+    let (stream, dev_name, sample_format) = match setup() {
+        Ok(v) => {
+            let _ = ready.send(Ok(()));
+            v
+        }
+        Err(e) => {
+            let _ = ready.send(Err(e));
+            return;
+        }
+    };
+    crate::log_msg(&format!(
+        "cpal playback started: device='{dev_name}' format={sample_format:?} channels={PLAYBACK_CHANNELS} rate={SAMPLE_RATE} Hz"
+    ));
 
     let logger = spawn_health_logger(
         running.clone(),
@@ -456,7 +574,6 @@ fn run_playback(
     running.store(false, Ordering::Relaxed);
     let _ = logger.join();
     drop(stream);
-    Ok(())
 }
 
 fn build_output<T, C>(