diff --git a/src/audio/cpal_impl.rs b/src/audio/cpal_impl.rs index 81e200e..f73c030 100644 --- a/src/audio/cpal_impl.rs +++ b/src/audio/cpal_impl.rs @@ -184,21 +184,72 @@ fn wait_for_stream_start( /// Windows audio backend. See module docs. pub struct CpalBackend { - capture: Mutex>, - playback: Mutex>, + capture: Mutex, + playback: Mutex, } -/// A spawned owning thread plus the flag that tells it to drop its stream and exit. +/// A spawned owning thread plus the flags that coordinate its lifetime: `running` +/// tells it to drop its stream and exit; `exited` is flipped true (by [`ExitGuard`] +/// in the thread body) when it actually returns, so a *detached* wedged start can be +/// detected as finished later (review B3). struct StreamWorker { running: Arc, + exited: Arc, thread: JoinHandle<()>, } +/// Flips its flag true when dropped, marking a worker thread as exited. Lives at the +/// top of the worker closure so it fires on normal return, panic unwind, or whenever +/// a wedged driver call finally releases the thread — which is what lets a [`SlotState::Wedged`] +/// tombstone (B3) know its orphan is gone. +struct ExitGuard(Arc); +impl Drop for ExitGuard { + fn drop(&mut self) { + self.0.store(true, Ordering::Relaxed); + } +} + +/// The lifecycle state of a capture or playback slot. +enum SlotState { + /// No stream — a new start may proceed. + Idle, + /// A live, started stream owned by its worker thread. + Live(StreamWorker), + /// A start that timed out wedged in a driver call (review B3). Its worker thread + /// was *detached* rather than joined — joining would re-introduce the unbounded + /// hang [`FINISH_START_TIMEOUT`] exists to prevent — so it may still be alive, + /// holding the COM/device handle. `exited` flips true when that orphan finally + /// returns. New starts are rejected until then, so retries against a permanently + /// wedged device don't pile up more orphan threads. + Wedged { exited: Arc }, +} + +/// Inspect a slot before starting a stream into it. Clears a [`SlotState::Wedged`] +/// tombstone whose orphan has since exited (the slot becomes reusable), but rejects +/// a start while a wedged orphan is still alive or a live stream already owns the +/// slot. Pure w.r.t. the passed state, so the tombstone logic is unit-testable (B3). +fn ensure_idle(state: &mut SlotState, what: &str) -> Result<(), AudioError> { + match state { + SlotState::Idle => Ok(()), + SlotState::Live(_) => Err(AudioError::Stream(format!("{what} already started"))), + SlotState::Wedged { exited } => { + if exited.load(Ordering::Relaxed) { + *state = SlotState::Idle; + Ok(()) + } else { + Err(AudioError::Stream(format!( + "{what} is recovering from an unresponsive audio device; retry shortly" + ))) + } + } + } +} + impl CpalBackend { pub fn new() -> Self { Self { - capture: Mutex::new(None), - playback: Mutex::new(None), + capture: Mutex::new(SlotState::Idle), + playback: Mutex::new(SlotState::Idle), } } } @@ -215,20 +266,30 @@ impl AudioBackend for CpalBackend { tx: Sender>, target_node: Option, ) -> Result<(), AudioError> { - let guard = self.capture.lock().unwrap(); - if guard.is_some() { - return Err(AudioError::Stream("Capture already started".to_string())); - } + let mut guard = self.capture.lock().unwrap(); + ensure_idle(&mut guard, "capture")?; let running = Arc::new(AtomicBool::new(true)); + let exited = Arc::new(AtomicBool::new(false)); let running_thread = running.clone(); + let exited_thread = exited.clone(); let (ready_tx, ready_rx) = channel::>(); let thread = thread::Builder::new() .name("peerspeak-cpal-capture".to_string()) .spawn(move || { + let _exit = ExitGuard(exited_thread); run_capture(tx, target_node, running_thread, ready_tx); }) .map_err(|e| AudioError::Init(e.to_string()))?; - finish_start(guard, StreamWorker { running, thread }, ready_rx, "capture") + finish_start( + guard, + StreamWorker { + running, + exited, + thread, + }, + ready_rx, + "capture", + ) } fn start_playback( @@ -237,22 +298,27 @@ impl AudioBackend for CpalBackend { target_node: Option, ring_fill: Arc, ) -> Result<(), AudioError> { - let guard = self.playback.lock().unwrap(); - if guard.is_some() { - return Err(AudioError::Stream("Playback already started".to_string())); - } + let mut guard = self.playback.lock().unwrap(); + ensure_idle(&mut guard, "playback")?; let running = Arc::new(AtomicBool::new(true)); + let exited = Arc::new(AtomicBool::new(false)); let running_thread = running.clone(); + let exited_thread = exited.clone(); let (ready_tx, ready_rx) = channel::>(); let thread = thread::Builder::new() .name("peerspeak-cpal-playback".to_string()) .spawn(move || { + let _exit = ExitGuard(exited_thread); run_playback(rx, target_node, ring_fill, running_thread, ready_tx); }) .map_err(|e| AudioError::Init(e.to_string()))?; finish_start( guard, - StreamWorker { running, thread }, + StreamWorker { + running, + exited, + thread, + }, ready_rx, "playback", ) @@ -260,9 +326,21 @@ impl AudioBackend for CpalBackend { fn stop(&self) -> Result<(), AudioError> { for slot in [&self.capture, &self.playback] { - if let Some(worker) = slot.lock().unwrap().take() { - worker.running.store(false, Ordering::Relaxed); - let _ = worker.thread.join(); + let mut guard = slot.lock().unwrap(); + match std::mem::replace(&mut *guard, SlotState::Idle) { + SlotState::Live(worker) => { + worker.running.store(false, Ordering::Relaxed); + let _ = worker.thread.join(); + } + // A wedged orphan was detached and can't be joined. If it has since + // exited the slot is now clear; otherwise restore the tombstone so a + // later start still sees the device is recovering (B3). + SlotState::Wedged { exited } => { + if !exited.load(Ordering::Relaxed) { + *guard = SlotState::Wedged { exited }; + } + } + SlotState::Idle => {} } } Ok(()) @@ -274,7 +352,7 @@ impl AudioBackend for CpalBackend { /// 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>, + mut guard: std::sync::MutexGuard<'_, SlotState>, worker: StreamWorker, ready_rx: Receiver>, what: &str, @@ -284,7 +362,7 @@ fn finish_start( // wedged the worker before it could report (review W6). match ready_rx.recv_timeout(FINISH_START_TIMEOUT) { Ok(Ok(())) => { - *guard = Some(worker); + *guard = SlotState::Live(worker); Ok(()) } // Setup failed (Err) or the worker disconnected before reporting: either @@ -304,10 +382,17 @@ fn finish_start( Err(RecvTimeoutError::Timeout) => { // The worker is wedged in a driver call. Signal it to exit, but DETACH // rather than join — joining would re-introduce the unbounded hang this - // timeout exists to prevent. The thread unwinds on its own if/when the - // driver call ever returns. - worker.running.store(false, Ordering::Relaxed); - drop(worker.thread); + // timeout exists to prevent. Leave a Wedged tombstone so subsequent + // starts are rejected until the orphan's ExitGuard flips `exited`, rather + // than spawning more orphan threads against the same dead device (B3). + let StreamWorker { + running, + exited, + thread, + } = worker; + running.store(false, Ordering::Relaxed); + drop(thread); + *guard = SlotState::Wedged { exited }; Err(AudioError::Init(format!( "cpal {what} did not start within {FINISH_START_TIMEOUT:?}" ))) @@ -431,14 +516,28 @@ fn find_device_by_name(host: &cpal::Host, output: bool, name: &str) -> Option Option { + let lo = r_min.max(MIN_DEVICE_RATE); + let hi = r_max.min(MAX_DEVICE_RATE); + (lo <= hi).then(|| SAMPLE_RATE.clamp(lo, hi)) +} + /// Pick a stream config. Preference order, best (no conversion) first: /// 1. exactly [`SAMPLE_RATE`] at the preferred layout (stereo out / mono in), /// 2. exactly [`SAMPLE_RATE`] at any channel count (rate-exact, backend remaps), -/// 3. the device's default config (native rate/layout, backend resamples + remaps). +/// 3. a supported config at a [`bounded_rate`] near 48 kHz (backend resamples + remaps), +/// 4. the device's default config (only if nothing above is drivable). /// -/// Only case 3 incurs resampling; the backend reads the returned config's rate and -/// channel count and converts at the boundary (W4). A device that exposes no config -/// at all is still a hard error. +/// Cases 3–4 incur resampling; the backend reads the returned config's rate and +/// channel count and converts at the boundary (W4). Case 3 (review B5) is what keeps +/// an oddball endpoint whose default rate is outside the drivable window — but which +/// also exposes a usable in-window config — from being rejected by [`resolve`]. A +/// device that exposes no config at all is still a hard error. fn choose_config(device: &Device, output: bool) -> Result { let ranges: Vec = if output { device @@ -474,7 +573,35 @@ fn choose_config(device: &Device, output: bool) -> Result| -> Option<(cpal::SupportedStreamConfigRange, u32)> { + ranges + .iter() + .filter(|r| channels.is_none_or(|c| r.channels() == c)) + .filter_map(|r| { + bounded_rate(r.min_sample_rate().0, r.max_sample_rate().0) + .map(|rate| (r.clone(), rate)) + }) + .min_by_key(|(_, rate)| rate.abs_diff(SAMPLE_RATE)) + }; + let preferred_channels = if output { PLAYBACK_CHANNELS as u16 } else { 1 }; + if let Some((r, rate)) = pick_bounded(Some(preferred_channels)).or_else(|| pick_bounded(None)) { + crate::log_msg(&format!( + "cpal: device '{}' has no native {SAMPLE_RATE} Hz {} config; using bounded {rate} Hz / {} ch with linear resampling (W4/B5)", + device.name().unwrap_or_else(|_| "".to_string()), + if output { "output" } else { "input" }, + r.channels(), + )); + return Ok(r.with_sample_rate(SampleRate(rate))); + } + + // Case 4: last resort — the device's default config. If its rate is outside the + // drivable window, resolve() rejects it with a clear device error, which is the + // honest outcome: the device exposes nothing this backend can drive. let def = if output { device.default_output_config() } else { @@ -482,7 +609,7 @@ fn choose_config(device: &Device, output: bool) -> Result".to_string()), if output { "output" } else { "input" }, def.sample_rate().0, @@ -1201,6 +1328,63 @@ mod tests { drain_loop(&rx, &running, |_| panic!("no frame should arrive")); } + #[test] + fn bounded_rate_prefers_48k_when_in_window() { + // A device span that contains 48 kHz resolves exactly. + assert_eq!(bounded_rate(44_100, 96_000), Some(SAMPLE_RATE)); + assert_eq!( + bounded_rate(MIN_DEVICE_RATE, MAX_DEVICE_RATE), + Some(SAMPLE_RATE) + ); + } + + #[test] + fn bounded_rate_clamps_to_nearest_in_window_bound() { + // Entirely below 48 kHz → the top bound (closest reachable to 48 kHz). + assert_eq!(bounded_rate(8_000, 16_000), Some(16_000)); + // Entirely above 48 kHz → the bottom bound. + assert_eq!(bounded_rate(88_200, 192_000), Some(88_200)); + } + + #[test] + fn bounded_rate_rejects_spans_outside_the_window() { + assert_eq!(bounded_rate(1_000, 4_000), None); // below the floor + assert_eq!(bounded_rate(400_000, 500_000), None); // above the ceiling + } + + #[test] + fn bounded_rate_intersects_window_edges() { + // Overlaps only the floor: [4k, 8k] ∩ [8k, 384k] = {8k}. + assert_eq!(bounded_rate(4_000, MIN_DEVICE_RATE), Some(MIN_DEVICE_RATE)); + // Overlaps only the ceiling. + assert_eq!( + bounded_rate(MAX_DEVICE_RATE, 500_000), + Some(MAX_DEVICE_RATE) + ); + } + + #[test] + fn ensure_idle_allows_an_idle_slot() { + let mut s = SlotState::Idle; + assert!(ensure_idle(&mut s, "capture").is_ok()); + assert!(matches!(s, SlotState::Idle)); + } + + #[test] + fn ensure_idle_rejects_a_live_wedged_orphan_then_clears_when_it_exits() { + let exited = Arc::new(AtomicBool::new(false)); + let mut s = SlotState::Wedged { + exited: exited.clone(), + }; + // Orphan still alive → reject, tombstone preserved. + assert!(ensure_idle(&mut s, "playback").is_err()); + assert!(matches!(s, SlotState::Wedged { .. })); + // Orphan's ExitGuard fired → the next start clears the tombstone and proceeds. + exited.store(true, Ordering::Relaxed); + assert!(ensure_idle(&mut s, "playback").is_ok()); + assert!(matches!(s, SlotState::Idle)); + } + #[test] fn drain_loop_delivers_frames() { let (tx, rx) = mpsc::channel::>();