diff --git a/Cargo.toml b/Cargo.toml index 8ed7269..f9bfde3 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -49,7 +49,7 @@ tokio-stream = "0.1.18" # app talks to the `AudioBackend` trait and the `PlatformAudioBackend` alias (see # `src/audio/mod.rs`), so platform selection is confined to these few lines. -[target.'cfg(unix)'.dependencies] +[target.'cfg(target_os = "linux")'.dependencies] # Linux audio backend. v0_3_49 exposes `Buffer::requested()` (the graph's per-cycle # quantum), used by the playback RT callback to fill exactly what the device asks # for instead of a hard-coded 1024-frame quantum (crackle on non-1024 hardware). diff --git a/docs/WINDOWS.md b/docs/WINDOWS.md new file mode 100644 index 0000000..37ca748 --- /dev/null +++ b/docs/WINDOWS.md @@ -0,0 +1,77 @@ +# PeerSpeak on Windows + +Current status: the Windows port cross-compiles to `x86_64-pc-windows-gnu` and the `.exe` +launches under Wine. A real Windows/WASAPI host is still needed for the final audio-device +checks listed below. + +## What works today + +| Area | Status | +|---|---| +| GUI | Iced/wgpu builds and renders under Wine. | +| Networking | Iroh QUIC transport and gossip compile on Windows. | +| Audio backend | `cpal` drives WASAPI capture/playback behind `AudioBackend`. | +| Codec | Opus remains 48 kHz mono, 20 ms frames. | +| Identity | `ring` identity generation/load is platform-neutral. | +| Chimes | Windows uses PowerShell `System.Media.SoundPlayer` for WAV playback. | + +Windows paths are resolved through `dirs`: + +- Config: `%APPDATA%\peerspeak\config.json` +- Identity: `%APPDATA%\peerspeak\identity.key` +- Log: `%LOCALAPPDATA%\peerspeak\peerspeak.log` + +## Building + +### Native Windows + +Install MSVC Build Tools and CMake, then build normally: + +```powershell +cargo build --release +``` + +If CMake is 4.x or newer, the vendored `opus`/`libopus` build may need: + +```powershell +$env:CMAKE_POLICY_VERSION_MINIMUM = "3.5" +cargo build --release +``` + +### Cross-compile from Linux + +The current dev path cross-compiles from an Arch environment to the GNU Windows target: + +```sh +rustup target add x86_64-pc-windows-gnu +sudo pacman -S mingw-w64-gcc cmake +CMAKE_POLICY_VERSION_MINIMUM=3.5 cargo build --release --target x86_64-pc-windows-gnu --bin peerspeak +``` + +Wine is useful for launch/render smoke tests, but it is not a substitute for a real +Windows audio-device pass. The deeper migration plan (phases, decisions, the opus build +spike) lives in the maintainer's handoff docs, outside the repo. + +## First run and networking + +Expect a Windows Firewall prompt the first time the app opens network sockets. Allow it: +PeerSpeak uses UDP for QUIC, plus relay traffic when direct NAT traversal is not available. + +The default network mode keeps the n0 relay available for NAT traversal without publishing +presence to n0 DNS. Direct peer-to-peer paths may work when both networks allow them; relayed +connections are expected and valid. + +## Known gaps + +| Item | Status | +|---|---| +| 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 | 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 | 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 +play notification chimes. diff --git a/src/app/mod.rs b/src/app/mod.rs index 928157e..fc6dfe5 100644 --- a/src/app/mod.rs +++ b/src/app/mod.rs @@ -2,7 +2,7 @@ use crate::core::{CoreController, messages::{CoreCommand, UiEvent}}; use crate::network::PeerState; use crate::notify::{self, Sound}; use crate::audio::eq::{EqSettings, EQ_GAIN_DB_MAX, EQ_GAIN_DB_MIN}; -use crate::audio::pw_cli::{AudioDevice, enumerate_audio_devices}; +use crate::audio::{AudioDevice, enumerate_audio_devices}; use crate::config::{AppConfig, NetworkMode, RecordingMode, RoomLayout}; use crate::hotkeys::{format_binding, HotkeyAction, HotkeyContext, KeyBinding}; use crate::presence::PresenceMode; @@ -553,11 +553,9 @@ pub fn run_gui() -> iced::Result { // the icon from the .desktop file matched by app_id instead). icon: window_icon(), // app_id must match the .desktop basename so Wayland compositors - // (e.g. KWin) attach our launcher icon to the window. - platform_specific: iced::window::settings::PlatformSpecific { - application_id: "peerspeak".to_string(), - ..Default::default() - }, + // (e.g. KWin) attach our launcher icon to the window. The field is + // Linux-only in iced (X11/Wayland); see platform_specific_settings(). + platform_specific: platform_specific_settings(), // We save the final size ourselves on CloseRequested, then exit. exit_on_close_request: false, ..Default::default() @@ -565,6 +563,22 @@ pub fn run_gui() -> iced::Result { .run() } +/// Window `PlatformSpecific` settings. `application_id` (used by X11/Wayland to +/// match our `.desktop` launcher icon) only exists in iced on Linux, so it is +/// set there and left at defaults on Windows. +#[cfg(target_os = "linux")] +fn platform_specific_settings() -> iced::window::settings::PlatformSpecific { + iced::window::settings::PlatformSpecific { + application_id: "peerspeak".to_string(), + ..Default::default() + } +} + +#[cfg(not(target_os = "linux"))] +fn platform_specific_settings() -> iced::window::settings::PlatformSpecific { + iced::window::settings::PlatformSpecific::default() +} + /// Build the window icon from an embedded 128×128 straight-RGBA blob rendered /// from `assets/icons/peerspeak.svg`. Using `from_rgba` (always available) keeps /// us off iced's heavy `image` feature — the blob is raw pixels, no decoder. @@ -2557,10 +2571,28 @@ fn view(state: &AppState) -> Element<'_, AppMessage> { mic_meter, text("Drag the yellow handle to set the gate. While talking, place it just above your quiet-room level so silence is muted but your voice passes through.").size(11).color(color_subtext), vertical_space(4.0), - checkbox(state.config.echo_cancellation_enabled) - .label("Echo cancellation") - .on_toggle(AppMessage::ToggleEchoCancellation), - text("Cancels speaker echo + suppresses noise (PipeWire). Takes effect on your next room join.").size(11).color(color_subtext), + { + let control: Element<'_, AppMessage> = { + #[cfg(target_os = "linux")] + { + column![ + checkbox(state.config.echo_cancellation_enabled) + .label("Echo cancellation") + .on_toggle(AppMessage::ToggleEchoCancellation), + text("Cancels speaker echo + suppresses noise (PipeWire). Takes effect on your next room join.").size(11).color(color_subtext), + ].spacing(8).into() + } + #[cfg(not(target_os = "linux"))] + { + column![ + checkbox(false) + .label("Echo cancellation"), + text("Echo cancellation is not available on Windows yet.").size(11).color(color_subtext), + ].spacing(8).into() + } + }; + control + }, ].spacing(8).width(iced::Length::Fill), ] .spacing(10) @@ -3263,26 +3295,40 @@ fn view(state: &AppState) -> Element<'_, AppMessage> { column![] }, vertical_space(20.0), - // Echo cancellation — same flag + message as the Settings checkbox, so - // toggling here and there stay in sync automatically (single source of - // truth: config.echo_cancellation_enabled). Tooltip is explicit that it - // applies on the NEXT join (the PipeWire-module AEC is wired at join - // time, not hot-swappable mid-call). - tooltip( - checkbox(state.config.echo_cancellation_enabled) - .label("Echo cancellation") - .on_toggle(AppMessage::ToggleEchoCancellation), - container( - text("Cancels speaker echo + suppresses noise. Applies on your next room join.") - .size(11) - .color(color_text), - ) - .padding(8) - .max_width(260.0) - .style(c_style(color_crust, color_surface, 6.0)), - iced::widget::tooltip::Position::Top, - ) - .gap(8), + { + // Echo cancellation is wired at join time on Linux; other + // targets show an inert status row instead of a dead toggle. + let control: Element<'_, AppMessage> = { + #[cfg(target_os = "linux")] + { + tooltip( + checkbox(state.config.echo_cancellation_enabled) + .label("Echo cancellation") + .on_toggle(AppMessage::ToggleEchoCancellation), + container( + text("Cancels speaker echo + suppresses noise. Applies on your next room join.") + .size(11) + .color(color_text), + ) + .padding(8) + .max_width(260.0) + .style(c_style(color_crust, color_surface, 6.0)), + iced::widget::tooltip::Position::Top, + ) + .gap(8) + .into() + } + #[cfg(not(target_os = "linux"))] + { + column![ + checkbox(false) + .label("Echo cancellation"), + text("Not available on Windows yet.").size(11).color(color_subtext), + ].spacing(4).into() + } + }; + control + }, vertical_space(20.0), { let (rec_kind, rec_label, rec_bg, rec_hover, rec_fg) = if state.recording { diff --git a/src/audio/cpal_impl.rs b/src/audio/cpal_impl.rs index 4555a97..4ae5b7d 100644 --- a/src/audio/cpal_impl.rs +++ b/src/audio/cpal_impl.rs @@ -7,6 +7,9 @@ //! //! - **Capture**: mono, 48 kHz, S16 PCM, emitted as `Vec` frames of //! [`CAPTURE_FRAME`] (960 = 20 ms) samples — matching the encoder/jitter frame. +//! The RT capture callback only downmixes and pushes samples into a lock-free +//! ring; the owning thread drains that ring, frames it, and sends — so the +//! callback never allocates, locks, or touches an mpsc channel. //! - **Playback**: stereo interleaved ([`PLAYBACK_CHANNELS`]) S16 PCM at 48 kHz, //! drained from a ring buffer that is paced to the device's hardware clock via //! `ring_fill` exactly as the PipeWire backend does. @@ -20,35 +23,64 @@ //! stream, plays it, and keeps it alive until the per-worker `running` flag flips //! (set by `stop`). The struct holds only `Send` handles (the flag + the join //! handle). The stream's RT callback does the actual audio work; the owning -//! thread additionally feeds the playback ring from the network mixer. +//! thread additionally feeds the playback ring (or drains the capture ring). //! -//! ## Sample rate +//! `start_*` does not return until the owning thread reports back over a readiness +//! channel that the device resolved and the stream is built and playing — so a +//! device/format/WASAPI failure surfaces as a real `Err` to the caller instead of +//! leaving the UI in a joined-but-silent room. //! -//! The whole pipeline assumes 48 kHz (Opus + the 960-sample frame). Phase 1 only -//! selects a native-48 kHz device config; if the device can't do 48 kHz we return -//! a clear error rather than silently producing pitch-shifted audio. Arbitrary -//! sample-rate support (resampling) is a Phase 1.1 follow-up. +//! ## Sample rate and channel layout (W4) +//! +//! The whole pipeline runs internally at 48 kHz (Opus + the 960-sample frame) and +//! mono capture / stereo playback. We prefer a native-48 kHz device config so the +//! common case is conversion-free and bit-exact. When the device can't do 48 kHz +//! (commonly a 44.1 kHz-only endpoint) or can't do stereo output, we fall back to +//! the device's default config and convert at the boundary with the dep-free +//! [`super::resample`] linear resamplers instead of hard-erroring: +//! +//! - **Capture**: the device-rate mono stream is resampled to 48 kHz on the +//! capture drain thread (off the RT callback) before framing. +//! - **Playback**: the internal 48 kHz stereo bus is resampled to the device rate +//! and remapped to the device channel count inside the output RT callback, which +//! pulls internal frames from the ring on demand (allocation-free, so RT-safe). +//! The ring, prefill, and `ring_fill` pacing stay in internal 48 kHz-stereo +//! units, so the mixer is unchanged. +//! +//! Linear interpolation has no anti-aliasing filter (see [`super::resample`] docs); +//! 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::mpsc::{Receiver, RecvTimeoutError, Sender}; +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::{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::{AudioBackend, AudioError, PLAYBACK_CHANNELS, PLAYBACK_TARGET_SAMPLES}; +use super::resample::{PushResampler, StereoPullResampler}; +use super::{AudioBackend, AudioDevice, AudioError, PLAYBACK_CHANNELS, PLAYBACK_TARGET_SAMPLES}; /// The one sample rate the pipeline supports (Opus + the 20 ms frame). const SAMPLE_RATE: u32 = 48_000; /// Mono capture frame: 960 samples = 20 ms @ 48 kHz. Matches the PipeWire backend /// and `core::jitter::FRAME_SAMPLES`. const CAPTURE_FRAME: usize = 960; +/// Lock-free capture ring capacity (mono samples) between the RT callback and the +/// owning drain thread: 8 frames = 160 ms of headroom, so a scheduling hiccup on +/// the drain thread doesn't immediately overrun the RT producer. +const CAPTURE_RING_CAPACITY: usize = CAPTURE_FRAME * 8; +/// How long the capture drain thread sleeps when the ring is momentarily empty, +/// before polling again. Small enough to stay well under the 20 ms frame cadence. +const CAPTURE_POLL: Duration = Duration::from_millis(5); /// Playback ring capacity in interleaved samples: 200 ms of stereo @ 48 kHz. /// Comfortably above [`PLAYBACK_TARGET_SAMPLES`] so the clock-paced producer has /// headroom and never has to drop frames in steady state. @@ -56,24 +88,168 @@ 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 prove itself (deliver its first RT +/// callbacks) 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); +/// Number of completed RT callbacks the owner waits for before declaring the stream +/// live. One callback isn't proof: a stream can fire once and immediately fail in +/// the same processing cycle, so requiring a couple of cycles (plus the terminal +/// error check) keeps a one-shot-then-dead stream from being reported Ok (B1). +const MIN_START_CALLBACKS: usize = 2; +/// Backstop for [`finish_start`]: bounds the WHOLE owner path (resolve + build + +/// play + the [`STREAM_START_TIMEOUT`] callback wait + any wedged-stream cleanup). +/// Sized as a generous setup budget plus the callback wait plus slack so a slow but +/// valid device (e.g. a Bluetooth endpoint that takes seconds to spin up) is not +/// falsely failed, while a driver that wedges before the owner can report is still +/// released eventually (review W6, B4). +const FINISH_START_TIMEOUT: Duration = Duration::from_secs(10); +/// 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 RT data +/// callback bumps `callbacks`, or an error callback sets `err_code`. Returns `Ok` +/// once [`MIN_START_CALLBACKS`] cycles have run with no error, `Err` on an +/// error-callback code or [`STREAM_START_TIMEOUT`], or a clean abort if `stop()` +/// cleared `running` mid-start. Polls a few cheap atomics on the owner thread — +/// never the RT thread (review W1). +/// +/// The error is **terminal and wins any race** with `callbacks`: a stream can run a +/// callback and then fail in the same processing cycle, so `err_code` is checked +/// first each loop AND re-checked before declaring success (review B1). +fn wait_for_stream_start( + callbacks: &AtomicUsize, + err_code: &AtomicU8, + running: &AtomicBool, +) -> Result<(), AudioError> { + let deadline = Instant::now() + STREAM_START_TIMEOUT; + let as_err = |code: u8| Err(AudioError::Stream(stream_err_text(code).to_string())); + loop { + let code = err_code.load(Ordering::Relaxed); + if code != STREAM_ERR_NONE { + return as_err(code); + } + if callbacks.load(Ordering::Relaxed) >= MIN_START_CALLBACKS { + // Re-check: a callback that pushed us to the threshold may have been the + // last before a same-cycle failure. Let a terminal error win. + let code = err_code.load(Ordering::Relaxed); + if code != STREAM_ERR_NONE { + return as_err(code); + } + return Ok(()); + } + 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 { - 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), } } } @@ -91,21 +267,29 @@ impl AudioBackend for CpalBackend { target_node: Option, ) -> Result<(), AudioError> { let mut guard = self.capture.lock().unwrap(); - if guard.is_some() { - return Err(AudioError::Stream("Capture already started".to_string())); - } + 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 || { - if let Err(e) = run_capture(tx, target_node, running_thread) { - crate::log_msg(&format!("cpal capture error: {e}")); - } + let _exit = ExitGuard(exited_thread); + run_capture(tx, target_node, running_thread, ready_tx); }) .map_err(|e| AudioError::Init(e.to_string()))?; - *guard = Some(StreamWorker { running, thread }); - Ok(()) + finish_start( + guard, + StreamWorker { + running, + exited, + thread, + }, + ready_rx, + "capture", + ) } fn start_playback( @@ -115,45 +299,158 @@ impl AudioBackend for CpalBackend { ring_fill: Arc, ) -> Result<(), AudioError> { let mut guard = self.playback.lock().unwrap(); - if guard.is_some() { - return Err(AudioError::Stream("Playback already started".to_string())); - } + 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 || { - if let Err(e) = run_playback(rx, target_node, ring_fill, running_thread) { - crate::log_msg(&format!("cpal playback error: {e}")); - } + let _exit = ExitGuard(exited_thread); + run_playback(rx, target_node, ring_fill, running_thread, ready_tx); }) .map_err(|e| AudioError::Init(e.to_string()))?; - *guard = Some(StreamWorker { running, thread }); - Ok(()) + finish_start( + guard, + StreamWorker { + running, + exited, + thread, + }, + ready_rx, + "playback", + ) } 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(()) } } +/// Block until the just-spawned worker reports (over `ready_rx`) that its stream +/// is built and playing, then either install it (`Ok`) or join it and surface the +/// 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<'_, SlotState>, + worker: StreamWorker, + ready_rx: Receiver>, + what: &str, +) -> Result<(), AudioError> { + // 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 = SlotState::Live(worker); + Ok(()) + } + // 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(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. 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:?}" + ))) + } + } +} + +// --------------------------------------------------------------------------- +// Device enumeration (for the settings device pickers) +// --------------------------------------------------------------------------- + +/// Enumerate WASAPI input/output devices via cpal, sorted by description to match +/// the PipeWire backend's stable UI ordering. +/// +/// cpal exposes a single friendly name per device, which is also what [`resolve`] +/// matches `target_node` against — so `name` and `description` are the same string +/// and a saved selection round-trips. Note: WASAPI device names are less stable +/// across driver/endpoint changes than PipeWire node names, so a saved device may +/// not always be found again; selection then falls back to the system default. +pub fn enumerate_audio_devices() -> Vec { + let host = cpal::default_host(); + let mut devices = Vec::new(); + + if let Ok(inputs) = host.input_devices() { + for device in inputs { + if let Ok(name) = device.name() { + devices.push(AudioDevice { + description: name.clone(), + name, + is_input: true, + }); + } + } + } + if let Ok(outputs) = host.output_devices() { + for device in outputs { + if let Ok(name) = device.name() { + devices.push(AudioDevice { + description: name.clone(), + name, + is_input: false, + }); + } + } + } + + devices.sort_by(|a, b| a.description.cmp(&b.description)); + devices +} + // --------------------------------------------------------------------------- // Device / config selection // --------------------------------------------------------------------------- /// Resolve a device (by `target` name, else the system default) and a stream -/// config running natively at [`SAMPLE_RATE`]. -/// -/// For output we require [`PLAYBACK_CHANNELS`] (stereo) so the interleaved ring -/// maps 1:1 to the device buffer; for input we prefer mono but accept any channel -/// count and downmix. A device with no 48 kHz config is a hard error (no -/// resampling yet — see module docs). +/// config. We prefer a config running natively at [`SAMPLE_RATE`] (conversion-free); +/// if the device has none, we fall back to its default config and resample/remap at +/// the boundary (W4 — see module docs and [`choose_config`]). fn resolve( output: bool, target: Option, @@ -168,7 +465,20 @@ fn resolve( } }; let device = match target { - Some(name) => find_device_by_name(&host, output, &name).or_else(default), + // A saved device name that no longer resolves falls back to the system + // default — but log it, because WASAPI friendly names can change across + // driver/endpoint changes, so a silent fallback otherwise looks like + // "audio went to the wrong device for no reason" (review W7). + Some(ref name) => match find_device_by_name(&host, output, name) { + Some(dev) => Some(dev), + None => { + crate::log_msg(&format!( + "cpal: saved {} device '{name}' not found; using system default", + if output { "output" } else { "input" }, + )); + default() + } + }, None => default(), } .ok_or_else(|| AudioError::Device("no audio device available".to_string()))?; @@ -176,6 +486,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)) } @@ -185,15 +511,54 @@ fn find_device_by_name(host: &cpal::Host, output: bool, name: &str) -> Option Result { +/// Whether the backend can actually open this config range. The workers only build +/// `F32`/`I16`/`U16` streams ([`build_input`]/[`build_output`] — every other sample +/// format hits the `other => Err(...)` arm), and a zero-channel range would later be +/// rejected by [`resolve`]'s geometry check. Filtering both here keeps +/// [`choose_config`] from *ranking* a range it can't drive ahead of a usable one and +/// then hard-failing the start instead of trying the next candidate (Codex B5 +/// re-review, P3). +fn usable_range(r: &cpal::SupportedStreamConfigRange) -> bool { + r.channels() > 0 && format_supported(r.sample_format()) +} + +/// Sample formats the capture/playback stream builders accept. Pure, so it's +/// unit-testable independently of the cpal range types. +fn format_supported(fmt: SampleFormat) -> bool { + matches!( + fmt, + SampleFormat::F32 | SampleFormat::I16 | SampleFormat::U16 + ) +} + +/// 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 { + 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. 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). +/// +/// 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 .supported_output_configs() @@ -213,25 +578,64 @@ fn choose_config( let pick = |channels: Option| { ranges .iter() - .find(|r| supports_48k(r) && channels.is_none_or(|c| r.channels() == c)) + .find(|r| usable_range(r) && supports_48k(r) && channels.is_none_or(|c| r.channels() == c)) .cloned() }; - let chosen = if output { - pick(Some(PLAYBACK_CHANNELS as u16)) + // Cases 1 + 2: an exact-48 kHz config, preferring the native layout but + // accepting any channel count (the backend remaps without resampling). + let exact = if output { + pick(Some(PLAYBACK_CHANNELS as u16)).or_else(|| pick(None)) } else { pick(Some(1)).or_else(|| pick(None)) }; + if let Some(r) = exact { + return Ok(r.with_sample_rate(SampleRate(SAMPLE_RATE))); + } - chosen - .map(|r| r.with_sample_rate(SampleRate(SAMPLE_RATE))) - .ok_or_else(|| { - AudioError::Device(format!( - "device '{}' has no {SAMPLE_RATE} Hz {} config; resampling not yet implemented (Phase 1.1)", - device.name().unwrap_or_else(|_| "".to_string()), - if output { "stereo output" } else { "input" }, - )) - }) + // 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| -> Option<(cpal::SupportedStreamConfigRange, u32)> { + ranges + .iter() + .filter(|r| usable_range(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 { + device.default_input_config() + } + .map_err(|e| AudioError::Device(e.to_string()))?; + crate::log_msg(&format!( + "cpal: device '{}' has no bounded {} config near {SAMPLE_RATE} Hz; falling back to default {} Hz / {} ch (W4)", + device.name().unwrap_or_else(|_| "".to_string()), + if output { "output" } else { "input" }, + def.sample_rate().0, + def.channels(), + )); + Ok(def) } // --------------------------------------------------------------------------- @@ -242,53 +646,166 @@ fn run_capture( tx: Sender>, target: Option, running: Arc, -) -> Result<(), AudioError> { - let (device, config, sample_format) = resolve(false, target)?; - let channels = config.channels as usize; + ready: Sender>, +) { + // The RT callback pushes mono samples into this lock-free ring; we drain it on + // this (non-RT) thread, so the callback never allocates or sends on a channel. + 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`: each RT data callback + // bumps `callbacks`; the RT error callback sets `err_code` (it does NOT log — + // that would allocate/syscall on the time-critical thread). See W1/W2/B1. + let callbacks = Arc::new(AtomicUsize::new(0)); + let err_code = Arc::new(AtomicU8::new(STREAM_ERR_NONE)); - let stream = match sample_format { - SampleFormat::F32 => build_input::(&device, &config, tx, channels), - SampleFormat::I16 => build_input::(&device, &config, tx, channels), - SampleFormat::U16 => build_input::(&device, &config, tx, channels), - other => Err(AudioError::Stream(format!( - "unsupported capture sample format: {other:?}" - ))), - }?; + // 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(), callbacks.clone(), + err_code.clone(), + ), + SampleFormat::I16 => build_input::( + &device, &config, producer, channels, overrun.clone(), callbacks.clone(), + err_code.clone(), + ), + SampleFormat::U16 => build_input::( + &device, &config, producer, channels, overrun.clone(), callbacks.clone(), + err_code.clone(), + ), + other => Err(AudioError::Stream(format!( + "unsupported capture sample format: {other:?}" + ))), + }?; + stream + .play() + .map_err(|e| AudioError::Stream(e.to_string()))?; + let name = device.name().unwrap_or_else(|_| "".to_string()); + Ok((stream, name, sample_format, channels, device_rate)) + }; - stream.play().map_err(|e| AudioError::Stream(e.to_string()))?; + // 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) => match wait_for_stream_start(&callbacks, &err_code, &running) { + Ok(()) => { + let _ = ready.send(Ok(())); + v + } + Err(e) => { + // Drop the (possibly wedged) stream BEFORE reporting: cpal's + // Stream::drop joins its WASAPI worker, so if that wedges we want + // the Err withheld and finish_start's backstop to detach, rather + // than finish_start joining this owner forever (B2). + drop(v.0); + let _ = ready.send(Err(e)); + return; + } + }, + Err(e) => { + let _ = ready.send(Err(e)); + return; + } + }; + crate::log_msg(&format!( + "cpal capture started: device='{dev_name}' format={sample_format:?} channels={channels} device_rate={device_rate} Hz -> {SAMPLE_RATE} Hz" + )); - // The RT callback does the work; this thread just keeps `stream` alive until - // `stop()` flips the flag, at which point the stream is dropped (= stopped). + // 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)); + // 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 = Vec::new(); + + // Drain the RT ring on this thread: pop mono samples, (resample,) frame them + // (the `Vec` allocation lives here, off the RT path), and send completed + // 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) { - thread::sleep(WORKER_POLL); + let mut drained = false; + while let Some(sample) = consumer.try_pop() { + drained = true; + resampled.clear(); + match resampler { + Some(ref mut rs) => { + rs.push(i16_to_f32(sample), |out| resampled.push(f32_to_i16(out))); + } + None => resampled.push(sample), + } + for s in resampled.drain(..) { + if let Some(frame) = acc.push(s) { + // Consumer gone (call ended) → stop feeding; the stream is + // dropped below on the way out. + if tx.send(frame).is_err() { + drop(stream); + return; + } + } + } + } + let o = overrun.load(Ordering::Relaxed); + if o != last_overrun { + crate::log_msg(&format!( + "cpal capture overrun: dropped {} samples (drain thread fell behind)", + o - last_overrun + )); + 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); + } } - Ok(()) + drop(stream); } -fn build_input( +#[allow(clippy::too_many_arguments)] +fn build_input( device: &Device, config: &StreamConfig, - tx: Sender>, + mut producer: P, channels: usize, + overrun: Arc, + callbacks: Arc, + err_code: Arc, ) -> Result where T: SizedSample + Send + 'static, i16: FromSample, + P: Producer + Send + 'static, { - let mut acc = FrameAccumulator::new(CAPTURE_FRAME); - 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], _| { + // Count callbacks so the owner can confirm the stream is really + // running before reporting Ok (W1/B1). + callbacks.fetch_add(1, 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) { let mono = downmix_to_mono(frame); - if let Some(full) = acc.push(mono) { - // Consumer gone (call ended) → stop feeding; the owning - // thread will drop the stream on `stop()`. - if tx.send(full).is_err() { - return; - } + if producer.try_push(mono).is_err() { + overrun.fetch_add(1, Ordering::Relaxed); } } }, @@ -312,6 +829,18 @@ where (sum / frame.len() as i32) as i16 } +/// Scale an i16 PCM sample to f32 in roughly `[-1, 1]` for interpolation. +#[inline] +fn i16_to_f32(s: i16) -> f32 { + s as f32 / 32768.0 +} + +/// Convert an interpolated f32 sample back to i16, clamping to range. +#[inline] +fn f32_to_i16(x: f32) -> i16 { + (x * 32768.0).clamp(i16::MIN as f32, i16::MAX as f32) as i16 +} + /// Accumulates mono samples into fixed-size [`CAPTURE_FRAME`] frames. Pulled out /// of the RT callback so the framing is unit-testable. struct FrameAccumulator { @@ -350,9 +879,8 @@ fn run_playback( target: Option, ring_fill: Arc, running: Arc, -) -> Result<(), AudioError> { - let (device, config, sample_format) = resolve(true, target)?; - + ready: Sender>, +) { let rb = HeapRb::::new(RING_CAPACITY); let (mut producer, consumer) = rb.split(); @@ -369,29 +897,76 @@ fn run_playback( // Diagnostics (mirrors the PipeWire backend's playout-health line). let underrun = Arc::new(AtomicU64::new(0)); let dropped = Arc::new(AtomicU64::new(0)); + // Largest single output-callback length seen (interleaved samples). WASAPI + // shared-mode picks its own period, so this can exceed the prefill target — + // 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, B1): each RT callback + // bumps `callbacks`; the RT error callback sets `err_code` without logging. + let callbacks = Arc::new(AtomicUsize::new(0)); + let err_code = Arc::new(AtomicU8::new(STREAM_ERR_NONE)); - let stream = match sample_format { - SampleFormat::F32 => { - build_output::(&device, &config, consumer, ring_fill.clone(), underrun.clone()) - } - SampleFormat::I16 => { - build_output::(&device, &config, consumer, ring_fill.clone(), underrun.clone()) - } - SampleFormat::U16 => { - build_output::(&device, &config, consumer, ring_fill.clone(), underrun.clone()) - } - other => Err(AudioError::Stream(format!( - "unsupported playback sample format: {other:?}" - ))), - }?; + // 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(), callbacks.clone(), err_code.clone(), + ), + SampleFormat::I16 => build_output::( + &device, &config, consumer, ring_fill.clone(), underrun.clone(), + max_cb.clone(), callbacks.clone(), err_code.clone(), + ), + SampleFormat::U16 => build_output::( + &device, &config, consumer, ring_fill.clone(), underrun.clone(), + max_cb.clone(), callbacks.clone(), err_code.clone(), + ), + other => Err(AudioError::Stream(format!( + "unsupported playback sample format: {other:?}" + ))), + }?; + stream + .play() + .map_err(|e| AudioError::Stream(e.to_string()))?; + let name = device.name().unwrap_or_else(|_| "".to_string()); + Ok((stream, name, sample_format, channels, device_rate)) + }; - stream.play().map_err(|e| AudioError::Stream(e.to_string()))?; + // Build + queue, then wait for real callbacks before reporting readiness (W1). + let (stream, dev_name, sample_format, channels, device_rate) = match setup() { + Ok(v) => match wait_for_stream_start(&callbacks, &err_code, &running) { + Ok(()) => { + let _ = ready.send(Ok(())); + v + } + Err(e) => { + // Drop before reporting so a wedged Stream::drop withholds the Err + // and lets finish_start's backstop detach instead of hanging (B2). + drop(v.0); + let _ = ready.send(Err(e)); + return; + } + }, + Err(e) => { + let _ = ready.send(Err(e)); + return; + } + }; + crate::log_msg(&format!( + "cpal playback started: device='{dev_name}' format={sample_format:?} channels={channels} device_rate={device_rate} Hz <- {SAMPLE_RATE} Hz" + )); let logger = spawn_health_logger( running.clone(), ring_fill.clone(), underrun.clone(), dropped.clone(), + max_cb.clone(), + err_code.clone(), ); // Feed the ring from the network mixer until `stop()` flips `running` or the @@ -402,10 +977,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 @@ -413,40 +998,95 @@ fn run_playback( running.store(false, Ordering::Relaxed); let _ = logger.join(); drop(stream); - Ok(()) } +#[allow(clippy::too_many_arguments)] fn build_output( device: &Device, config: &StreamConfig, mut consumer: C, ring_fill: Arc, underrun: Arc, + max_cb: Arc, + callbacks: 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}")); - device - .build_output_stream::( - config, - move |data: &mut [T], _| { - let (popped, starved) = fill_output(&mut consumer, data); - 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())) + // 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 { + device + .build_output_stream::( + config, + move |data: &mut [T], _| { + callbacks.fetch_add(1, 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); + } + 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())) + } else { + let mut resampler = StereoPullResampler::new(SAMPLE_RATE, device_rate); + device + .build_output_stream::( + config, + move |data: &mut [T], _| { + callbacks.fetch_add(1, 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 { + 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())) + } +} + +/// 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. @@ -473,6 +1113,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( + consumer: &mut C, + out: &mut [T], + device_channels: usize, + resampler: &mut StereoPullResampler, +) -> (usize, u64) +where + T: Sample + FromSample, + C: Consumer, +{ + 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( @@ -480,10 +1174,14 @@ fn spawn_health_logger( ring_fill: Arc, 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); @@ -498,6 +1196,31 @@ fn spawn_health_logger( fill / (48 * PLAYBACK_CHANNELS), )); } + // 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 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 demands up to {cb} internal samples/cycle (~{ms}ms), target {PLAYBACK_TARGET_SAMPLES}", + )); + } + } } }) } @@ -558,6 +1281,48 @@ mod tests { assert_eq!(out, [1, 2, 3, 0, 0]); } + #[test] + fn fill_output_remap_downmixes_to_mono() { + let rb = HeapRb::::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::::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::::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::>(); @@ -583,6 +1348,84 @@ 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 format_supported_matches_the_stream_builders() { + // Exactly the three the build_input/build_output match arms accept. + for f in [SampleFormat::F32, SampleFormat::I16, SampleFormat::U16] { + assert!(format_supported(f), "{f:?} should be drivable"); + } + // Everything else cpal can expose must be filtered out before ranking, or a + // start could pick it and then hit the `unsupported sample format` arm (P3). + for f in [ + SampleFormat::I8, + SampleFormat::U8, + SampleFormat::I32, + SampleFormat::U32, + SampleFormat::I64, + SampleFormat::U64, + SampleFormat::F64, + ] { + assert!(!format_supported(f), "{f:?} must not be reported drivable"); + } + } + + #[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::>(); diff --git a/src/audio/mod.rs b/src/audio/mod.rs index c601a36..1d068e0 100644 --- a/src/audio/mod.rs +++ b/src/audio/mod.rs @@ -56,19 +56,50 @@ pub trait AudioBackend: Send + Sync { fn stop(&self) -> Result<(), AudioError>; } -pub mod echo_cancel; pub mod eq; pub mod gate; pub mod limiter; pub mod multitrack; pub mod pan; -#[cfg(unix)] +// Linear resamplers used by the Windows/cpal backend (W4). Platform-neutral and +// pure, so it builds (and its tests run) everywhere even though only the cpal +// backend wires it in. +pub mod resample; +#[cfg(target_os = "linux")] +pub mod echo_cancel; +#[cfg(target_os = "linux")] pub mod pipewire_impl; #[cfg(windows)] pub mod cpal_impl; +#[cfg(target_os = "linux")] pub mod pw_cli; pub mod recorder; +/// A selectable audio device for the input/output pickers. `name` is the stable +/// identifier the backend uses to request the device (`target_node`); +/// `description` is the human-facing label shown in the UI. The two may be equal +/// (cpal/WASAPI) or differ (PipeWire node name vs. description). +#[derive(Debug, Clone, PartialEq, Eq)] +pub struct AudioDevice { + pub name: String, + pub description: String, + pub is_input: bool, +} + +impl std::fmt::Display for AudioDevice { + fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result { + write!(f, "{}", self.description) + } +} + +// Enumerate audio input/output devices for the pickers (sorted by description), +// returning the same `AudioDevice` shape regardless of platform: PipeWire +// (`pw-cli`) on Linux, cpal/WASAPI on Windows. +#[cfg(target_os = "linux")] +pub use pw_cli::enumerate_audio_devices; +#[cfg(windows)] +pub use cpal_impl::enumerate_audio_devices; + /// The audio backend implementation for the current platform. /// /// The whole app constructs and threads this alias (via @@ -76,10 +107,9 @@ pub mod recorder; /// platform selection lives entirely here. Both implementations satisfy the /// [`AudioBackend`] trait, which is the only interface the core talks to. /// -/// - Linux/Unix → PipeWire ([`pipewire_impl::PipeWireBackend`]). -/// - Windows → cpal/WASAPI ([`cpal_impl::CpalBackend`]); a no-op stub until the -/// Phase 1 capture/playback implementation lands. -#[cfg(unix)] +/// - Linux → PipeWire ([`pipewire_impl::PipeWireBackend`]). +/// - Windows → cpal/WASAPI ([`cpal_impl::CpalBackend`]). +#[cfg(target_os = "linux")] pub type PlatformAudioBackend = pipewire_impl::PipeWireBackend; #[cfg(windows)] pub type PlatformAudioBackend = cpal_impl::CpalBackend; diff --git a/src/audio/pw_cli.rs b/src/audio/pw_cli.rs index 51d3ed6..f1e4e61 100644 --- a/src/audio/pw_cli.rs +++ b/src/audio/pw_cli.rs @@ -1,18 +1,6 @@ +use super::AudioDevice; use std::process::Command; -#[derive(Debug, Clone, PartialEq, Eq)] -pub struct AudioDevice { - pub name: String, - pub description: String, - pub is_input: bool, -} - -impl std::fmt::Display for AudioDevice { - fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result { - write!(f, "{}", self.description) - } -} - pub fn enumerate_audio_devices() -> Vec { let output = Command::new("pw-cli") .arg("list-objects") diff --git a/src/audio/resample.rs b/src/audio/resample.rs new file mode 100644 index 0000000..4f5723f --- /dev/null +++ b/src/audio/resample.rs @@ -0,0 +1,307 @@ +//! Dep-free linear-interpolation resamplers for the Windows/cpal backend (W4). +//! +//! The pipeline runs internally at 48 kHz (Opus + the 20 ms frame), but a WASAPI +//! endpoint may run at a different rate (commonly 44.1 kHz) and/or a non-stereo +//! channel layout. These convert at the device boundary so such a device plays and +//! captures instead of hard-erroring (the W4 limitation in the Windows port). +//! +//! ## Where each is used +//! - [`PushResampler`] (single channel) converts **capture** from the device rate +//! to 48 kHz on the capture drain thread — off the RT callback. +//! - [`StereoPullResampler`] converts **playback** from the internal 48 kHz stereo +//! bus to the device rate inside the output RT callback, pulling internal frames +//! from the ring on demand. It allocates nothing in `next`, so it is RT-safe. +//! +//! ## Quality +//! This is plain linear interpolation with no anti-aliasing filter: correct, +//! allocation-free, and adequate for speech, but it adds some aliasing when +//! downsampling. The seam is intentionally tiny so a higher-quality polyphase/FIR +//! resampler (e.g. the `rubato` crate, pending a supply-chain decision) can later +//! replace the internals without touching the cpal backend. The matching-rate / +//! matching-layout path in the backend bypasses these entirely and stays bit-exact. + +/// Linear interpolation between `a` and `b` at fractional position `frac` in `[0, 1)`. +#[inline] +fn lerp(a: f32, b: f32, frac: f32) -> f32 { + a + (b - a) * frac +} + +/// Stateful single-channel **push** resampler: feed input samples at `in_rate`, +/// receive output samples at `out_rate` through an `emit` callback. It carries the +/// fractional read position and the previous input sample across calls, so feeding +/// the stream block-by-block joins seamlessly. Neither [`push`](Self::push) nor +/// [`process`](Self::process) allocates. +pub struct PushResampler { + /// Input samples consumed per output sample (`in_rate / out_rate`). + step: f64, + /// Position of the next output sample, in input-sample units, measured from the + /// index of `prev` (the most recent input). Always advanced to stay `< 1.0` + /// after each input is consumed. + next: f64, + /// The previous input sample (left edge of the current interpolation segment). + prev: f32, + /// Whether any input has been seen yet (anchors the first output at input[0]). + started: bool, +} + +impl PushResampler { + /// 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 { + Self { + step: in_rate.max(1) as f64 / out_rate.max(1) as f64, + next: 0.0, + prev: 0.0, + started: false, + } + } + + /// Feed one input sample; `emit` is called for each output sample produced + /// (zero or more, depending on the rate ratio). + pub fn push(&mut self, cur: f32, mut emit: impl FnMut(f32)) { + if !self.started { + // First sample: just establish the left edge. Linear interpolation + // needs the next input as the right edge, so the first output is + // produced on the next push. This gives exact alignment + // (`output[k] == input[k]` at equal rates) with one input-sample of + // latency — negligible (~20 µs at 48 kHz). + self.started = true; + self.prev = cur; + self.next = 0.0; + return; + } + // `prev` sits at position 0 of this segment and `cur` at position 1; emit + // every output whose position falls in [0, 1). + while self.next < 1.0 { + emit(lerp(self.prev, cur, self.next as f32)); + self.next += self.step; + } + self.next -= 1.0; + self.prev = cur; + } + + /// Convenience for tests / batch callers: push a whole slice. + pub fn process(&mut self, input: &[f32], mut emit: impl FnMut(f32)) { + for &s in input { + self.push(s, &mut emit); + } + } +} + +/// Stateful stereo **pull** resampler: produce output frames at `out_rate` by +/// pulling input frames at `in_rate` from a closure on demand. Call +/// [`next`](Self::next) once per output frame; it pulls as many input frames as the +/// ratio requires and returns the interpolated `(left, right)`, or `None` when the +/// puller runs dry (an underrun). Allocates nothing, so it is safe in an RT output +/// callback. +pub struct StereoPullResampler { + /// Input frames consumed per output frame (`in_rate / out_rate`). + step: f64, + /// Position of the next output frame within `[prev, cur)`, in `[0, 1)`. + frac: f64, + /// Left edge of the current interpolation segment. + prev: (f32, f32), + /// Right edge of the current interpolation segment. + cur: (f32, f32), + /// Whether `prev`/`cur` have been primed from the puller yet. + primed: bool, +} + +impl StereoPullResampler { + /// 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 { + Self { + 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), + primed: false, + } + } + + /// Produce the next output frame, pulling input frames via `pull` as needed. + /// Returns `None` if `pull` returns `None` before the frame can be formed + /// (underrun); the caller should substitute silence for that frame. + pub fn next(&mut self, mut pull: impl FnMut() -> Option<(f32, f32)>) -> Option<(f32, f32)> { + if !self.primed { + // Prime both edges from two pulls so the first output frame aligns + // exactly with the first input frame (`out[0] == in[0]` at equal + // rates). Needs two frames available to start, which the prefilled + // playback ring always has. + self.prev = pull()?; + self.cur = pull()?; + self.primed = true; + self.frac = 0.0; + } + // Advance the segment until the read position lands inside [prev, cur). + while self.frac >= 1.0 { + self.prev = self.cur; + self.cur = pull()?; + self.frac -= 1.0; + } + let f = self.frac as f32; + let out = (lerp(self.prev.0, self.cur.0, f), lerp(self.prev.1, self.cur.1, f)); + self.frac += self.step; + Some(out) + } +} + +#[cfg(test)] +mod tests { + use super::*; + + /// Equal rates align exactly: `output[k] == input[k]`. The final input lands on + /// the next push (one-sample streaming latency), so we get `n - 1` outputs. + #[test] + fn push_identity_when_rates_match() { + let mut r = PushResampler::new(48_000, 48_000); + let input = [0.0, 0.1, 0.2, 0.3, 0.4]; + let mut out = Vec::new(); + r.process(&input, |s| out.push(s)); + assert_eq!(out.len(), input.len() - 1); + for (a, b) in out.iter().zip(input.iter()) { + assert!((a - b).abs() < 1e-6, "{a} vs {b}"); + } + } + + /// Upsampling 2x roughly doubles the output count and the midpoints interpolate. + #[test] + fn push_upsample_2x_interpolates_midpoints() { + let mut r = PushResampler::new(24_000, 48_000); // step = 0.5 + let input = [0.0, 1.0, 2.0, 3.0]; + let mut out = Vec::new(); + r.process(&input, |s| out.push(s)); + // (n - 1) segments at 2 outputs each = 6. + assert_eq!(out.len(), 6, "out {out:?}"); + // A half-step between 1.0 and 2.0 must appear near 1.5. + assert!( + out.iter().any(|&s| (s - 1.5).abs() < 1e-3), + "expected a ~1.5 midpoint in {out:?}" + ); + } + + /// Downsampling drops the rate: fewer outputs than inputs, monotonic ramp preserved. + #[test] + fn push_downsample_reduces_count() { + let mut r = PushResampler::new(48_000, 44_100); // step ~1.088 + let input: Vec = (0..441).map(|i| i as f32).collect(); + let mut out = Vec::new(); + r.process(&input, |s| out.push(s)); + // 441 in @ 48k -> ~405 out @ 44.1k. + assert!( + (390..=410).contains(&out.len()), + "expected ~405 outputs, got {}", + out.len() + ); + // Output stays within the input's value range and is non-decreasing. + for w in out.windows(2) { + assert!(w[1] >= w[0] - 1e-3, "ramp should not reverse: {w:?}"); + } + assert!(*out.last().unwrap() <= 440.0 + 1e-3); + } + + /// Pull resampler at equal rates returns each input frame in order, aligned. + /// Two-pull priming uses one frame of lookahead, so `n` inputs yield `n - 1` + /// outputs (the last frame emits once a successor arrives). + #[test] + fn pull_identity_when_rates_match() { + let mut r = StereoPullResampler::new(48_000, 48_000); + let frames = [(0.0, 9.0), (1.0, 8.0), (2.0, 7.0), (3.0, 6.0)]; + let mut idx = 0; + let mut out = Vec::new(); + while let Some(f) = r.next(|| { + let v = frames.get(idx).copied(); + idx += 1; + v + }) { + out.push(f); + } + assert_eq!(out.len(), frames.len() - 1, "out {out:?}"); + for (got, want) in out.iter().zip(frames.iter()) { + assert!((got.0 - want.0).abs() < 1e-6 && (got.1 - want.1).abs() < 1e-6); + } + } + + /// Pull resampler reports underrun (`None`) once the source is exhausted. + #[test] + fn pull_returns_none_on_underrun() { + let mut r = StereoPullResampler::new(48_000, 44_100); // step ~1.088 -> pulls >1 per out + let frames = [(0.0, 0.0), (1.0, -1.0)]; + let mut idx = 0; + let mut pull = || { + let v = frames.get(idx).copied(); + idx += 1; + v + }; + // First frame primes + emits; subsequent calls eventually exhaust the source. + let mut produced = 0; + let mut hit_none = false; + for _ in 0..10 { + if r.next(&mut pull).is_some() { + produced += 1; + } else { + hit_none = true; + break; + } + } + assert!(produced >= 1, "should produce at least the primed frame"); + assert!(hit_none, "should report underrun once the puller is dry"); + } + + /// Downsampling via pull consumes more input frames than it emits output frames. + #[test] + fn pull_downsample_consumes_more_than_it_emits() { + let mut r = StereoPullResampler::new(48_000, 24_000); // step = 2.0 + let input: Vec<(f32, f32)> = (0..100).map(|i| (i as f32, -(i as f32))).collect(); + let mut idx = 0; + let mut emitted = 0; + for _ in 0..40 { + let f = r.next(|| { + let v = input.get(idx).copied(); + idx += 1; + v + }); + if f.is_some() { + emitted += 1; + } else { + break; + } + } + // 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()); + } +} diff --git a/src/bin/audio_probe.rs b/src/bin/audio_probe.rs index c3fcee3..f7429a3 100644 --- a/src/bin/audio_probe.rs +++ b/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 -//! (`PipeWireBackend::start_playback`), using the *same* fill-paced production -//! the production mixer uses (`core/mod.rs`): generate a frame only while the +//! Drives a phase-continuous sine tone through the *real* playback path +//! (PipeWire on Linux, cpal/WASAPI on Windows), using the *same* fill-paced +//! 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 -//! output path so we can confirm the clock-paced playout is glitch-free. +//! hardware clock. No network, no microphone — this isolates the local output +//! 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,20 +18,27 @@ //! 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 Unix-only tool. -//! On non-Unix targets `main` is a stub that explains the limitation. +//! This probe exercises the platform playback backend directly: PipeWire on Linux +//! and cpal/WASAPI on Windows. Other targets use a stub that explains the limitation. -#[cfg(unix)] +#[cfg(target_os = "linux")] fn main() { unix_probe::run(); } -#[cfg(not(unix))] +#[cfg(windows)] fn main() { - eprintln!("audio_probe is only supported on Unix builds (it drives the PipeWire backend directly)."); + win_probe::run(); } -#[cfg(unix)] +#[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")] mod unix_probe { use std::io::{BufRead, BufReader, Seek, SeekFrom}; use std::sync::Arc; @@ -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 = 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::>(); + 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); diff --git a/src/core/mod.rs b/src/core/mod.rs index 41869b1..afd8341 100644 --- a/src/core/mod.rs +++ b/src/core/mod.rs @@ -390,6 +390,7 @@ struct ActiveSession { grace_timers: GraceTimers, transport: Arc, /// Loaded PipeWire echo-cancel module (if enabled); unloads on drop. + #[cfg(target_os = "linux")] echo_cancel: Option, /// Our pixelpass screen-share host child while sharing (`kill_on_drop`, so it /// also dies if the session is dropped without an explicit stop). @@ -431,6 +432,7 @@ impl ActiveSession { // Unload the echo-cancel module now that the audio streams releasing its // virtual nodes have stopped. (Dropping the guard runs `pactl unload`.) + #[cfg(target_os = "linux")] drop(self.echo_cancel); crate::log_msg("Leaving room..."); @@ -1095,7 +1097,9 @@ async fn run_core_loop( // The guard unloads the module on drop — including the early-return // paths below, since it's a local until moved into the session. On // any failure, warn and fall back to the direct devices. + #[cfg(target_os = "linux")] let mut echo_cancel_guard = None; + #[cfg(target_os = "linux")] let (capture_target, playback_target) = if echo_cancellation { match crate::audio::echo_cancel::enable( input_device.as_deref(), @@ -1122,6 +1126,10 @@ async fn run_core_loop( } else { (input_device.clone(), output_device.clone()) }; + #[cfg(not(target_os = "linux"))] + let _ = echo_cancellation; + #[cfg(not(target_os = "linux"))] + let (capture_target, playback_target) = (input_device.clone(), output_device.clone()); if let Err(e) = audio_backend.start_capture(capture_tx, capture_target) { let _ = ui_tx.send(UiEvent::Error(format!("Failed to start capture: {}", e))).await; @@ -1632,6 +1640,7 @@ async fn run_core_loop( conn_event_task, grace_timers, transport: transport.clone(), + #[cfg(target_os = "linux")] echo_cancel: echo_cancel_guard, screenshare_host: None, screenshare_viewers: Vec::new(), @@ -1807,17 +1816,26 @@ async fn run_core_loop( } CoreCommand::SetNetworkMode(mode) => { - network_mode = mode; - // Rebuild the persistent stack to the new posture immediately if - // idle; if a call is active, defer to the next Leave/Join so the - // live call isn't disrupted (preserves "applies on next join"). - if active_session.is_none() { - let lookup = net.memory_lookup.clone(); - net.shutdown().await; - let publish = presence_mode.lock().unwrap().publishes_to_discovery(); - net = build_net_stack(secret_key.clone(), network_mode, lookup, friends_handler.clone(), publish).await?; - } else { - net_rebuild_pending = true; + // Skip when the posture is unchanged. The GUI re-sends the saved + // network mode as part of its startup config-sync, and that mode + // usually already matches the freshly-built stack — rebuilding the + // iroh endpoint for an identical posture just churns the network + // and adds a needless ~1s teardown+rebuild bounce at every launch + // (seen on both Linux and Windows/Wine). A real change still + // rebuilds exactly as before. + if mode != network_mode { + network_mode = mode; + // Rebuild the persistent stack to the new posture immediately if + // idle; if a call is active, defer to the next Leave/Join so the + // live call isn't disrupted (preserves "applies on next join"). + if active_session.is_none() { + let lookup = net.memory_lookup.clone(); + net.shutdown().await; + let publish = presence_mode.lock().unwrap().publishes_to_discovery(); + net = build_net_stack(secret_key.clone(), network_mode, lookup, friends_handler.clone(), publish).await?; + } else { + net_rebuild_pending = true; + } } } diff --git a/src/notify.rs b/src/notify.rs index 0e7baf9..58355d9 100644 --- a/src/notify.rs +++ b/src/notify.rs @@ -3,11 +3,12 @@ //! //! The WAVs are embedded in the binary (`include_bytes!`) so a deployed single //! binary is self-contained — no asset directory to ship alongside it. On first -//! use each sound is written once to a temp file, then played fire-and-forget -//! via `pw-play` (PipeWire-native; falls back to `paplay`/`aplay`). Playback runs -//! on a detached thread that waits on the child, so it never blocks the UI and -//! never leaves a zombie. Any failure (no player, no audio) is silent by design — -//! a missing chime should never disrupt a call. +//! use each sound is written once to a temp file, then played fire-and-forget. +//! Linux uses `pw-play` (PipeWire-native; falls back to `paplay`/`aplay`); +//! Windows uses PowerShell's `System.Media.SoundPlayer`. Playback runs on a +//! detached thread that waits on the child, so it never blocks the UI and never +//! leaves a zombie. Any failure (no player, no audio) is silent by design — a +//! missing chime should never disrupt a call. use std::collections::HashMap; use std::path::{Path, PathBuf}; @@ -202,8 +203,14 @@ fn cached_path(sound: Sound) -> Option { Some(path) } +#[cfg(any(windows, test))] +fn escape_powershell_single_quoted(s: &str) -> String { + s.replace('\'', "''") +} + /// Try each available player in turn, waiting on the first that starts (which /// reaps the child). Runs on a detached thread, so the wait is harmless. +#[cfg(not(windows))] fn spawn_player(path: &Path) { for player in ["pw-play", "paplay", "aplay"] { let started = Command::new(player) @@ -221,6 +228,23 @@ fn spawn_player(path: &Path) { } } +/// Play through Windows' built-in WAV player. Runs on a detached thread, so +/// `PlaySync()` blocking for the sound duration is fine. +#[cfg(windows)] +fn spawn_player(path: &Path) { + let path = escape_powershell_single_quoted(&path.display().to_string()); + let command = format!("(New-Object System.Media.SoundPlayer '{path}').PlaySync()"); + let _ = Command::new("powershell") + .arg("-NoProfile") + .arg("-NonInteractive") + .arg("-Command") + .arg(command) + .stdin(Stdio::null()) + .stdout(Stdio::null()) + .stderr(Stdio::null()) + .status(); +} + #[cfg(test)] mod tests { use super::*; @@ -234,6 +258,18 @@ mod tests { assert!(!should_play(false, false)); } + #[test] + fn test_powershell_single_quote_escape() { + assert_eq!( + escape_powershell_single_quoted(r"C:\Users\O'Brien\chime.wav"), + r"C:\Users\O''Brien\chime.wav" + ); + assert_eq!( + escape_powershell_single_quoted("a'b'c"), + "a''b''c" + ); + } + #[test] fn test_sound_indices_unique_and_match_all() { // `index()` must be a 0..COUNT bijection in `ALL` order, or the flag diff --git a/src/screenshare/mod.rs b/src/screenshare/mod.rs index 2029220..6187673 100644 --- a/src/screenshare/mod.rs +++ b/src/screenshare/mod.rs @@ -25,6 +25,16 @@ use tokio::process::{Child, Command}; /// points elsewhere. const PIXELPASS_BIN: &str = "pixelpass"; +#[cfg(windows)] +fn pixelpass_path_candidates(dir: &Path) -> [PathBuf; 2] { + [dir.join(PIXELPASS_BIN), dir.join("pixelpass.exe")] +} + +#[cfg(not(windows))] +fn pixelpass_path_candidates(dir: &Path) -> [PathBuf; 1] { + [dir.join(PIXELPASS_BIN)] +} + /// Pixelpass endpoint tickets are normally ~140 chars. Leave headroom for format /// growth, but reject unbounded gossip payloads before the UI offers "Watch". const MAX_TICKET_LEN: usize = 512; @@ -143,7 +153,7 @@ pub fn pixelpass_path(config_override: Option<&str>) -> Option { } let path_var = std::env::var_os("PATH")?; std::env::split_paths(&path_var) - .map(|dir| dir.join(PIXELPASS_BIN)) + .flat_map(|dir| pixelpass_path_candidates(&dir)) .find(|c| c.is_file()) } @@ -513,4 +523,14 @@ mod tests { // only assert it doesn't return the empty path as a match. assert_ne!(pixelpass_path(Some(" ")).as_deref(), Some(Path::new(""))); } + + #[test] + fn pixelpass_path_candidates_are_platform_specific() { + let dir = Path::new("bin"); + let candidates: Vec = pixelpass_path_candidates(dir).into_iter().collect(); + #[cfg(windows)] + assert_eq!(candidates, vec![dir.join("pixelpass"), dir.join("pixelpass.exe")]); + #[cfg(not(windows))] + assert_eq!(candidates, vec![dir.join("pixelpass")]); + } }