From 99bd1f0c576f9898eda8aa5ba01ae9ffa0cdcf3b Mon Sep 17 00:00:00 2001 From: Mollusk Date: Mon, 1 Jun 2026 16:41:06 -0400 Subject: [PATCH] feat: echo cancellation via PipeWire module-echo-cancel MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Add an opt-in Echo Cancellation toggle (Settings) that routes the call through PipeWire's module-echo-cancel (WebRTC AEC + noise suppression + AGC) instead of running an in-process canceller. PipeWire already sees both the mic and the speaker monitor, so it handles the echo-reference alignment for free and we avoid a C++ DSP dependency. src/audio/echo_cancel.rs (new): - enable(real_source, real_sink) loads the module via pactl (aec_method=webrtc), bound to the chosen devices with source_master/sink_master (defaults if unset), waits for the virtual nodes to appear, and returns an RAII guard that unloads the module on drop. Best-effort pre-clean of a stale instance from a crashed run. - EC_SOURCE / EC_SINK are the virtual cleaned-mic source and reference sink. core: when echo_cancellation is set on Join, load the module and point capture at EC_SOURCE / playback at EC_SINK; stash the guard in ActiveSession so it unloads on shutdown (after the audio streams release the nodes). Any failure logs + warns the UI and falls back to the direct devices — never blocks the call. config: new echo_cancellation_enabled (serde default false). app: Settings checkbox under Mic Sensitivity, applied on next room join. An ignored live smoke test covers the real load/unload path. Co-Authored-By: Claude Opus 4.8 --- src/app/mod.rs | 13 +++ src/audio/echo_cancel.rs | 166 +++++++++++++++++++++++++++++++++++++++ src/audio/mod.rs | 1 + src/config.rs | 5 ++ src/core/messages.rs | 2 +- src/core/mod.rs | 47 ++++++++++- 6 files changed, 230 insertions(+), 4 deletions(-) create mode 100644 src/audio/echo_cancel.rs diff --git a/src/app/mod.rs b/src/app/mod.rs index 6f693e0..d9e582b 100644 --- a/src/app/mod.rs +++ b/src/app/mod.rs @@ -46,6 +46,7 @@ pub enum AppMessage { NavigateToSettings, NavigateBack, ToggleNotifications(bool), + ToggleEchoCancellation(bool), CustomSoundPathChanged(Sound, String), } @@ -196,6 +197,7 @@ fn update(state: &mut AppState, message: AppMessage) -> Task { ticket: state.ticket_input.clone(), input_device, output_device, + echo_cancellation: state.config.echo_cancellation_enabled, }); } } @@ -208,6 +210,7 @@ fn update(state: &mut AppState, message: AppMessage) -> Task { ticket: "create".to_string(), input_device, output_device, + echo_cancellation: state.config.echo_cancellation_enabled, }); } AppMessage::LeavePressed => { @@ -331,6 +334,11 @@ fn update(state: &mut AppState, message: AppMessage) -> Task { state.config.save(); notify::set_enabled(enabled); } + AppMessage::ToggleEchoCancellation(enabled) => { + state.config.echo_cancellation_enabled = enabled; + state.config.save(); + // Applied on the next join, since the audio graph is rebuilt then. + } AppMessage::CustomSoundPathChanged(sound, path) => { let path_opt = if path.trim().is_empty() { None } else { Some(path) }; match sound { @@ -515,6 +523,11 @@ fn view(state: &AppState) -> Element<'_, AppMessage> { text(format!("Mic Sensitivity (Noise Gate): {:.1}%", state.config.noise_gate_threshold * 100.0)).size(14).color(color_subtext), slider(0.0..=0.1, state.config.noise_gate_threshold, AppMessage::NoiseGateChanged).step(0.001), text("Smoothly fades out audio below this level. 0% disables the gate.").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), ].spacing(8).width(iced::Length::Fill), column![ text("Network Privacy").size(14).color(color_subtext), diff --git a/src/audio/echo_cancel.rs b/src/audio/echo_cancel.rs new file mode 100644 index 0000000..a278c06 --- /dev/null +++ b/src/audio/echo_cancel.rs @@ -0,0 +1,166 @@ +//! Acoustic echo cancellation via PipeWire's `module-echo-cancel`. +//! +//! Rather than running an echo canceller in-process (which would mean a C++ DSP +//! dependency plus plumbing the playout signal back to the capture path as a +//! delay-aligned reference), we delegate to the audio server. PipeWire already +//! sees both the real microphone and the speaker monitor, so loading +//! `module-echo-cancel` (WebRTC AEC + noise suppression + AGC) solves the +//! reference-alignment problem for free. +//! +//! When enabled, the module creates two virtual nodes: +//! - [`EC_SOURCE`] — the cleaned microphone (we capture from this). +//! - [`EC_SINK`] — playout goes here; it doubles as the AEC reference (we play to +//! this, and the module forwards it to the real output while using it to cancel +//! the echo from the mic). +//! +//! Binding to a specific mic/speaker (the user's chosen devices) is done with the +//! `source_master`/`sink_master` args; omitting them binds to the system defaults. +//! The loaded module is owned by an [`EchoCancelGuard`] that unloads it on drop, so +//! it never outlives a call (even if the join path bails out early). + +use std::process::Command; +use std::time::{Duration, Instant}; + +/// node.name of the virtual (cleaned) capture source the module creates. +pub const EC_SOURCE: &str = "peerspeak_echocancel_source"; +/// node.name of the virtual playback sink (also the AEC reference). +pub const EC_SINK: &str = "peerspeak_echocancel_sink"; + +/// How long to wait for the virtual nodes to appear after loading the module +/// before giving up — the nodes show up a beat after `load-module` returns. +const NODE_READY_TIMEOUT: Duration = Duration::from_secs(3); + +/// Owns a loaded `module-echo-cancel` instance; unloads it on drop so the virtual +/// nodes never leak past the call that created them. +pub struct EchoCancelGuard { + module_index: String, +} + +impl Drop for EchoCancelGuard { + fn drop(&mut self) { + let _ = Command::new("pactl") + .arg("unload-module") + .arg(&self.module_index) + .output(); + crate::log_msg(&format!("Echo cancel: unloaded module {}", self.module_index)); + } +} + +/// Loads `module-echo-cancel` (WebRTC AEC) bound to the given real devices, waits +/// for its virtual nodes to come up, and returns a guard that unloads it on drop. +/// +/// `real_source`/`real_sink` are the node.name of the chosen mic/speaker; pass +/// `None` (or an empty string) to bind to the system defaults. Returns `Err` with +/// a human-readable reason if `pactl` is missing, the load fails, or the nodes +/// don't appear — the caller should fall back to the direct devices. +pub fn enable(real_source: Option<&str>, real_sink: Option<&str>) -> Result { + // Best-effort: clear any stale instance left by a crashed prior run so we + // don't stack duplicate modules / fight over the virtual node names. + unload_stale(); + + let mut cmd = Command::new("pactl"); + cmd.arg("load-module") + .arg("module-echo-cancel") + .arg("aec_method=webrtc") + .arg(format!("source_name={EC_SOURCE}")) + .arg(format!("sink_name={EC_SINK}")); + if let Some(src) = real_source.filter(|s| !s.is_empty()) { + cmd.arg(format!("source_master={src}")); + } + if let Some(sink) = real_sink.filter(|s| !s.is_empty()) { + cmd.arg(format!("sink_master={sink}")); + } + + let out = cmd + .output() + .map_err(|e| format!("pactl not available: {e}"))?; + if !out.status.success() { + return Err(format!( + "pactl load-module failed: {}", + String::from_utf8_lossy(&out.stderr).trim() + )); + } + + let module_index = String::from_utf8_lossy(&out.stdout).trim().to_string(); + if module_index.parse::().is_err() { + return Err(format!("unexpected pactl output: {module_index:?}")); + } + let guard = EchoCancelGuard { module_index }; + + // The virtual nodes appear shortly after the module loads; wait for both so + // the subsequent capture/playback streams can actually target them. If they + // never show, drop the guard (unloads) and report failure. + if !wait_for_nodes() { + return Err("echo-cancel virtual nodes did not appear in time".to_string()); + } + + crate::log_msg(&format!( + "Echo cancel: loaded module {} (source_master={:?}, sink_master={:?})", + guard.module_index, real_source, real_sink + )); + Ok(guard) +} + +/// Polls until both virtual nodes exist or the timeout elapses. +fn wait_for_nodes() -> bool { + let deadline = Instant::now() + NODE_READY_TIMEOUT; + loop { + if node_present("sources", EC_SOURCE) && node_present("sinks", EC_SINK) { + return true; + } + if Instant::now() >= deadline { + return false; + } + std::thread::sleep(Duration::from_millis(100)); + } +} + +/// Whether `pactl list short` lists a node named `name`. +/// `kind` is "sources" or "sinks". +fn node_present(kind: &str, name: &str) -> bool { + let Ok(out) = Command::new("pactl").arg("list").arg(kind).arg("short").output() else { + return false; + }; + String::from_utf8_lossy(&out.stdout) + .lines() + .any(|line| line.split('\t').nth(1) == Some(name)) +} + +/// Unloads any leftover `module-echo-cancel` instance we previously created +/// (identified by our virtual node names in its argument string). Best-effort. +fn unload_stale() { + let Ok(out) = Command::new("pactl").arg("list").arg("modules").arg("short").output() else { + return; + }; + for line in String::from_utf8_lossy(&out.stdout).lines() { + let mut cols = line.split('\t'); + let index = cols.next().unwrap_or(""); + let name = cols.next().unwrap_or(""); + let args = cols.next().unwrap_or(""); + if name == "module-echo-cancel" && args.contains(EC_SOURCE) && index.parse::().is_ok() { + let _ = Command::new("pactl").arg("unload-module").arg(index).output(); + crate::log_msg(&format!("Echo cancel: cleaned up stale module {index}")); + } + } +} + +#[cfg(test)] +mod tests { + use super::*; + + /// Exercises the real load -> wait-for-nodes -> drop/unload path against the + /// live PipeWire daemon. Side-effecting (briefly creates virtual audio nodes), + /// so it's ignored by default; run with `cargo test -- --ignored echo_cancel`. + #[test] + #[ignore] + fn enable_creates_and_unloads_nodes() { + let guard = enable(None, None).expect("module-echo-cancel should load"); + assert!(node_present("sources", EC_SOURCE), "cleaned source must exist"); + assert!(node_present("sinks", EC_SINK), "reference sink must exist"); + drop(guard); + // Give pactl a moment to tear the nodes down. + std::thread::sleep(Duration::from_millis(300)); + assert!(!node_present("sources", EC_SOURCE), "source must be gone after unload"); + assert!(!node_present("sinks", EC_SINK), "sink must be gone after unload"); + } +} diff --git a/src/audio/mod.rs b/src/audio/mod.rs index 5ff8a30..935a990 100644 --- a/src/audio/mod.rs +++ b/src/audio/mod.rs @@ -51,6 +51,7 @@ pub trait AudioBackend: Send + Sync { fn stop(&self) -> Result<(), AudioError>; } +pub mod echo_cancel; pub mod gate; pub mod pipewire_impl; pub mod pw_cli; diff --git a/src/config.rs b/src/config.rs index e1bca9c..6a1f0c5 100644 --- a/src/config.rs +++ b/src/config.rs @@ -47,6 +47,10 @@ pub struct AppConfig { pub noise_gate_threshold: f32, #[serde(default)] pub network_mode: NetworkMode, + /// Route audio through PipeWire's echo-cancel module (AEC + noise suppression). + /// Takes effect on the next room join. Off by default. + #[serde(default)] + pub echo_cancellation_enabled: bool, #[serde(default = "default_true")] pub notifications_enabled: bool, #[serde(default)] @@ -74,6 +78,7 @@ impl Default for AppConfig { output_device: "".to_string(), noise_gate_threshold: 0.01, network_mode: NetworkMode::default(), + echo_cancellation_enabled: false, notifications_enabled: true, custom_sound_self_join: None, custom_sound_peer_join: None, diff --git a/src/core/messages.rs b/src/core/messages.rs index 2eee7af..4ab8ed2 100644 --- a/src/core/messages.rs +++ b/src/core/messages.rs @@ -4,7 +4,7 @@ use iroh::EndpointId; #[derive(Debug, Clone)] pub enum CoreCommand { - Join { name: String, ticket: String, input_device: Option, output_device: Option }, + Join { name: String, ticket: String, input_device: Option, output_device: Option, echo_cancellation: bool }, Leave, ToggleMute, ToggleDeafen, diff --git a/src/core/mod.rs b/src/core/mod.rs index 7d04c05..8afc34d 100644 --- a/src/core/mod.rs +++ b/src/core/mod.rs @@ -203,6 +203,8 @@ struct ActiveSession { conn_event_task: tokio::task::JoinHandle<()>, grace_timers: GraceTimers, transport: Arc, + /// Loaded PipeWire echo-cancel module (if enabled); unloads on drop. + echo_cancel: Option, } impl ActiveSession { @@ -226,6 +228,10 @@ impl ActiveSession { crate::log_msg("Audio backend stopped"); }).await; + // Unload the echo-cancel module now that the audio streams releasing its + // virtual nodes have stopped. (Dropping the guard runs `pactl unload`.) + drop(self.echo_cancel); + crate::log_msg("Leaving room..."); let _ = self.room_state.leave().await; // Close peer links with the graceful goodbye code so remotes evict us @@ -266,7 +272,7 @@ async fn run_core_loop( while let Some(cmd) = cmd_rx.recv().await { match cmd { - CoreCommand::Join { name, ticket, input_device, output_device } => { + CoreCommand::Join { name, ticket, input_device, output_device, echo_cancellation } => { current_name = name.clone(); // Clean up any existing session @@ -363,7 +369,41 @@ async fn run_core_loop( let (capture_tx, capture_rx) = std::sync::mpsc::channel(); let (playback_tx, playback_rx) = std::sync::mpsc::channel(); - if let Err(e) = audio_backend.start_capture(capture_tx, input_device) { + // Echo cancellation: if enabled, load PipeWire's echo-cancel module + // bound to the chosen real devices and route capture/playback + // through its virtual nodes (the sink doubles as the AEC reference). + // 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. + let mut echo_cancel_guard = None; + let (capture_target, playback_target) = if echo_cancellation { + match crate::audio::echo_cancel::enable( + input_device.as_deref(), + output_device.as_deref(), + ) { + Ok(guard) => { + echo_cancel_guard = Some(guard); + crate::log_msg("Echo cancellation enabled"); + ( + Some(crate::audio::echo_cancel::EC_SOURCE.to_string()), + Some(crate::audio::echo_cancel::EC_SINK.to_string()), + ) + } + Err(e) => { + crate::log_msg(&format!( + "Echo cancellation unavailable, using direct devices: {e}" + )); + let _ = ui_tx + .send(UiEvent::Error(format!("Echo cancellation unavailable: {e}"))) + .await; + (input_device.clone(), output_device.clone()) + } + } + } else { + (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; let _ = room_state.leave().await; let _ = router.shutdown().await; @@ -374,7 +414,7 @@ async fn run_core_loop( // here (drain side + fill side); the mixer reads it to pace // production to the hardware clock instead of a fixed timer. let ring_fill = Arc::new(AtomicUsize::new(0)); - if let Err(e) = audio_backend.start_playback(playback_rx, output_device, ring_fill.clone()) { + if let Err(e) = audio_backend.start_playback(playback_rx, playback_target, ring_fill.clone()) { let _ = ui_tx.send(UiEvent::Error(format!("Failed to start playback: {}", e))).await; let _ = audio_backend.stop(); let _ = room_state.leave().await; @@ -675,6 +715,7 @@ async fn run_core_loop( conn_event_task, grace_timers, transport: transport.clone(), + echo_cancel: echo_cancel_guard, }; let self_id = endpoint.id().to_string();