Complete Windows audio remap path

docs/WINDOWS.md

@@ -68,9 +68,9 @@ connections are expected and valid.
 | Echo cancellation | Linux-only PipeWire feature. The Windows UI shows it disabled as unavailable. |
 | Screen share | Requires a Windows `pixelpass.exe` on `PATH` or a configured override. |
 | Chimes | Now routed through Windows `SoundPlayer`; needs a real Windows host to audibly verify. |
-| Resampling/device format | Open. Devices must support 48 kHz, and output must support stereo; a 44.1 kHz-only/default device currently errors instead of playing. |
+| Resampling/device format | Cross-compiled. cpal/WASAPI now chooses native 48 kHz when available and otherwise resamples/remaps at the device boundary; needs real Windows hardware audio verification. |
 | Device persistence | Open. WASAPI friendly names may duplicate or change across driver/profile changes. |
-| Playback pacing | Open. The fixed playback target under WASAPI shared mode still needs real-hardware verification. |
+| Playback pacing | Cross-compiled. The fixed playback target under WASAPI shared mode still needs real-hardware verification with `audio_probe`. |
 
 Before calling Windows support done, verify a real Windows machine can create/join a room,
 capture mic audio, hear remote audio, select devices, restart with selections preserved, and

src/audio/cpal_impl.rs

@@ -52,16 +52,18 @@
 //! seam ready for a higher-quality resampler later.
 
 use std::sync::atomic::{AtomicBool, AtomicU64, AtomicUsize, Ordering};
-use std::sync::mpsc::{channel, Receiver, RecvTimeoutError, Sender};
+use std::sync::mpsc::{Receiver, RecvTimeoutError, Sender, channel};
 use std::sync::{Arc, Mutex};
 use std::thread::{self, JoinHandle};
 use std::time::Duration;
 
 use cpal::traits::{DeviceTrait, HostTrait, StreamTrait};
-use cpal::{Device, FromSample, Sample, SampleFormat, SampleRate, SizedSample, Stream, StreamConfig};
+use cpal::{
+    Device, FromSample, Sample, SampleFormat, SampleRate, SizedSample, Stream, StreamConfig,
+};
 use ringbuf::{
-    traits::{Consumer, Producer, Split},
     HeapRb,
+    traits::{Consumer, Producer, Split},
 };
 
 use super::resample::{PushResampler, StereoPullResampler};
@@ -155,7 +157,12 @@ impl AudioBackend for CpalBackend {
                 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 }, ready_rx, "playback")
+        finish_start(
+            guard,
+            StreamWorker { running, thread },
+            ready_rx,
+            "playback",
+        )
     }
 
     fn stop(&self) -> Result<(), AudioError> {
@@ -296,7 +303,9 @@ fn find_device_by_name(host: &cpal::Host, output: bool, name: &str) -> Option<De
     } else {
         host.input_devices().ok()?
     };
-    devices.into_iter().find(|d| d.name().is_ok_and(|n| n == name))
+    devices
+        .into_iter()
+        .find(|d| d.name().is_ok_and(|n| n == name))
 }
 
 /// Pick a stream config. Preference order, best (no conversion) first:
@@ -307,10 +316,7 @@ fn find_device_by_name(host: &cpal::Host, output: bool, name: &str) -> Option<De
 /// 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.
-fn choose_config(
+fn choose_config(device: &Device, output: bool) -> Result<cpal::SupportedStreamConfig, AudioError> {
-    device: &Device,
-    output: bool,
-) -> Result<cpal::SupportedStreamConfig, AudioError> {
     let ranges: Vec<cpal::SupportedStreamConfigRange> = if output {
         device
             .supported_output_configs()
@@ -398,7 +404,9 @@ fn run_capture(
                 "unsupported capture sample format: {other:?}"
             ))),
         }?;
-        stream.play().map_err(|e| AudioError::Stream(e.to_string()))?;
+        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, device_rate))
     };
@@ -420,7 +428,8 @@ fn run_capture(
     // If the device isn't at 48 kHz, resample its mono stream up/down to 48 kHz on
     // this (non-RT) thread before framing (W4). At 48 kHz this stays None and the
     // samples pass straight through, bit-exact.
-    let mut resampler = (device_rate != SAMPLE_RATE).then(|| PushResampler::new(device_rate, SAMPLE_RATE));
+    let mut resampler =
+        (device_rate != SAMPLE_RATE).then(|| PushResampler::new(device_rate, SAMPLE_RATE));
     // Reused scratch for a sample's resampled output (off-RT alloc; tiny — at most
     // a couple of samples per input). Avoids a nested-closure borrow over `acc`/`tx`.
     let mut resampled: Vec<i16> = Vec::new();
@@ -590,28 +599,47 @@ fn run_playback(
     // 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.
-    let setup = || -> Result<(Stream, String, SampleFormat), AudioError> {
+    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 => {
+            SampleFormat::F32 => build_output::<f32, _>(
-                build_output::<f32, _>(&device, &config, consumer, ring_fill.clone(), underrun.clone(), max_cb.clone())
+                &device,
-            }
+                &config,
-            SampleFormat::I16 => {
+                consumer,
-                build_output::<i16, _>(&device, &config, consumer, ring_fill.clone(), underrun.clone(), max_cb.clone())
+                ring_fill.clone(),
-            }
+                underrun.clone(),
-            SampleFormat::U16 => {
+                max_cb.clone(),
-                build_output::<u16, _>(&device, &config, consumer, ring_fill.clone(), underrun.clone(), max_cb.clone())
+            ),
-            }
+            SampleFormat::I16 => build_output::<i16, _>(
+                &device,
+                &config,
+                consumer,
+                ring_fill.clone(),
+                underrun.clone(),
+                max_cb.clone(),
+            ),
+            SampleFormat::U16 => build_output::<u16, _>(
+                &device,
+                &config,
+                consumer,
+                ring_fill.clone(),
+                underrun.clone(),
+                max_cb.clone(),
+            ),
             other => Err(AudioError::Stream(format!(
                 "unsupported playback sample format: {other:?}"
             ))),
         }?;
-        stream.play().map_err(|e| AudioError::Stream(e.to_string()))?;
+        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))
+        Ok((stream, name, sample_format, channels, device_rate))
     };
 
-    let (stream, dev_name, sample_format) = match setup() {
+    let (stream, dev_name, sample_format, channels, device_rate) = match setup() {
         Ok(v) => {
             let _ = ready.send(Ok(()));
             v
@@ -622,7 +650,7 @@ fn run_playback(
         }
     };
     crate::log_msg(&format!(
-        "cpal playback started: device='{dev_name}' format={sample_format:?} channels={PLAYBACK_CHANNELS} rate={SAMPLE_RATE} Hz"
+        "cpal playback started: device='{dev_name}' format={sample_format:?} channels={channels} device_rate={device_rate} Hz <- {SAMPLE_RATE} Hz"
     ));
 
     let logger = spawn_health_logger(
@@ -667,27 +695,55 @@ where
     C: Consumer<Item = i16> + Send + 'static,
 {
     let err_fn = |e| crate::log_msg(&format!("cpal playback stream error: {e}"));
-    device
+    let device_rate = config.sample_rate.0;
-        .build_output_stream::<T, _, _>(
+    let device_channels = config.channels as usize;
-            config,
+    if device_rate == SAMPLE_RATE && device_channels == PLAYBACK_CHANNELS {
-            move |data: &mut [T], _| {
+        device
-                // Wait-free; the logger thread reads this off the RT path.
+            .build_output_stream::<T, _, _>(
-                max_cb.fetch_max(data.len(), Ordering::Relaxed);
+                config,
-                let (popped, starved) = fill_output(&mut consumer, data);
+                move |data: &mut [T], _| {
-                if starved > 0 {
+                    // Wait-free; the logger thread reads this off the RT path.
-                    underrun.fetch_add(starved, Ordering::Relaxed);
+                    max_cb.fetch_max(data.len(), Ordering::Relaxed);
-                }
+                    let (popped, starved) = fill_output(&mut consumer, data);
-                if popped > 0 {
+                    if starved > 0 {
-                    // Decrement the exact occupancy by what we actually pulled
+                        underrun.fetch_add(starved, Ordering::Relaxed);
-                    // (underruns removed nothing) so the mixer paces against the
+                    }
-                    // true ring depth.
+                    if popped > 0 {
-                    ring_fill.fetch_sub(popped, Ordering::Relaxed);
+                        // Decrement the exact occupancy by what we actually pulled
-                }
+                        // (underruns removed nothing) so the mixer paces against the
-            },
+                        // true ring depth.
-            err_fn,
+                        ring_fill.fetch_sub(popped, Ordering::Relaxed);
-            None,
+                    }
-        )
+                },
-        .map_err(|e| AudioError::Stream(e.to_string()))
+                err_fn,
+                None,
+            )
+            .map_err(|e| AudioError::Stream(e.to_string()))
+    } else {
+        let mut resampler = StereoPullResampler::new(SAMPLE_RATE, device_rate);
+        device
+            .build_output_stream::<T, _, _>(
+                config,
+                move |data: &mut [T], _| {
+                    // Wait-free; the logger thread reads this off the RT path.
+                    max_cb.fetch_max(data.len(), Ordering::Relaxed);
+                    let (popped, starved) =
+                        fill_output_remap(&mut consumer, data, device_channels, &mut resampler);
+                    if starved > 0 {
+                        underrun.fetch_add(starved, Ordering::Relaxed);
+                    }
+                    if popped > 0 {
+                        // Decrement the exact occupancy by what we actually pulled
+                        // (underruns removed nothing) so the mixer paces against the
+                        // true ring depth.
+                        ring_fill.fetch_sub(popped, Ordering::Relaxed);
+                    }
+                },
+                err_fn,
+                None,
+            )
+            .map_err(|e| AudioError::Stream(e.to_string()))
+    }
 }
 
 /// Drain the ring into the device buffer, substituting silence on underrun.
@@ -714,6 +770,60 @@ where
     (popped, starved)
 }
 
+/// Resample/remap internal 48 kHz stereo ring samples into the device buffer.
+/// Returns `(internal_samples_popped, device_samples_starved)`. RT-safe.
+fn fill_output_remap<T, C>(
+    consumer: &mut C,
+    out: &mut [T],
+    device_channels: usize,
+    resampler: &mut StereoPullResampler,
+) -> (usize, u64)
+where
+    T: Sample + FromSample<i16>,
+    C: Consumer<Item = i16>,
+{
+    let mut popped = 0usize;
+    let mut starved = 0u64;
+    for frame in out.chunks_mut(device_channels) {
+        match resampler.next(|| {
+            let l = match consumer.try_pop() {
+                Some(v) => {
+                    popped += 1;
+                    v
+                }
+                None => return None,
+            };
+            let r = match consumer.try_pop() {
+                Some(v) => {
+                    popped += 1;
+                    v
+                }
+                None => return None,
+            };
+            Some((i16_to_f32(l), i16_to_f32(r)))
+        }) {
+            Some((l, r)) => {
+                if device_channels == 1 {
+                    frame[0] = T::from_sample(f32_to_i16((l + r) * 0.5));
+                } else {
+                    frame[0] = T::from_sample(f32_to_i16(l));
+                    frame[1] = T::from_sample(f32_to_i16(r));
+                    for slot in &mut frame[2..] {
+                        *slot = T::from_sample(0i16);
+                    }
+                }
+            }
+            None => {
+                for slot in frame {
+                    *slot = T::from_sample(0i16);
+                }
+                starved += device_channels as u64;
+            }
+        }
+    }
+    (popped, starved)
+}
+
 /// Once-per-second playout-health line (mirrors the PipeWire backend). Quiet
 /// unless a second actually glitched, or `PEERSPEAK_AUDIO_VERBOSE` is set.
 fn spawn_health_logger(
@@ -819,6 +929,48 @@ mod tests {
         assert_eq!(out, [1, 2, 3, 0, 0]);
     }
 
+    #[test]
+    fn fill_output_remap_downmixes_to_mono() {
+        let rb = HeapRb::<i16>::new(8);
+        let (mut prod, mut cons) = rb.split();
+        for v in [100, 300, 500, -100, 7, 9] {
+            prod.try_push(v).unwrap();
+        }
+        let mut out = [0i16; 2];
+        let mut resampler = StereoPullResampler::new(SAMPLE_RATE, SAMPLE_RATE);
+        let (popped, starved) = fill_output_remap(&mut cons, &mut out, 1, &mut resampler);
+        assert_eq!(popped, 6);
+        assert_eq!(starved, 0);
+        assert_eq!(out, [200, 200]);
+    }
+
+    #[test]
+    fn fill_output_remap_silences_underrun() {
+        let rb = HeapRb::<i16>::new(8);
+        let (_prod, mut cons) = rb.split();
+        let mut out = [11i16; 4];
+        let mut resampler = StereoPullResampler::new(SAMPLE_RATE, SAMPLE_RATE);
+        let (popped, starved) = fill_output_remap(&mut cons, &mut out, 2, &mut resampler);
+        assert_eq!(popped, 0);
+        assert_eq!(starved, out.len() as u64);
+        assert_eq!(out, [0, 0, 0, 0]);
+    }
+
+    #[test]
+    fn fill_output_remap_copies_stereo_at_matching_rate() {
+        let rb = HeapRb::<i16>::new(8);
+        let (mut prod, mut cons) = rb.split();
+        for v in [1, -1, 2, -2, 3, -3] {
+            prod.try_push(v).unwrap();
+        }
+        let mut out = [0i16; 4];
+        let mut resampler = StereoPullResampler::new(SAMPLE_RATE, SAMPLE_RATE);
+        let (popped, starved) = fill_output_remap(&mut cons, &mut out, 2, &mut resampler);
+        assert_eq!(popped, 6);
+        assert_eq!(starved, 0);
+        assert_eq!(out, [1, -1, 2, -2]);
+    }
+
     #[test]
     fn drain_loop_exits_when_running_flips_even_with_sender_alive() {
         let (tx, rx) = mpsc::channel::<Vec<i16>>();

src/bin/audio_probe.rs

@@ -1,11 +1,11 @@
 //! Audio playout diagnostic probe.
 //!
-//! Drives a phase-continuous sine tone through the *real* PipeWire playback path
+//! Drives a phase-continuous sine tone through the *real* playback path
-//! (`PipeWireBackend::start_playback`), using the *same* fill-paced production
+//! (PipeWire on Linux, cpal/WASAPI on Windows), using the *same* fill-paced
-//! the production mixer uses (`core/mod.rs`): generate a frame only while the
+//! production the production mixer uses (`core/mod.rs`): generate a frame only while the
 //! playback ring is below `PLAYBACK_TARGET_SAMPLES`, so production tracks the
-//! PipeWire hardware clock. No network, no microphone — this isolates the local
+//! hardware clock. No network, no microphone — this isolates the local output
-//! output path so we can confirm the clock-paced playout is glitch-free.
+//! path so we can confirm the clock-paced playout is glitch-free.
 //!
 //! Use your ears on the tone (any click/pop is a glitch) together with the
 //! `playout-health:` lines tailed to stdout:
@@ -18,17 +18,24 @@
 //! Run:  cargo run --bin audio_probe -- [freq_hz] [seconds] [target_node]
 //!   e.g. cargo run --release --bin audio_probe -- 440 30
 //!
-//! This probe exercises the PipeWire backend directly, so it is a Linux-only tool.
+//! This probe exercises the platform playback backend directly: PipeWire on Linux
-//! On non-Linux targets `main` is a stub that explains the limitation.
+//! and cpal/WASAPI on Windows. Other targets use a stub that explains the limitation.
 
 #[cfg(target_os = "linux")]
 fn main() {
     unix_probe::run();
 }
 
-#[cfg(not(target_os = "linux"))]
+#[cfg(windows)]
 fn main() {
-    eprintln!("audio_probe is only supported on Linux builds (it drives the PipeWire backend directly).");
+    win_probe::run();
+}
+
+#[cfg(not(any(target_os = "linux", windows)))]
+fn main() {
+    eprintln!(
+        "audio_probe is only supported on Linux and Windows builds (it drives the platform playback backend directly)."
+    );
 }
 
 #[cfg(target_os = "linux")]
@@ -88,7 +95,114 @@ mod unix_probe {
             for _ in 0..FRAME_SAMPLES {
                 let t = n as f32 / SAMPLE_RATE;
                 // 0.25 amplitude: clearly audible but not harsh.
-                let sample = (0.25 * i16::MAX as f32 * (2.0 * std::f32::consts::PI * freq * t).sin()) as i16;
+                let sample =
+                    (0.25 * i16::MAX as f32 * (2.0 * std::f32::consts::PI * freq * t).sin()) as i16;
+                // Stereo playback bus: duplicate the probe tone to L/R.
+                frame.push(sample);
+                frame.push(sample);
+                n += 1;
+            }
+            if tx.send(frame).is_err() {
+                eprintln!("playback channel closed early");
+                break;
+            }
+        }
+
+        // Let the ring drain, then stop.
+        tokio::time::sleep(Duration::from_millis(300)).await;
+        let _ = backend.stop();
+        println!("\naudio_probe: done.");
+    }
+
+    /// Open the app log, seek to the end, and echo new lines (the `playout-health:`
+    /// reports) to stdout once they appear.
+    fn spawn_log_tailer() {
+        let path = peerspeak::log_file_path();
+        std::thread::spawn(move || {
+            // Wait for the file to exist (first log_msg creates it).
+            let file = loop {
+                if let Ok(f) = std::fs::File::open(&path) {
+                    break f;
+                }
+                std::thread::sleep(Duration::from_millis(100));
+            };
+            let mut reader = BufReader::new(file);
+            let _ = reader.seek(SeekFrom::End(0));
+            loop {
+                let mut line = String::new();
+                match reader.read_line(&mut line) {
+                    Ok(0) => std::thread::sleep(Duration::from_millis(150)),
+                    Ok(_) => {
+                        if line.contains("playout-health:") {
+                            print!("{line}");
+                        }
+                    }
+                    Err(_) => std::thread::sleep(Duration::from_millis(150)),
+                }
+            }
+        });
+    }
+}
+
+#[cfg(windows)]
+mod win_probe {
+    use std::io::{BufRead, BufReader, Seek, SeekFrom};
+    use std::sync::Arc;
+    use std::sync::atomic::AtomicUsize;
+    use std::sync::mpsc;
+    use std::time::Duration;
+
+    use peerspeak::audio::AudioBackend;
+    use peerspeak::audio::cpal_impl::CpalBackend;
+    use peerspeak::core::jitter::FRAME_SAMPLES; // 960 mono frames = 20ms @ 48kHz
+
+    const SAMPLE_RATE: f32 = 48_000.0;
+
+    #[tokio::main]
+    pub async fn run() {
+        let mut args = std::env::args().skip(1);
+        let freq: f32 = args.next().and_then(|s| s.parse().ok()).unwrap_or(440.0);
+        let secs: u64 = args.next().and_then(|s| s.parse().ok()).unwrap_or(30);
+        let target_node: Option<String> = args.next();
+
+        // The playout-health logger is quiet in normal operation (it only logs
+        // glitches); ask it for the full once-per-second heartbeat so the probe can
+        // show the steady-state numbers.
+        // SAFETY: set before any playback thread starts, so no concurrent env read.
+        unsafe { std::env::set_var("PEERSPEAK_AUDIO_VERBOSE", "1") };
+
+        println!("audio_probe: {freq} Hz tone for {secs}s through the real playback path.");
+        println!("Listen for clicks/pops; watch the playout-health lines below.\n");
+
+        // Tail the app log (where playout-health lines land) to stdout in the
+        // background so it's all in one terminal.
+        spawn_log_tailer();
+
+        let backend = CpalBackend::new();
+        let (tx, rx) = mpsc::channel::<Vec<i16>>();
+        let ring_fill = Arc::new(AtomicUsize::new(0));
+        if let Err(e) = backend.start_playback(rx, target_node, ring_fill.clone()) {
+            eprintln!("failed to start playback: {e}");
+            return;
+        }
+
+        // Phase-continuous sine, generated one 20ms frame at a time, fill-paced
+        // exactly like the production mixer: only produce while the ring is below
+        // target, so production tracks the cpal/WASAPI hardware clock.
+        use std::sync::atomic::Ordering;
+        let deadline = tokio::time::Instant::now() + Duration::from_secs(secs);
+        let mut n: u64 = 0; // running sample index keeps phase continuous across frames
+        while tokio::time::Instant::now() < deadline {
+            if ring_fill.load(Ordering::Relaxed) >= peerspeak::audio::PLAYBACK_TARGET_SAMPLES {
+                tokio::time::sleep(Duration::from_millis(2)).await;
+                continue;
+            }
+            let mut frame = Vec::with_capacity(FRAME_SAMPLES * peerspeak::audio::PLAYBACK_CHANNELS);
+            for _ in 0..FRAME_SAMPLES {
+                let t = n as f32 / SAMPLE_RATE;
+                // 0.25 amplitude: clearly audible but not harsh.
+                let sample =
+                    (0.25 * i16::MAX as f32 * (2.0 * std::f32::consts::PI * freq * t).sin()) as i16;
                 // Stereo playback bus: duplicate the probe tone to L/R.
                 frame.push(sample);
                 frame.push(sample);