From f52b5ea64e5e811fd2cb32d579969de350177133 Mon Sep 17 00:00:00 2001 From: Mollusk Date: Fri, 19 Jun 2026 16:15:10 -0400 Subject: [PATCH] audio(win): fix RT-safety + start-handshake bugs in the cpal backend Addresses Codex's xhigh RT-audio audit of the new Windows cpal path (review 2026-06-19; all Windows-only, no Linux-path change): - W1 (P1): start_capture/start_playback reported Ok as soon as cpal's play() returned, but cpal's WASAPI play() only QUEUES IAudioClient::Start(); a later Start failure left the UI joined-but-silent. Readiness is now driven by the stream actually proving itself: the first RT data callback sets a started flag (or the error callback sets an error code), and the owner thread waits (bounded by STREAM_START_TIMEOUT) before reporting Ok. - W2: both RT error callbacks ran format!+log_msg on the time-critical stream thread. They now store a category in an AtomicU8 only; the owner / health logger translate + log off the RT path. - W3: the playback ring was published one interleaved sample at a time, letting the RT consumer read a half-written L/R pair and letting a raced fetch_sub wrap ring_fill to usize::MAX (wedging mixer pacing). Now reserves occupancy before publishing and writes the whole frame with a single push_slice. - W6: finish_start did an unbounded recv() while holding the slot mutex, so a wedged driver hung start_* and any concurrent stop. Now recv_timeout with a FINISH_START_TIMEOUT backstop; on timeout it signals + detaches (never joins). - W7: OS-reported device geometry is validated in resolve() (channels>0, rate in 8k-384k) so 0 channels can't panic chunks_exact(0) and a 0/absurd rate can't make an infinite/huge resample ratio. resample.rs constructors also clamp rates >=1 (release-safe; +2 tests) instead of a debug-only assert. - W4 (diagnostic half): the playout-health logger compared raw device samples against the internal-stereo prefill target. The callback now records demand in internal 48 kHz-stereo units (internal_demand) so the comparison is correct for remapped/non-48k devices. The dynamic-target restructure stays deferred. Deferred (logged in review-2026-06-19-cpal-rt-audit.md): W5 (bounded mixer-> worker channel) touches the shared Linux audio path and wants its own design + regression pass; the W2 dynamic-target sizing needs a real WASAPI callback. Verified: Linux cargo test --lib 326/0, clippy --all-targets clean; windows-gnu cargo check --lib --tests --bins clean; windows-gnu release peerspeak.exe builds. Co-Authored-By: Claude Opus 4.8 --- src/audio/cpal_impl.rs | 301 ++++++++++++++++++++++++++++++++--------- src/audio/resample.rs | 45 +++++- 2 files changed, 279 insertions(+), 67 deletions(-) diff --git a/src/audio/cpal_impl.rs b/src/audio/cpal_impl.rs index 4a51156..31676f7 100644 --- a/src/audio/cpal_impl.rs +++ b/src/audio/cpal_impl.rs @@ -51,11 +51,11 @@ //! it is adequate for speech and keeps the matching-rate path bit-exact, with the //! seam ready for a higher-quality resampler later. -use std::sync::atomic::{AtomicBool, AtomicU64, AtomicUsize, Ordering}; +use std::sync::atomic::{AtomicBool, AtomicU8, AtomicU64, AtomicUsize, Ordering}; use std::sync::mpsc::{Receiver, RecvTimeoutError, Sender, channel}; use std::sync::{Arc, Mutex}; use std::thread::{self, JoinHandle}; -use std::time::Duration; +use std::time::{Duration, Instant}; use cpal::traits::{DeviceTrait, HostTrait, StreamTrait}; use cpal::{ @@ -88,6 +88,80 @@ const RING_CAPACITY: usize = 9600 * PLAYBACK_CHANNELS; /// How often a blocked playback worker re-checks its `running` flag, bounding how /// long `stop()` can take to join it (mirrors the PipeWire backend's `WORKER_POLL`). const WORKER_POLL: Duration = Duration::from_millis(100); +/// How long a freshly-played stream has to deliver its first RT callback (proof +/// WASAPI actually `Start()`ed it) before the start is treated as failed. cpal's +/// `play()` only *queues* the WASAPI start, so a queued-but-failed start would +/// otherwise masquerade as success and join the UI into a silent room (review W1). +const STREAM_START_TIMEOUT: Duration = Duration::from_secs(3); +/// Backstop for [`finish_start`]: covers the whole owner setup (resolve + build + +/// play + first-callback wait). Larger than [`STREAM_START_TIMEOUT`] so the owner's +/// own timeout normally fires first with a precise error; this only trips if the +/// owner itself wedges in a driver call before it can report readiness (review W6). +const FINISH_START_TIMEOUT: Duration = Duration::from_secs(6); +/// Lowest / highest device sample rate the backend will drive. The floor bounds +/// the playback pull-resampler's input-pulls-per-output-frame (≈48000/rate) so a +/// pathological low rate can't blow the RT callback's deadline; the ceiling and a +/// nonzero floor also reject the 0 Hz / absurd values a misbehaving driver could +/// report, which would otherwise panic or spin (review W7). +const MIN_DEVICE_RATE: u32 = 8_000; +const MAX_DEVICE_RATE: u32 = 384_000; + +// Stream-error categories carried from the RT error callback to the owner thread +// through an `AtomicU8`, so the callback itself never allocates or logs — both of +// which it previously did via `format!`/`log_msg` on the time-critical stream +// thread (review W2). The owner/logger translates the code off the RT path. +const STREAM_ERR_NONE: u8 = 0; +const STREAM_ERR_DEVICE_UNAVAILABLE: u8 = 1; +const STREAM_ERR_BACKEND: u8 = 2; + +/// Map a cpal stream error to its [`STREAM_ERR_*`](STREAM_ERR_NONE) code. Pure + +/// allocation-free, so it is safe to call from the RT error callback. +fn stream_err_code(e: &cpal::StreamError) -> u8 { + match e { + cpal::StreamError::DeviceNotAvailable => STREAM_ERR_DEVICE_UNAVAILABLE, + _ => STREAM_ERR_BACKEND, + } +} + +/// Human-readable text for a [`STREAM_ERR_*`](STREAM_ERR_NONE) code, logged off the +/// RT path by the owner/health-logger thread. +fn stream_err_text(code: u8) -> &'static str { + match code { + STREAM_ERR_DEVICE_UNAVAILABLE => "audio device became unavailable", + _ => "audio backend stream error", + } +} + +/// Wait for a just-played stream to prove it actually started: its first RT +/// callback sets `started`, or an error callback sets `err_code`. Returns `Ok` on +/// the first callback, `Err` on an error-callback code or [`STREAM_START_TIMEOUT`], +/// or a clean abort if `stop()` cleared `running` mid-start. Polls a couple of +/// cheap atomics on the owner thread — never the RT thread (review W1). +fn wait_for_stream_start( + started: &AtomicBool, + err_code: &AtomicU8, + running: &AtomicBool, +) -> Result<(), AudioError> { + let deadline = Instant::now() + STREAM_START_TIMEOUT; + loop { + if started.load(Ordering::Relaxed) { + return Ok(()); + } + let code = err_code.load(Ordering::Relaxed); + if code != STREAM_ERR_NONE { + return Err(AudioError::Stream(stream_err_text(code).to_string())); + } + if !running.load(Ordering::Relaxed) { + return Err(AudioError::Stream("stream start aborted".to_string())); + } + if Instant::now() >= deadline { + return Err(AudioError::Stream( + "stream did not start within timeout (no WASAPI callback)".to_string(), + )); + } + thread::sleep(Duration::from_millis(5)); + } +} /// Windows audio backend. See module docs. pub struct CpalBackend { @@ -186,25 +260,39 @@ fn finish_start( ready_rx: Receiver>, what: &str, ) -> Result<(), AudioError> { - match ready_rx.recv() { + // Bounded wait. An unbounded `recv()` here would hang `start_*` forever — and + // any concurrent `stop()` behind the same slot mutex — if a WASAPI/driver call + // wedged the worker before it could report (review W6). + match ready_rx.recv_timeout(FINISH_START_TIMEOUT) { 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. + // Setup failed (Err) or the worker disconnected before reporting: 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(_) => { + Err(RecvTimeoutError::Disconnected) => { worker.running.store(false, Ordering::Relaxed); let _ = worker.thread.join(); Err(AudioError::Init(format!( "cpal {what} worker exited before reporting readiness" ))) } + 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); + Err(AudioError::Init(format!( + "cpal {what} did not start within {FINISH_START_TIMEOUT:?}" + ))) + } } } @@ -294,6 +382,22 @@ fn resolve( let supported = choose_config(&device, output)?; let sample_format = supported.sample_format(); let config = supported.config(); + + // Validate the OS-reported geometry before any code divides by it or sizes a + // loop from it (review W7). Zero channels would panic `chunks_exact(0)` / + // `chunks_mut(0)`; a zero or absurd rate would yield an infinite/huge resample + // ratio. Reject up front with a real error instead of panicking or spinning. + if config.channels == 0 { + return Err(AudioError::Device( + "audio device reports zero channels".to_string(), + )); + } + if !(MIN_DEVICE_RATE..=MAX_DEVICE_RATE).contains(&config.sample_rate.0) { + return Err(AudioError::Device(format!( + "audio device sample rate {} Hz is outside the supported {MIN_DEVICE_RATE}–{MAX_DEVICE_RATE} Hz range", + config.sample_rate.0, + ))); + } Ok((device, config, sample_format)) } @@ -383,23 +487,30 @@ fn run_capture( let rb = HeapRb::::new(CAPTURE_RING_CAPACITY); let (producer, mut consumer) = rb.split(); let overrun = Arc::new(AtomicU64::new(0)); + // Stream-liveness signals read by `wait_for_stream_start`: the first RT data + // callback sets `started`; the RT error callback sets `err_code` (it does NOT + // log — that would allocate/syscall on the time-critical thread). See W1/W2. + let started = Arc::new(AtomicBool::new(false)); + let err_code = Arc::new(AtomicU8::new(STREAM_ERR_NONE)); - // Fallible device/stream setup. We report the real error to `start_capture` - // before doing any work, so a join never lands in a silent room. + // Fallible device/stream setup: resolve, build, and *queue* the stream start. let setup = || -> Result<(Stream, String, SampleFormat, usize, u32), AudioError> { let (device, config, sample_format) = resolve(false, target)?; let channels = config.channels as usize; let device_rate = config.sample_rate.0; let stream = match sample_format { - SampleFormat::F32 => { - build_input::(&device, &config, producer, channels, overrun.clone()) - } - SampleFormat::I16 => { - build_input::(&device, &config, producer, channels, overrun.clone()) - } - SampleFormat::U16 => { - build_input::(&device, &config, producer, channels, overrun.clone()) - } + SampleFormat::F32 => build_input::( + &device, &config, producer, channels, overrun.clone(), started.clone(), + err_code.clone(), + ), + SampleFormat::I16 => build_input::( + &device, &config, producer, channels, overrun.clone(), started.clone(), + err_code.clone(), + ), + SampleFormat::U16 => build_input::( + &device, &config, producer, channels, overrun.clone(), started.clone(), + err_code.clone(), + ), other => Err(AudioError::Stream(format!( "unsupported capture sample format: {other:?}" ))), @@ -411,11 +522,21 @@ fn run_capture( Ok((stream, name, sample_format, channels, device_rate)) }; + // Build + queue, then wait for the stream to actually prove it's live before + // reporting readiness. `play()` returning Ok only means WASAPI's `Start()` was + // queued; a later Start failure would otherwise leave us joined-but-silent (W1). let (stream, dev_name, sample_format, channels, device_rate) = match setup() { - Ok(v) => { - let _ = ready.send(Ok(())); - v - } + Ok(v) => match wait_for_stream_start(&started, &err_code, &running) { + Ok(()) => { + let _ = ready.send(Ok(())); + v + } + Err(e) => { + let _ = ready.send(Err(e)); + drop(v.0); // the stream + return; + } + }, Err(e) => { let _ = ready.send(Err(e)); return; @@ -439,6 +560,7 @@ fn run_capture( // frames. Keep `stream` alive until `stop()` flips the flag. let mut acc = FrameAccumulator::new(CAPTURE_FRAME); let mut last_overrun = 0u64; + let mut last_err = STREAM_ERR_NONE; while running.load(Ordering::Relaxed) { let mut drained = false; while let Some(sample) = consumer.try_pop() { @@ -469,6 +591,12 @@ fn run_capture( )); last_overrun = o; } + // Surface a stream error the RT callback flagged (it can't log itself). + let ec = err_code.load(Ordering::Relaxed); + if ec != STREAM_ERR_NONE && ec != last_err { + crate::log_msg(&format!("cpal capture stream error: {}", stream_err_text(ec))); + last_err = ec; + } if !drained { thread::sleep(CAPTURE_POLL); } @@ -476,23 +604,31 @@ fn run_capture( drop(stream); } +#[allow(clippy::too_many_arguments)] fn build_input( device: &Device, config: &StreamConfig, mut producer: P, channels: usize, overrun: Arc, + started: Arc, + err_code: Arc, ) -> Result where T: SizedSample + Send + 'static, i16: FromSample, P: Producer + Send + 'static, { - let err_fn = |e| crate::log_msg(&format!("cpal capture stream error: {e}")); + // RT-safe error callback: record a category in an atomic only. Formatting + + // logging happen on the owner thread (the cpal/WASAPI error callback runs on + // the time-critical stream thread, where alloc/syscall are forbidden — W2). + let err_fn = move |e: cpal::StreamError| err_code.store(stream_err_code(&e), Ordering::Relaxed); device .build_input_stream::( config, move |data: &[T], _| { + // First callback proves WASAPI actually started the stream (W1). + started.store(true, Ordering::Relaxed); // 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) { @@ -595,38 +731,28 @@ fn run_playback( // which would force an underrun every cycle (review W2). The callback only // does a wait-free fetch_max; the health logger reports/warns off the RT path. let max_cb = Arc::new(AtomicUsize::new(0)); + // Stream-liveness signals (see the capture path / W1, W2): first RT callback + // sets `started`; the RT error callback sets `err_code` without logging. + let started = Arc::new(AtomicBool::new(false)); + let err_code = Arc::new(AtomicU8::new(STREAM_ERR_NONE)); - // Fallible device/stream setup; report the real error to `start_playback` - // before any work so a failure surfaces instead of a silent room. `consumer` - // is moved into the output callback here. + // Fallible device/stream setup. `consumer` is moved into the output callback. let setup = || -> Result<(Stream, String, SampleFormat, usize, u32), AudioError> { let (device, config, sample_format) = resolve(true, target)?; let channels = config.channels as usize; let device_rate = config.sample_rate.0; let stream = match sample_format { SampleFormat::F32 => build_output::( - &device, - &config, - consumer, - ring_fill.clone(), - underrun.clone(), - max_cb.clone(), + &device, &config, consumer, ring_fill.clone(), underrun.clone(), + max_cb.clone(), started.clone(), err_code.clone(), ), SampleFormat::I16 => build_output::( - &device, - &config, - consumer, - ring_fill.clone(), - underrun.clone(), - max_cb.clone(), + &device, &config, consumer, ring_fill.clone(), underrun.clone(), + max_cb.clone(), started.clone(), err_code.clone(), ), SampleFormat::U16 => build_output::( - &device, - &config, - consumer, - ring_fill.clone(), - underrun.clone(), - max_cb.clone(), + &device, &config, consumer, ring_fill.clone(), underrun.clone(), + max_cb.clone(), started.clone(), err_code.clone(), ), other => Err(AudioError::Stream(format!( "unsupported playback sample format: {other:?}" @@ -639,11 +765,19 @@ fn run_playback( Ok((stream, name, sample_format, channels, device_rate)) }; + // Build + queue, then wait for a real callback before reporting readiness (W1). let (stream, dev_name, sample_format, channels, device_rate) = match setup() { - Ok(v) => { - let _ = ready.send(Ok(())); - v - } + Ok(v) => match wait_for_stream_start(&started, &err_code, &running) { + Ok(()) => { + let _ = ready.send(Ok(())); + v + } + Err(e) => { + let _ = ready.send(Err(e)); + drop(v.0); // the stream + return; + } + }, Err(e) => { let _ = ready.send(Err(e)); return; @@ -659,6 +793,7 @@ fn run_playback( underrun.clone(), dropped.clone(), max_cb.clone(), + err_code.clone(), ); // Feed the ring from the network mixer until `stop()` flips `running` or the @@ -669,10 +804,20 @@ fn run_playback( dropped.fetch_add(1, Ordering::Relaxed); return; } - for &sample in &frame { - let _ = producer.try_push(sample); - } + // Reserve occupancy BEFORE publishing samples, and publish the whole frame + // in one `push_slice` (review W3). Per-sample pushes let the RT consumer + // observe a half-written stereo pair (L without R) → channel tear, and a + // pop that raced the post-loop `fetch_add` could drive `ring_fill` below + // zero and wrap it to usize::MAX, wedging the mixer's pacing. Reserving + // first means the consumer can never pop a sample that isn't yet counted. ring_fill.fetch_add(frame.len(), Ordering::Relaxed); + let pushed = producer.push_slice(&frame); + if pushed != frame.len() { + // The capacity check above should make this unreachable (the consumer + // only drains), but stay exact if it ever isn't. + ring_fill.fetch_sub(frame.len() - pushed, Ordering::Relaxed); + dropped.fetch_add(1, Ordering::Relaxed); + } }); // We're shutting down (either stop() or disconnect). Ensure the logger sees it @@ -682,6 +827,7 @@ fn run_playback( drop(stream); } +#[allow(clippy::too_many_arguments)] fn build_output( device: &Device, config: &StreamConfig, @@ -689,12 +835,15 @@ fn build_output( ring_fill: Arc, underrun: Arc, max_cb: Arc, + started: Arc, + err_code: Arc, ) -> Result where T: SizedSample + FromSample + Send + 'static, C: Consumer + Send + 'static, { - let err_fn = |e| crate::log_msg(&format!("cpal playback stream error: {e}")); + // RT-safe error callback: atomic store only, no alloc/log (review W2). + let err_fn = move |e: cpal::StreamError| err_code.store(stream_err_code(&e), Ordering::Relaxed); let device_rate = config.sample_rate.0; let device_channels = config.channels as usize; if device_rate == SAMPLE_RATE && device_channels == PLAYBACK_CHANNELS { @@ -702,8 +851,14 @@ where .build_output_stream::( config, move |data: &mut [T], _| { - // Wait-free; the logger thread reads this off the RT path. - max_cb.fetch_max(data.len(), Ordering::Relaxed); + started.store(true, Ordering::Relaxed); + // Record demand in INTERNAL 48 kHz-stereo samples (not raw + // device samples) so the health logger's prefill-target + // comparison is apples-to-apples for any rate/layout (W4). + max_cb.fetch_max( + internal_demand(data.len(), device_channels, device_rate), + Ordering::Relaxed, + ); let (popped, starved) = fill_output(&mut consumer, data); if starved > 0 { underrun.fetch_add(starved, Ordering::Relaxed); @@ -725,8 +880,11 @@ where .build_output_stream::( config, move |data: &mut [T], _| { - // Wait-free; the logger thread reads this off the RT path. - max_cb.fetch_max(data.len(), Ordering::Relaxed); + started.store(true, Ordering::Relaxed); + max_cb.fetch_max( + internal_demand(data.len(), device_channels, device_rate), + Ordering::Relaxed, + ); let (popped, starved) = fill_output_remap(&mut consumer, data, device_channels, &mut resampler); if starved > 0 { @@ -746,6 +904,18 @@ where } } +/// Convert an output callback's raw device-sample length into the equivalent +/// internal 48 kHz-stereo sample demand, so the prefill-target comparison stays +/// meaningful regardless of the device's rate/channel layout (W4 diagnostic fix). +/// Integer-only and allocation-free, so it is safe on the RT callback thread. +#[inline] +fn internal_demand(device_len: usize, device_channels: usize, device_rate: u32) -> usize { + let device_frames = device_len / device_channels.max(1); + let need_frames = + (device_frames as u64 * SAMPLE_RATE as u64).div_ceil(device_rate.max(1) as u64) as usize; + need_frames * PLAYBACK_CHANNELS +} + /// Drain the ring into the device buffer, substituting silence on underrun. /// Returns `(samples_popped, samples_starved)`. RT-safe (wait-free `try_pop`). fn fill_output(consumer: &mut C, out: &mut [T]) -> (usize, u64) @@ -832,11 +1002,13 @@ fn spawn_health_logger( underrun: Arc, dropped: Arc, max_cb: Arc, + err_code: Arc, ) -> JoinHandle<()> { let verbose = std::env::var_os("PEERSPEAK_AUDIO_VERBOSE").is_some(); thread::spawn(move || { let (mut last_u, mut last_d) = (0u64, 0u64); let mut reported_cb = 0usize; + let mut last_err = STREAM_ERR_NONE; while running.load(Ordering::Relaxed) { thread::sleep(Duration::from_secs(1)); let u = underrun.load(Ordering::Relaxed); @@ -851,21 +1023,28 @@ fn spawn_health_logger( fill / (48 * PLAYBACK_CHANNELS), )); } - // Report the device's callback size the first time it's seen (and on - // any new high). If a callback asks for more than the prefill target, - // the ring can't satisfy it and underruns every cycle — the W2 bug - // signature; warn so a real-host log shows whether it's biting. + // Surface a stream error the RT callback flagged (it can't log itself). + let ec = err_code.load(Ordering::Relaxed); + if ec != STREAM_ERR_NONE && ec != last_err { + crate::log_msg(&format!("cpal playback stream error: {}", stream_err_text(ec))); + last_err = ec; + } + // Report the device's per-cycle demand (in internal 48 kHz-stereo + // samples) the first time it's seen, and on any new high. If a callback + // demands more than the prefill target, the ring can't satisfy it and + // underruns every cycle — the W2 signature; warn so a real-host log + // shows whether it's biting. let cb = max_cb.load(Ordering::Relaxed); if cb > reported_cb { reported_cb = cb; let ms = cb / (48 * PLAYBACK_CHANNELS); if cb > PLAYBACK_TARGET_SAMPLES { crate::log_msg(&format!( - "cpal output callback up to {cb} samples/cycle (~{ms}ms) EXCEEDS prefill target {PLAYBACK_TARGET_SAMPLES} — expect periodic underruns; needs a larger target or a fixed buffer size (review W2)", + "cpal output callback demands up to {cb} internal samples/cycle (~{ms}ms) EXCEEDS prefill target {PLAYBACK_TARGET_SAMPLES} — expect periodic underruns; needs a larger target or a fixed buffer size (review W2)", )); } else if verbose { crate::log_msg(&format!( - "cpal output callback up to {cb} samples/cycle (~{ms}ms), target {PLAYBACK_TARGET_SAMPLES}", + "cpal output callback demands up to {cb} internal samples/cycle (~{ms}ms), target {PLAYBACK_TARGET_SAMPLES}", )); } } diff --git a/src/audio/resample.rs b/src/audio/resample.rs index 4560a67..4f5723f 100644 --- a/src/audio/resample.rs +++ b/src/audio/resample.rs @@ -45,11 +45,14 @@ pub struct PushResampler { } impl PushResampler { - /// Build a resampler from `in_rate` to `out_rate` (both in Hz, must be > 0). + /// Build a resampler from `in_rate` to `out_rate` (both in Hz). Rates are + /// clamped to `>= 1` so `step` is always finite and non-zero: a zero `step` + /// would make [`push`](Self::push)'s `while self.next < 1.0` loop forever. The + /// cpal backend's `resolve()` also rejects such rates up front, so this is + /// belt-and-suspenders against a future caller (review W7). pub fn new(in_rate: u32, out_rate: u32) -> Self { - debug_assert!(in_rate > 0 && out_rate > 0); Self { - step: in_rate as f64 / out_rate as f64, + step: in_rate.max(1) as f64 / out_rate.max(1) as f64, next: 0.0, prev: 0.0, started: false, @@ -108,11 +111,12 @@ pub struct StereoPullResampler { } impl StereoPullResampler { - /// Build a resampler from `in_rate` to `out_rate` (both in Hz, must be > 0). + /// Build a resampler from `in_rate` to `out_rate` (both in Hz). Rates are + /// clamped to `>= 1` so `step` is finite and non-zero — otherwise + /// [`next`](Self::next)'s `while self.frac >= 1.0` could spin (review W7). pub fn new(in_rate: u32, out_rate: u32) -> Self { - debug_assert!(in_rate > 0 && out_rate > 0); Self { - step: in_rate as f64 / out_rate as f64, + step: in_rate.max(1) as f64 / out_rate.max(1) as f64, frac: 0.0, prev: (0.0, 0.0), cur: (0.0, 0.0), @@ -271,4 +275,33 @@ mod tests { // At step 2.0 we consume ~2 input frames per output frame. assert!(idx > emitted, "consumed {idx} input, emitted {emitted} output"); } + + /// A zero rate must not produce a zero `step` (which would spin `push`'s inner + /// `while self.next < 1.0` forever). Clamping makes the call terminate (W7). + #[test] + fn push_zero_rate_does_not_spin() { + let mut r = PushResampler::new(0, 48_000); + let mut count = 0usize; + // Feed two samples; with a clamped non-zero step this returns promptly. + r.push(0.0, |_| count += 1); + r.push(1.0, |_| count += 1); + // Reaching here at all is the assertion (no hang); some output is produced. + assert!(count >= 1); + } + + /// A zero output rate must not make the pull resampler's segment-advance loop + /// spin. Clamping keeps `step` finite so `next` terminates (W7). + #[test] + fn pull_zero_out_rate_does_not_spin() { + let mut r = StereoPullResampler::new(48_000, 0); + let frames = [(0.0, 0.0), (1.0, 1.0), (2.0, 2.0)]; + let mut idx = 0; + let got = r.next(|| { + let v = frames.get(idx).copied(); + idx += 1; + v + }); + // Terminates and yields the primed frame instead of hanging. + assert!(got.is_some()); + } }