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<Option<StreamWorker>>,
-    playback: Mutex<Option<StreamWorker>>,
+    capture: Mutex<SlotState>,
+    playback: Mutex<SlotState>,
 }
 
-/// 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<AtomicBool>,
+    exited: Arc<AtomicBool>,
     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<AtomicBool>);
+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<AtomicBool> },
+}
+
+/// 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<Vec<i16>>,
         target_node: Option<String>,
     ) -> 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::<Result<(), AudioError>>();
         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<String>,
         ring_fill: Arc<AtomicUsize>,
     ) -> 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::<Result<(), AudioError>>();
         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<StreamWorker>>,
+    mut guard: std::sync::MutexGuard<'_, SlotState>,
     worker: StreamWorker,
     ready_rx: Receiver<Result<(), AudioError>>,
     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<De
         .find(|d| d.name().is_ok_and(|n| n == name))
 }
 
+/// Pick a sample rate inside both a device's supported `[r_min, r_max]` span and the
+/// backend's drivable `[MIN_DEVICE_RATE, MAX_DEVICE_RATE]` window, preferring
+/// [`SAMPLE_RATE`] when it's reachable and otherwise the nearest in-window bound.
+/// Returns `None` when the device span doesn't overlap the window at all. Pure and
+/// integer-only, so the selection policy is unit-testable (review B5).
+fn bounded_rate(r_min: u32, r_max: u32) -> Option<u32> {
+    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<cpal::SupportedStreamConfig, AudioError> {
     let ranges: Vec<cpal::SupportedStreamConfigRange> = if output {
         device
@@ -474,7 +573,35 @@ fn choose_config(device: &Device, output: bool) -> Result<cpal::SupportedStreamC
         return Ok(r.with_sample_rate(SampleRate(SAMPLE_RATE)));
     }
 
-    // Case 3: no native 48 kHz — fall back to the device default and convert.
+    // Case 3: no native 48 kHz. Before falling back to the device default — which
+    // resolve() rejects outright if its rate is outside the drivable window — look
+    // for a supported config whose rate range overlaps that window and drive it at a
+    // bounded rate, resampling at the boundary (review B5). Prefer the native layout,
+    // then the bounded rate closest to 48 kHz.
+    let pick_bounded = |channels: Option<u16>| -> 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(|_| "<unknown>".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<cpal::SupportedStreamC
     }
     .map_err(|e| AudioError::Device(e.to_string()))?;
     crate::log_msg(&format!(
-        "cpal: device '{}' has no native {SAMPLE_RATE} Hz {} config; using {} Hz / {} ch with linear resampling (W4)",
+        "cpal: device '{}' has no bounded {} config near {SAMPLE_RATE} Hz; falling back to default {} Hz / {} ch (W4)",
         device.name().unwrap_or_else(|_| "<unknown>".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::<Vec<i16>>();