audio(win): land the deferred cpal start-resilience items (B3 + B5)

Closes the two Windows-only follow-ups Codex deferred in the RT-audit
re-review (review-2026-06-19-cpal-rt-audit.md). Both are cfg(windows),
so they carry zero risk to the shared Linux audio path.

B3 — orphan-thread tombstone on a wedged start. On the FINISH_START_TIMEOUT
path the owner thread is detached (not joined) so start_*/stop can't hang;
previously the slot was left empty, so a retry against a permanently wedged
device spawned ANOTHER orphan worker holding its own COM/device handle, and
so on without bound. The slot is now a SlotState { Idle | Live | Wedged }:

- Each worker carries an `exited: Arc<AtomicBool>` flipped true by an
  ExitGuard at the top of the thread body — fires on normal return, panic
  unwind, or whenever the wedged driver call finally releases the thread.
- A timed-out start detaches its thread and leaves a `Wedged { exited }`
  tombstone instead of an empty slot.
- `ensure_idle` (pure, unit-tested) rejects new starts while the orphan is
  still alive, but clears the tombstone once `exited` flips, so the slot
  becomes reusable after the device recovers. `stop` restores a still-live
  tombstone rather than silently clearing it.

B5 — choose_config picks a bounded supported rate before the device default.
A device whose default rate is outside the drivable 8k–384k window but which
also exposes a usable in-window config was previously rejected by resolve().
New case 3 scans the supported config ranges for one overlapping the window
and drives it at a `bounded_rate` (48 kHz when reachable, else the nearest
in-window bound), preferring the native layout; the device default is now a
last resort. `bounded_rate` is pure and unit-tested.

6 new unit tests (bounded_rate x4, ensure_idle x2) — they're in the
cfg(windows) module, so they compile/run under the windows-gnu target, not
the Linux lib suite.

Verified: Linux cargo test --lib 326/0 + clippy --lib --tests clean (shared
paths untouched); windows-gnu cargo check --release --lib --tests --bins
clean, no warnings.

Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>
This commit is contained in:
2026-06-19 16:48:07 -04:00
co-authored by Claude Opus 4.8
parent 8e0b4c16ec
commit 306bc295b1
+214 -30
View File
@@ -184,21 +184,72 @@ fn wait_for_stream_start(
/// Windows audio backend. See module docs. /// Windows audio backend. See module docs.
pub struct CpalBackend { pub struct CpalBackend {
capture: Mutex<Option<StreamWorker>>, capture: Mutex<SlotState>,
playback: Mutex<Option<StreamWorker>>, 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 { struct StreamWorker {
running: Arc<AtomicBool>, running: Arc<AtomicBool>,
exited: Arc<AtomicBool>,
thread: JoinHandle<()>, 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 { impl CpalBackend {
pub fn new() -> Self { pub fn new() -> Self {
Self { Self {
capture: Mutex::new(None), capture: Mutex::new(SlotState::Idle),
playback: Mutex::new(None), playback: Mutex::new(SlotState::Idle),
} }
} }
} }
@@ -215,20 +266,30 @@ impl AudioBackend for CpalBackend {
tx: Sender<Vec<i16>>, tx: Sender<Vec<i16>>,
target_node: Option<String>, target_node: Option<String>,
) -> Result<(), AudioError> { ) -> Result<(), AudioError> {
let guard = self.capture.lock().unwrap(); let mut guard = self.capture.lock().unwrap();
if guard.is_some() { ensure_idle(&mut guard, "capture")?;
return Err(AudioError::Stream("Capture already started".to_string()));
}
let running = Arc::new(AtomicBool::new(true)); let running = Arc::new(AtomicBool::new(true));
let exited = Arc::new(AtomicBool::new(false));
let running_thread = running.clone(); let running_thread = running.clone();
let exited_thread = exited.clone();
let (ready_tx, ready_rx) = channel::<Result<(), AudioError>>(); let (ready_tx, ready_rx) = channel::<Result<(), AudioError>>();
let thread = thread::Builder::new() let thread = thread::Builder::new()
.name("peerspeak-cpal-capture".to_string()) .name("peerspeak-cpal-capture".to_string())
.spawn(move || { .spawn(move || {
let _exit = ExitGuard(exited_thread);
run_capture(tx, target_node, running_thread, ready_tx); run_capture(tx, target_node, running_thread, ready_tx);
}) })
.map_err(|e| AudioError::Init(e.to_string()))?; .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( fn start_playback(
@@ -237,22 +298,27 @@ impl AudioBackend for CpalBackend {
target_node: Option<String>, target_node: Option<String>,
ring_fill: Arc<AtomicUsize>, ring_fill: Arc<AtomicUsize>,
) -> Result<(), AudioError> { ) -> Result<(), AudioError> {
let guard = self.playback.lock().unwrap(); let mut guard = self.playback.lock().unwrap();
if guard.is_some() { ensure_idle(&mut guard, "playback")?;
return Err(AudioError::Stream("Playback already started".to_string()));
}
let running = Arc::new(AtomicBool::new(true)); let running = Arc::new(AtomicBool::new(true));
let exited = Arc::new(AtomicBool::new(false));
let running_thread = running.clone(); let running_thread = running.clone();
let exited_thread = exited.clone();
let (ready_tx, ready_rx) = channel::<Result<(), AudioError>>(); let (ready_tx, ready_rx) = channel::<Result<(), AudioError>>();
let thread = thread::Builder::new() let thread = thread::Builder::new()
.name("peerspeak-cpal-playback".to_string()) .name("peerspeak-cpal-playback".to_string())
.spawn(move || { .spawn(move || {
let _exit = ExitGuard(exited_thread);
run_playback(rx, target_node, ring_fill, running_thread, ready_tx); run_playback(rx, target_node, ring_fill, running_thread, ready_tx);
}) })
.map_err(|e| AudioError::Init(e.to_string()))?; .map_err(|e| AudioError::Init(e.to_string()))?;
finish_start( finish_start(
guard, guard,
StreamWorker { running, thread }, StreamWorker {
running,
exited,
thread,
},
ready_rx, ready_rx,
"playback", "playback",
) )
@@ -260,9 +326,21 @@ impl AudioBackend for CpalBackend {
fn stop(&self) -> Result<(), AudioError> { fn stop(&self) -> Result<(), AudioError> {
for slot in [&self.capture, &self.playback] { for slot in [&self.capture, &self.playback] {
if let Some(worker) = slot.lock().unwrap().take() { let mut guard = slot.lock().unwrap();
worker.running.store(false, Ordering::Relaxed); match std::mem::replace(&mut *guard, SlotState::Idle) {
let _ = worker.thread.join(); 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(()) Ok(())
@@ -274,7 +352,7 @@ impl AudioBackend for CpalBackend {
/// real error. This is what makes `start_capture`/`start_playback` fail loudly /// 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). /// instead of returning `Ok` into a joined-but-silent room (Codex review W1).
fn finish_start( fn finish_start(
mut guard: std::sync::MutexGuard<'_, Option<StreamWorker>>, mut guard: std::sync::MutexGuard<'_, SlotState>,
worker: StreamWorker, worker: StreamWorker,
ready_rx: Receiver<Result<(), AudioError>>, ready_rx: Receiver<Result<(), AudioError>>,
what: &str, what: &str,
@@ -284,7 +362,7 @@ fn finish_start(
// wedged the worker before it could report (review W6). // wedged the worker before it could report (review W6).
match ready_rx.recv_timeout(FINISH_START_TIMEOUT) { match ready_rx.recv_timeout(FINISH_START_TIMEOUT) {
Ok(Ok(())) => { Ok(Ok(())) => {
*guard = Some(worker); *guard = SlotState::Live(worker);
Ok(()) Ok(())
} }
// Setup failed (Err) or the worker disconnected before reporting: either // Setup failed (Err) or the worker disconnected before reporting: either
@@ -304,10 +382,17 @@ fn finish_start(
Err(RecvTimeoutError::Timeout) => { Err(RecvTimeoutError::Timeout) => {
// The worker is wedged in a driver call. Signal it to exit, but DETACH // 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 // rather than join — joining would re-introduce the unbounded hang this
// timeout exists to prevent. The thread unwinds on its own if/when the // timeout exists to prevent. Leave a Wedged tombstone so subsequent
// driver call ever returns. // starts are rejected until the orphan's ExitGuard flips `exited`, rather
worker.running.store(false, Ordering::Relaxed); // than spawning more orphan threads against the same dead device (B3).
drop(worker.thread); let StreamWorker {
running,
exited,
thread,
} = worker;
running.store(false, Ordering::Relaxed);
drop(thread);
*guard = SlotState::Wedged { exited };
Err(AudioError::Init(format!( Err(AudioError::Init(format!(
"cpal {what} did not start within {FINISH_START_TIMEOUT:?}" "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)) .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: /// Pick a stream config. Preference order, best (no conversion) first:
/// 1. exactly [`SAMPLE_RATE`] at the preferred layout (stereo out / mono in), /// 1. exactly [`SAMPLE_RATE`] at the preferred layout (stereo out / mono in),
/// 2. exactly [`SAMPLE_RATE`] at any channel count (rate-exact, backend remaps), /// 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 /// Cases 34 incur resampling; the backend reads the returned config's rate and
/// channel count and converts at the boundary (W4). A device that exposes no config /// channel count and converts at the boundary (W4). Case 3 (review B5) is what keeps
/// at all is still a hard error. /// 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> { fn choose_config(device: &Device, output: bool) -> Result<cpal::SupportedStreamConfig, AudioError> {
let ranges: Vec<cpal::SupportedStreamConfigRange> = if output { let ranges: Vec<cpal::SupportedStreamConfigRange> = if output {
device device
@@ -474,7 +573,35 @@ fn choose_config(device: &Device, output: bool) -> Result<cpal::SupportedStreamC
return Ok(r.with_sample_rate(SampleRate(SAMPLE_RATE))); 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 { let def = if output {
device.default_output_config() device.default_output_config()
} else { } else {
@@ -482,7 +609,7 @@ fn choose_config(device: &Device, output: bool) -> Result<cpal::SupportedStreamC
} }
.map_err(|e| AudioError::Device(e.to_string()))?; .map_err(|e| AudioError::Device(e.to_string()))?;
crate::log_msg(&format!( 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()), device.name().unwrap_or_else(|_| "<unknown>".to_string()),
if output { "output" } else { "input" }, if output { "output" } else { "input" },
def.sample_rate().0, def.sample_rate().0,
@@ -1201,6 +1328,63 @@ mod tests {
drain_loop(&rx, &running, |_| panic!("no frame should arrive")); 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] #[test]
fn drain_loop_delivers_frames() { fn drain_loop_delivers_frames() {
let (tx, rx) = mpsc::channel::<Vec<i16>>(); let (tx, rx) = mpsc::channel::<Vec<i16>>();