From 405b005a0d0c59959cbd498395de918f9e725492 Mon Sep 17 00:00:00 2001 From: enricobuehler Date: Thu, 16 Jul 2026 15:21:36 +0200 Subject: [PATCH] =?UTF-8?q?refactor(host):=20extract=20audio/mic=5Fpump.rs?= =?UTF-8?q?=20=E2=80=94=20the=20host-lifetime=20virtual-mic=20pump?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Per plan §2.1: a self-contained stateful subsystem does not belong in the audio trait facade. Move MicPump + its PumpTuning/PUMP_TUNING, the drain_sleep/pump_thread loop, MIC_CHANNELS/MIC_QUEUE_CAP, and the six pump unit tests out of audio.rs into audio/mic_pump.rs. audio.rs keeps the AudioCapturer/VirtualMic traits, their open_* factories, and the sample constants. Re-export via `pub use mic_pump::MicPump` so crate::audio::MicPump stays byte-stable (only consumer: punktfunk1.rs). Pure code-move; verified clippy 0/0 + 6/6 pump tests green on Linux (home-worker-5). Co-Authored-By: Claude Opus 4.8 (1M context) --- crates/punktfunk-host/src/audio.rs | 424 +------------------ crates/punktfunk-host/src/audio/mic_pump.rs | 432 ++++++++++++++++++++ 2 files changed, 434 insertions(+), 422 deletions(-) create mode 100644 crates/punktfunk-host/src/audio/mic_pump.rs diff --git a/crates/punktfunk-host/src/audio.rs b/crates/punktfunk-host/src/audio.rs index 3523442a..21d1ab68 100644 --- a/crates/punktfunk-host/src/audio.rs +++ b/crates/punktfunk-host/src/audio.rs @@ -106,220 +106,6 @@ pub fn open_virtual_mic(_channels: u32) -> Result> { anyhow::bail!("virtual mic requires Linux + PipeWire or Windows + a virtual audio device") } -/// Mic is 48 kHz stereo — matches the Opus stereo decoder and the host→client audio layout. -pub const MIC_CHANNELS: u32 = 2; -/// Bound for the shared mic frame queue (drop-newest when full): the host-lifetime queue is -/// shared across all concurrent sessions and must not grow without limit under a near-line-rate -/// flood (security-review 2026-06-28 S6). 64 × 5–20 ms frames ≈ 0.3–1.3 s of slack. -const MIC_QUEUE_CAP: usize = 64; - -/// Tuning for [`MicPump`]'s open/reopen/flush behaviour — parameterized so the tests can run the -/// real pump loop in milliseconds instead of seconds. -#[derive(Clone, Copy)] -struct PumpTuning { - /// First-retry delay after a failed backend open; doubles per failure up to `backoff_cap` - /// (a persistently-absent PipeWire session / audio endpoint isn't hammered), resets on - /// success. - backoff_start: std::time::Duration, - backoff_cap: std::time::Duration, - /// Idle liveness-probe interval: with no frames flowing, the pump still notices a dead - /// backend this often and reopens — so the mic is healthy BEFORE the next session starts. - heartbeat: std::time::Duration, - /// An uplink gap longer than this discards the backend's buffered audio before pushing the - /// next frame (a recorder must never hear a stale burst from before a mute/session end). - stale_gap: std::time::Duration, - /// A backend that dies before living this long counts as a FAILED open for backoff purposes - /// (an open that succeeds but dies instantly — e.g. a flapping daemon — must not churn at - /// heartbeat rate); one that lived longer resets the backoff. - stable_after: std::time::Duration, -} - -const PUMP_TUNING: PumpTuning = PumpTuning { - backoff_start: std::time::Duration::from_secs(2), - backoff_cap: std::time::Duration::from_secs(60), - heartbeat: std::time::Duration::from_secs(1), - stale_gap: std::time::Duration::from_millis(600), - stable_after: std::time::Duration::from_secs(5), -}; - -/// Host-lifetime virtual-microphone pump: one thread owns the [`VirtualMic`] backend + an Opus -/// decoder; sessions forward the client's Opus mic frames (0xCB) over a clonable `Send` sender, -/// the thread decodes and feeds the backend. -/// -/// The rock-solid properties live HERE, not in the backends: -/// - **Eager**: the backend opens at host start (retrying with backoff), NOT on the first mic -/// frame — so the virtual mic device already exists when host apps/games launch and bind -/// their capture device (most games never re-follow a default-device change mid-run). -/// - **Self-healing**: a dead backend (PipeWire restart, Windows endpoint churn) is detected on -/// every push and on an idle heartbeat, and reopened with backoff. Sessions keep their -/// senders; nothing upstream notices. -/// - **Stale-flush**: buffered audio is discarded after an uplink gap (see [`PumpTuning`]). -/// -/// Per-frame Opus DECODE errors stay non-fatal (dropped frame): the mic is shared across every -/// concurrent session, so one paired client's junk frames must not deny everyone's mic -/// (security-review 2026-06-28 S2). The thread exits when every sender is dropped (host -/// shutdown), tearing the backend down. -pub struct MicPump { - tx: std::sync::mpsc::SyncSender>, -} - -impl MicPump { - /// Start the host-lifetime pump (Linux/Windows). On platforms without a virtual-mic backend - /// the thread just drains and drops frames (sessions still count the datagrams). - pub fn start() -> MicPump { - let (tx, rx) = std::sync::mpsc::sync_channel::>(MIC_QUEUE_CAP); - let spawned = std::thread::Builder::new() - .name("punktfunk-mic-pump".into()) - .spawn(move || { - #[cfg(any(target_os = "linux", target_os = "windows"))] - pump_thread(rx, || open_virtual_mic(MIC_CHANNELS), PUMP_TUNING); - #[cfg(not(any(target_os = "linux", target_os = "windows")))] - { - tracing::warn!("mic passthrough unsupported on this platform — frames dropped"); - for _ in rx {} - } - }); - if let Err(e) = spawned { - tracing::error!(error = %e, "mic pump thread spawn failed — mic passthrough disabled"); - } - MicPump { tx } - } - - /// A sender a session forwards the client's Opus mic frames to (`try_send` — never block a - /// datagram loop). Cloned per session; dropping a clone does NOT stop the pump (it holds - /// the original sender for the host life). - pub fn sender(&self) -> std::sync::mpsc::SyncSender> { - self.tx.clone() - } -} - -/// Sleep for `dur` while draining (and dropping) queued frames, so a closed/reopening backend -/// never accumulates a stale backlog and senders never see a wedged queue. Returns `false` when -/// every sender is gone (host shutdown). -#[cfg_attr(not(any(target_os = "linux", target_os = "windows")), allow(dead_code))] -fn drain_sleep(rx: &std::sync::mpsc::Receiver>, dur: std::time::Duration) -> bool { - use std::sync::mpsc::RecvTimeoutError; - let deadline = std::time::Instant::now() + dur; - loop { - let left = deadline.saturating_duration_since(std::time::Instant::now()); - if left.is_zero() { - return true; - } - match rx.recv_timeout(left.min(std::time::Duration::from_millis(250))) { - Ok(_) => {} // drop frames while closed - Err(RecvTimeoutError::Timeout) => {} // keep waiting - Err(RecvTimeoutError::Disconnected) => return false, // host shutdown - } - } -} - -/// The pump loop. `opener` is injected so the tests can run the REAL loop against a mock -/// backend; production passes [`open_virtual_mic`]. -#[cfg_attr(not(any(target_os = "linux", target_os = "windows")), allow(dead_code))] -fn pump_thread(rx: std::sync::mpsc::Receiver>, opener: O, tuning: PumpTuning) -where - O: Fn() -> Result>, -{ - use std::sync::mpsc::RecvTimeoutError; - use std::time::Instant; - - let mut backoff = tuning.backoff_start; - let mut open_fails: u64 = 0; - loop { - // Open phase — eager, from thread start. - let (mic, mut decoder) = loop { - let opened = opener().and_then(|m| { - let d = opus::Decoder::new(SAMPLE_RATE, opus::Channels::Stereo) - .map_err(|e| anyhow::anyhow!("opus decoder: {e}"))?; - Ok((m, d)) - }); - match opened { - Ok(pair) => break pair, - Err(e) => { - // Throttle (1st, 2nd, 4th, 8th … failure): a box without a PipeWire session - // or virtual audio device would otherwise log every backoff forever. - open_fails += 1; - if open_fails.is_power_of_two() { - tracing::warn!(error = %format!("{e:#}"), attempts = open_fails, - "virtual mic unavailable — retrying with backoff"); - } - if !drain_sleep(&rx, backoff) { - return; - } - backoff = (backoff * 2).min(tuning.backoff_cap); - } - } - }; - tracing::info!("virtual mic ready (host-lifetime)"); - // Drop anything queued while (re)opening — it predates the backend. (The backoff does - // NOT reset here: only an instance that proves stable resets it — see the death triage.) - while rx.try_recv().is_ok() {} - let opened_at = Instant::now(); - - // Pump phase — runs until the backend dies (break) or the host shuts down (return). - let mut decode_fails: u64 = 0; - let mut pcm = vec![0f32; 5760 * MIC_CHANNELS as usize]; // up to 120 ms scratch - let mut last_push = Instant::now(); - loop { - match rx.recv_timeout(tuning.heartbeat) { - Ok(frame) => { - if frame.is_empty() { - continue; // DTX silence — the source underruns to silence on its own - } - if last_push.elapsed() > tuning.stale_gap { - mic.discard(); - } - match decoder.decode_float(&frame, &mut pcm, false) { - Ok(samples_per_ch) => { - let total = (samples_per_ch * MIC_CHANNELS as usize).min(pcm.len()); - if !mic.push(&pcm[..total]) { - tracing::warn!("virtual mic backend died — reopening"); - break; - } - last_push = Instant::now(); - decode_fails = 0; - } - Err(e) => { - // Malformed/garbage frame: drop it, keep the shared mic + decoder - // (see the struct docs). Throttled log (1, 2, 4, … fails). - decode_fails += 1; - if decode_fails.is_power_of_two() { - tracing::warn!(error = %e, fails = decode_fails, - "mic opus decode failed — dropping frame"); - } - } - } - } - Err(RecvTimeoutError::Timeout) => { - if !mic.alive() { - tracing::warn!("virtual mic backend died while idle — reopening"); - break; - } - } - Err(RecvTimeoutError::Disconnected) => { - tracing::debug!("mic pump stopped (host shutting down)"); - return; - } - } - } - - // Death triage: an instance that lived is a one-off (PipeWire/audio-engine restart) — - // reopen immediately with the backoff reset. One that died right after opening is a - // failed open in disguise (flapping daemon, endpoint racing away): back off like the - // open loop, or the pump would churn open→die→reopen at heartbeat rate. - if opened_at.elapsed() >= tuning.stable_after { - backoff = tuning.backoff_start; - open_fails = 0; - } else { - open_fails += 1; - if !drain_sleep(&rx, backoff) { - return; - } - backoff = (backoff * 2).min(tuning.backoff_cap); - } - } -} - #[cfg(target_os = "windows")] #[path = "audio/windows/audio_control.rs"] mod audio_control; @@ -335,211 +121,5 @@ mod wasapi_mic; #[path = "audio/wiring_plan.rs"] pub(crate) mod wiring_plan; -#[cfg(test)] -mod pump_tests { - use super::*; - use std::sync::atomic::{AtomicBool, AtomicUsize, Ordering}; - use std::sync::{Arc, Mutex}; - use std::time::Duration; - - /// Mock backend: records pushes/discards, dies on command. - struct MockMic { - alive: Arc, - pushed: Arc, - discards: Arc, - } - impl VirtualMic for MockMic { - fn push(&self, pcm: &[f32]) -> bool { - if !self.alive.load(Ordering::Acquire) { - return false; - } - self.pushed.fetch_add(pcm.len(), Ordering::Relaxed); - true - } - fn alive(&self) -> bool { - self.alive.load(Ordering::Acquire) - } - fn discard(&self) { - self.discards.fetch_add(1, Ordering::Relaxed); - } - } - - struct Harness { - tx: std::sync::mpsc::SyncSender>, - opens: Arc, - alive: Arc>>>, // latest instance's kill switch - pushed: Arc, - discards: Arc, - join: std::thread::JoinHandle<()>, - } - - /// Run the REAL pump loop against mock backends; `fail_first` opens fail before the first - /// success (exercises the eager retry/backoff path). `dead_on_arrival` opens every instance - /// pre-killed (exercises the rapid-death churn guard). `stable_after` mirrors the tuning - /// field (ZERO = every death counts as stable → immediate reopen, keeping tests fast). - fn start_tuned(fail_first: usize, dead_on_arrival: bool, stable_after: Duration) -> Harness { - let (tx, rx) = std::sync::mpsc::sync_channel::>(MIC_QUEUE_CAP); - let opens = Arc::new(AtomicUsize::new(0)); - let alive = Arc::new(Mutex::new(None::>)); - let pushed = Arc::new(AtomicUsize::new(0)); - let discards = Arc::new(AtomicUsize::new(0)); - let (opens2, alive2, pushed2, discards2) = ( - opens.clone(), - alive.clone(), - pushed.clone(), - discards.clone(), - ); - let tuning = PumpTuning { - backoff_start: Duration::from_millis(10), - backoff_cap: Duration::from_millis(40), - heartbeat: Duration::from_millis(20), - stale_gap: Duration::from_millis(80), - stable_after, - }; - let join = std::thread::spawn(move || { - pump_thread( - rx, - move || { - let n = opens2.fetch_add(1, Ordering::SeqCst); - if n < fail_first { - anyhow::bail!("backend not up yet (simulated)"); - } - let a = Arc::new(AtomicBool::new(!dead_on_arrival)); - *alive2.lock().unwrap() = Some(a.clone()); - Ok(Box::new(MockMic { - alive: a, - pushed: pushed2.clone(), - discards: discards2.clone(), - }) as Box) - }, - tuning, - ) - }); - Harness { - tx, - opens, - alive, - pushed, - discards, - join, - } - } - - fn start(fail_first: usize) -> Harness { - start_tuned(fail_first, false, Duration::ZERO) - } - - fn wait_until(what: &str, mut cond: impl FnMut() -> bool) { - for _ in 0..200 { - if cond() { - return; - } - std::thread::sleep(Duration::from_millis(10)); - } - panic!("timed out waiting for: {what}"); - } - - fn opus_frame() -> Vec { - let mut enc = opus::Encoder::new(48_000, opus::Channels::Stereo, opus::Application::Voip) - .expect("opus encoder"); - let pcm = [0.1f32; 960 * 2]; // 20 ms stereo - let mut out = vec![0u8; 4000]; - let n = enc.encode_float(&pcm, &mut out).expect("encode"); - out.truncate(n); - out - } - - /// Eager: the backend opens (after transient failures) with NO frame ever sent. - #[test] - fn opens_eagerly_with_backoff() { - let h = start(3); - wait_until("eager open after 3 failures", || { - h.opens.load(Ordering::SeqCst) >= 4 && h.alive.lock().unwrap().is_some() - }); - drop(h.tx); - h.join.join().unwrap(); - } - - /// Frames flow: opus in → PCM pushed to the backend. - #[test] - fn decodes_and_pushes() { - let h = start(0); - wait_until("open", || h.alive.lock().unwrap().is_some()); - h.tx.send(opus_frame()).unwrap(); - wait_until("pcm pushed", || h.pushed.load(Ordering::SeqCst) > 0); - drop(h.tx); - h.join.join().unwrap(); - } - - /// A dead backend is noticed WHILE IDLE (heartbeat) and reopened without any traffic. - #[test] - fn reopens_after_idle_death() { - let h = start(0); - wait_until("first open", || h.opens.load(Ordering::SeqCst) >= 1); - wait_until("instance", || h.alive.lock().unwrap().is_some()); - h.alive - .lock() - .unwrap() - .as_ref() - .unwrap() - .store(false, Ordering::Release); // kill it - wait_until("reopen after idle death", || { - h.opens.load(Ordering::SeqCst) >= 2 - }); - drop(h.tx); - h.join.join().unwrap(); - } - - /// A death detected on push (frame flowing) also reopens, and the frame after reopen flows. - #[test] - fn reopens_after_push_death() { - let h = start(0); - wait_until("instance", || h.alive.lock().unwrap().is_some()); - h.alive - .lock() - .unwrap() - .as_ref() - .unwrap() - .store(false, Ordering::Release); - h.tx.send(opus_frame()).unwrap(); // push sees death → reopen - wait_until("reopen", || h.opens.load(Ordering::SeqCst) >= 2); - h.tx.send(opus_frame()).unwrap(); - wait_until("pcm after reopen", || h.pushed.load(Ordering::SeqCst) > 0); - drop(h.tx); - h.join.join().unwrap(); - } - - /// Instances that die immediately after opening must be retried with BACKOFF, not at - /// heartbeat rate — a flapping backend (daemon up but dropping us instantly) would - /// otherwise churn open→die→reopen every heartbeat forever. - #[test] - fn rapid_death_backs_off() { - // Every instance is dead on arrival; stability threshold high so each death counts - // as a failed open. Without the guard: ~1 reopen per heartbeat (20 ms) ≈ 25 opens in - // 500 ms. With backoff 10→20→40 (cap): ≈ 7. - let h = start_tuned(0, true, Duration::from_secs(10)); - std::thread::sleep(Duration::from_millis(500)); - let opens = h.opens.load(Ordering::SeqCst); - assert!(opens >= 2, "must keep retrying (got {opens})"); - assert!( - opens <= 15, - "must back off, not churn per heartbeat (got {opens})" - ); - drop(h.tx); - h.join.join().unwrap(); - } - - /// An uplink gap discards buffered-stale audio before the next frame plays. - #[test] - fn discards_after_gap() { - let h = start(0); - wait_until("instance", || h.alive.lock().unwrap().is_some()); - h.tx.send(opus_frame()).unwrap(); - wait_until("first push", || h.pushed.load(Ordering::SeqCst) > 0); - std::thread::sleep(Duration::from_millis(150)); // > stale_gap - h.tx.send(opus_frame()).unwrap(); - wait_until("discard on gap", || h.discards.load(Ordering::SeqCst) >= 1); - drop(h.tx); - h.join.join().unwrap(); - } -} +mod mic_pump; +pub use mic_pump::MicPump; diff --git a/crates/punktfunk-host/src/audio/mic_pump.rs b/crates/punktfunk-host/src/audio/mic_pump.rs new file mode 100644 index 00000000..e09b9e12 --- /dev/null +++ b/crates/punktfunk-host/src/audio/mic_pump.rs @@ -0,0 +1,432 @@ +//! Host-lifetime virtual-microphone pump: one thread owns the [`VirtualMic`] backend plus an Opus +//! decoder, self-heals across backend deaths, and feeds decoded client-mic PCM into the source so +//! the client's microphone reaches host apps. Split out of the `audio` facade (§2.1 — a +//! self-contained stateful subsystem does not belong in the trait facade); the [`VirtualMic`] +//! trait, its factory ([`open_virtual_mic`](super::open_virtual_mic)) and the audio-plane sample +//! rate stay in `super`. + +use super::{VirtualMic, SAMPLE_RATE}; +use anyhow::Result; + +/// Mic is 48 kHz stereo — matches the Opus stereo decoder and the host→client audio layout. +pub const MIC_CHANNELS: u32 = 2; +/// Bound for the shared mic frame queue (drop-newest when full): the host-lifetime queue is +/// shared across all concurrent sessions and must not grow without limit under a near-line-rate +/// flood (security-review 2026-06-28 S6). 64 × 5–20 ms frames ≈ 0.3–1.3 s of slack. +const MIC_QUEUE_CAP: usize = 64; + +/// Tuning for [`MicPump`]'s open/reopen/flush behaviour — parameterized so the tests can run the +/// real pump loop in milliseconds instead of seconds. +#[derive(Clone, Copy)] +struct PumpTuning { + /// First-retry delay after a failed backend open; doubles per failure up to `backoff_cap` + /// (a persistently-absent PipeWire session / audio endpoint isn't hammered), resets on + /// success. + backoff_start: std::time::Duration, + backoff_cap: std::time::Duration, + /// Idle liveness-probe interval: with no frames flowing, the pump still notices a dead + /// backend this often and reopens — so the mic is healthy BEFORE the next session starts. + heartbeat: std::time::Duration, + /// An uplink gap longer than this discards the backend's buffered audio before pushing the + /// next frame (a recorder must never hear a stale burst from before a mute/session end). + stale_gap: std::time::Duration, + /// A backend that dies before living this long counts as a FAILED open for backoff purposes + /// (an open that succeeds but dies instantly — e.g. a flapping daemon — must not churn at + /// heartbeat rate); one that lived longer resets the backoff. + stable_after: std::time::Duration, +} + +const PUMP_TUNING: PumpTuning = PumpTuning { + backoff_start: std::time::Duration::from_secs(2), + backoff_cap: std::time::Duration::from_secs(60), + heartbeat: std::time::Duration::from_secs(1), + stale_gap: std::time::Duration::from_millis(600), + stable_after: std::time::Duration::from_secs(5), +}; + +/// Host-lifetime virtual-microphone pump: one thread owns the [`VirtualMic`] backend + an Opus +/// decoder; sessions forward the client's Opus mic frames (0xCB) over a clonable `Send` sender, +/// the thread decodes and feeds the backend. +/// +/// The rock-solid properties live HERE, not in the backends: +/// - **Eager**: the backend opens at host start (retrying with backoff), NOT on the first mic +/// frame — so the virtual mic device already exists when host apps/games launch and bind +/// their capture device (most games never re-follow a default-device change mid-run). +/// - **Self-healing**: a dead backend (PipeWire restart, Windows endpoint churn) is detected on +/// every push and on an idle heartbeat, and reopened with backoff. Sessions keep their +/// senders; nothing upstream notices. +/// - **Stale-flush**: buffered audio is discarded after an uplink gap (see [`PumpTuning`]). +/// +/// Per-frame Opus DECODE errors stay non-fatal (dropped frame): the mic is shared across every +/// concurrent session, so one paired client's junk frames must not deny everyone's mic +/// (security-review 2026-06-28 S2). The thread exits when every sender is dropped (host +/// shutdown), tearing the backend down. +pub struct MicPump { + tx: std::sync::mpsc::SyncSender>, +} + +impl MicPump { + /// Start the host-lifetime pump (Linux/Windows). On platforms without a virtual-mic backend + /// the thread just drains and drops frames (sessions still count the datagrams). + pub fn start() -> MicPump { + let (tx, rx) = std::sync::mpsc::sync_channel::>(MIC_QUEUE_CAP); + let spawned = std::thread::Builder::new() + .name("punktfunk-mic-pump".into()) + .spawn(move || { + #[cfg(any(target_os = "linux", target_os = "windows"))] + pump_thread(rx, || super::open_virtual_mic(MIC_CHANNELS), PUMP_TUNING); + #[cfg(not(any(target_os = "linux", target_os = "windows")))] + { + tracing::warn!("mic passthrough unsupported on this platform — frames dropped"); + for _ in rx {} + } + }); + if let Err(e) = spawned { + tracing::error!(error = %e, "mic pump thread spawn failed — mic passthrough disabled"); + } + MicPump { tx } + } + + /// A sender a session forwards the client's Opus mic frames to (`try_send` — never block a + /// datagram loop). Cloned per session; dropping a clone does NOT stop the pump (it holds + /// the original sender for the host life). + pub fn sender(&self) -> std::sync::mpsc::SyncSender> { + self.tx.clone() + } +} + +/// Sleep for `dur` while draining (and dropping) queued frames, so a closed/reopening backend +/// never accumulates a stale backlog and senders never see a wedged queue. Returns `false` when +/// every sender is gone (host shutdown). +#[cfg_attr(not(any(target_os = "linux", target_os = "windows")), allow(dead_code))] +fn drain_sleep(rx: &std::sync::mpsc::Receiver>, dur: std::time::Duration) -> bool { + use std::sync::mpsc::RecvTimeoutError; + let deadline = std::time::Instant::now() + dur; + loop { + let left = deadline.saturating_duration_since(std::time::Instant::now()); + if left.is_zero() { + return true; + } + match rx.recv_timeout(left.min(std::time::Duration::from_millis(250))) { + Ok(_) => {} // drop frames while closed + Err(RecvTimeoutError::Timeout) => {} // keep waiting + Err(RecvTimeoutError::Disconnected) => return false, // host shutdown + } + } +} + +/// The pump loop. `opener` is injected so the tests can run the REAL loop against a mock +/// backend; production passes [`open_virtual_mic`](super::open_virtual_mic). +#[cfg_attr(not(any(target_os = "linux", target_os = "windows")), allow(dead_code))] +fn pump_thread(rx: std::sync::mpsc::Receiver>, opener: O, tuning: PumpTuning) +where + O: Fn() -> Result>, +{ + use std::sync::mpsc::RecvTimeoutError; + use std::time::Instant; + + let mut backoff = tuning.backoff_start; + let mut open_fails: u64 = 0; + loop { + // Open phase — eager, from thread start. + let (mic, mut decoder) = loop { + let opened = opener().and_then(|m| { + let d = opus::Decoder::new(SAMPLE_RATE, opus::Channels::Stereo) + .map_err(|e| anyhow::anyhow!("opus decoder: {e}"))?; + Ok((m, d)) + }); + match opened { + Ok(pair) => break pair, + Err(e) => { + // Throttle (1st, 2nd, 4th, 8th … failure): a box without a PipeWire session + // or virtual audio device would otherwise log every backoff forever. + open_fails += 1; + if open_fails.is_power_of_two() { + tracing::warn!(error = %format!("{e:#}"), attempts = open_fails, + "virtual mic unavailable — retrying with backoff"); + } + if !drain_sleep(&rx, backoff) { + return; + } + backoff = (backoff * 2).min(tuning.backoff_cap); + } + } + }; + tracing::info!("virtual mic ready (host-lifetime)"); + // Drop anything queued while (re)opening — it predates the backend. (The backoff does + // NOT reset here: only an instance that proves stable resets it — see the death triage.) + while rx.try_recv().is_ok() {} + let opened_at = Instant::now(); + + // Pump phase — runs until the backend dies (break) or the host shuts down (return). + let mut decode_fails: u64 = 0; + let mut pcm = vec![0f32; 5760 * MIC_CHANNELS as usize]; // up to 120 ms scratch + let mut last_push = Instant::now(); + loop { + match rx.recv_timeout(tuning.heartbeat) { + Ok(frame) => { + if frame.is_empty() { + continue; // DTX silence — the source underruns to silence on its own + } + if last_push.elapsed() > tuning.stale_gap { + mic.discard(); + } + match decoder.decode_float(&frame, &mut pcm, false) { + Ok(samples_per_ch) => { + let total = (samples_per_ch * MIC_CHANNELS as usize).min(pcm.len()); + if !mic.push(&pcm[..total]) { + tracing::warn!("virtual mic backend died — reopening"); + break; + } + last_push = Instant::now(); + decode_fails = 0; + } + Err(e) => { + // Malformed/garbage frame: drop it, keep the shared mic + decoder + // (see the struct docs). Throttled log (1, 2, 4, … fails). + decode_fails += 1; + if decode_fails.is_power_of_two() { + tracing::warn!(error = %e, fails = decode_fails, + "mic opus decode failed — dropping frame"); + } + } + } + } + Err(RecvTimeoutError::Timeout) => { + if !mic.alive() { + tracing::warn!("virtual mic backend died while idle — reopening"); + break; + } + } + Err(RecvTimeoutError::Disconnected) => { + tracing::debug!("mic pump stopped (host shutting down)"); + return; + } + } + } + + // Death triage: an instance that lived is a one-off (PipeWire/audio-engine restart) — + // reopen immediately with the backoff reset. One that died right after opening is a + // failed open in disguise (flapping daemon, endpoint racing away): back off like the + // open loop, or the pump would churn open→die→reopen at heartbeat rate. + if opened_at.elapsed() >= tuning.stable_after { + backoff = tuning.backoff_start; + open_fails = 0; + } else { + open_fails += 1; + if !drain_sleep(&rx, backoff) { + return; + } + backoff = (backoff * 2).min(tuning.backoff_cap); + } + } +} + +#[cfg(test)] +mod pump_tests { + use super::*; + use std::sync::atomic::{AtomicBool, AtomicUsize, Ordering}; + use std::sync::{Arc, Mutex}; + use std::time::Duration; + + /// Mock backend: records pushes/discards, dies on command. + struct MockMic { + alive: Arc, + pushed: Arc, + discards: Arc, + } + impl VirtualMic for MockMic { + fn push(&self, pcm: &[f32]) -> bool { + if !self.alive.load(Ordering::Acquire) { + return false; + } + self.pushed.fetch_add(pcm.len(), Ordering::Relaxed); + true + } + fn alive(&self) -> bool { + self.alive.load(Ordering::Acquire) + } + fn discard(&self) { + self.discards.fetch_add(1, Ordering::Relaxed); + } + } + + struct Harness { + tx: std::sync::mpsc::SyncSender>, + opens: Arc, + alive: Arc>>>, // latest instance's kill switch + pushed: Arc, + discards: Arc, + join: std::thread::JoinHandle<()>, + } + + /// Run the REAL pump loop against mock backends; `fail_first` opens fail before the first + /// success (exercises the eager retry/backoff path). `dead_on_arrival` opens every instance + /// pre-killed (exercises the rapid-death churn guard). `stable_after` mirrors the tuning + /// field (ZERO = every death counts as stable → immediate reopen, keeping tests fast). + fn start_tuned(fail_first: usize, dead_on_arrival: bool, stable_after: Duration) -> Harness { + let (tx, rx) = std::sync::mpsc::sync_channel::>(MIC_QUEUE_CAP); + let opens = Arc::new(AtomicUsize::new(0)); + let alive = Arc::new(Mutex::new(None::>)); + let pushed = Arc::new(AtomicUsize::new(0)); + let discards = Arc::new(AtomicUsize::new(0)); + let (opens2, alive2, pushed2, discards2) = ( + opens.clone(), + alive.clone(), + pushed.clone(), + discards.clone(), + ); + let tuning = PumpTuning { + backoff_start: Duration::from_millis(10), + backoff_cap: Duration::from_millis(40), + heartbeat: Duration::from_millis(20), + stale_gap: Duration::from_millis(80), + stable_after, + }; + let join = std::thread::spawn(move || { + pump_thread( + rx, + move || { + let n = opens2.fetch_add(1, Ordering::SeqCst); + if n < fail_first { + anyhow::bail!("backend not up yet (simulated)"); + } + let a = Arc::new(AtomicBool::new(!dead_on_arrival)); + *alive2.lock().unwrap() = Some(a.clone()); + Ok(Box::new(MockMic { + alive: a, + pushed: pushed2.clone(), + discards: discards2.clone(), + }) as Box) + }, + tuning, + ) + }); + Harness { + tx, + opens, + alive, + pushed, + discards, + join, + } + } + + fn start(fail_first: usize) -> Harness { + start_tuned(fail_first, false, Duration::ZERO) + } + + fn wait_until(what: &str, mut cond: impl FnMut() -> bool) { + for _ in 0..200 { + if cond() { + return; + } + std::thread::sleep(Duration::from_millis(10)); + } + panic!("timed out waiting for: {what}"); + } + + fn opus_frame() -> Vec { + let mut enc = opus::Encoder::new(48_000, opus::Channels::Stereo, opus::Application::Voip) + .expect("opus encoder"); + let pcm = [0.1f32; 960 * 2]; // 20 ms stereo + let mut out = vec![0u8; 4000]; + let n = enc.encode_float(&pcm, &mut out).expect("encode"); + out.truncate(n); + out + } + + /// Eager: the backend opens (after transient failures) with NO frame ever sent. + #[test] + fn opens_eagerly_with_backoff() { + let h = start(3); + wait_until("eager open after 3 failures", || { + h.opens.load(Ordering::SeqCst) >= 4 && h.alive.lock().unwrap().is_some() + }); + drop(h.tx); + h.join.join().unwrap(); + } + + /// Frames flow: opus in → PCM pushed to the backend. + #[test] + fn decodes_and_pushes() { + let h = start(0); + wait_until("open", || h.alive.lock().unwrap().is_some()); + h.tx.send(opus_frame()).unwrap(); + wait_until("pcm pushed", || h.pushed.load(Ordering::SeqCst) > 0); + drop(h.tx); + h.join.join().unwrap(); + } + + /// A dead backend is noticed WHILE IDLE (heartbeat) and reopened without any traffic. + #[test] + fn reopens_after_idle_death() { + let h = start(0); + wait_until("first open", || h.opens.load(Ordering::SeqCst) >= 1); + wait_until("instance", || h.alive.lock().unwrap().is_some()); + h.alive + .lock() + .unwrap() + .as_ref() + .unwrap() + .store(false, Ordering::Release); // kill it + wait_until("reopen after idle death", || { + h.opens.load(Ordering::SeqCst) >= 2 + }); + drop(h.tx); + h.join.join().unwrap(); + } + + /// A death detected on push (frame flowing) also reopens, and the frame after reopen flows. + #[test] + fn reopens_after_push_death() { + let h = start(0); + wait_until("instance", || h.alive.lock().unwrap().is_some()); + h.alive + .lock() + .unwrap() + .as_ref() + .unwrap() + .store(false, Ordering::Release); + h.tx.send(opus_frame()).unwrap(); // push sees death → reopen + wait_until("reopen", || h.opens.load(Ordering::SeqCst) >= 2); + h.tx.send(opus_frame()).unwrap(); + wait_until("pcm after reopen", || h.pushed.load(Ordering::SeqCst) > 0); + drop(h.tx); + h.join.join().unwrap(); + } + + /// Instances that die immediately after opening must be retried with BACKOFF, not at + /// heartbeat rate — a flapping backend (daemon up but dropping us instantly) would + /// otherwise churn open→die→reopen every heartbeat forever. + #[test] + fn rapid_death_backs_off() { + // Every instance is dead on arrival; stability threshold high so each death counts + // as a failed open. Without the guard: ~1 reopen per heartbeat (20 ms) ≈ 25 opens in + // 500 ms. With backoff 10→20→40 (cap): ≈ 7. + let h = start_tuned(0, true, Duration::from_secs(10)); + std::thread::sleep(Duration::from_millis(500)); + let opens = h.opens.load(Ordering::SeqCst); + assert!(opens >= 2, "must keep retrying (got {opens})"); + assert!( + opens <= 15, + "must back off, not churn per heartbeat (got {opens})" + ); + drop(h.tx); + h.join.join().unwrap(); + } + + /// An uplink gap discards buffered-stale audio before the next frame plays. + #[test] + fn discards_after_gap() { + let h = start(0); + wait_until("instance", || h.alive.lock().unwrap().is_some()); + h.tx.send(opus_frame()).unwrap(); + wait_until("first push", || h.pushed.load(Ordering::SeqCst) > 0); + std::thread::sleep(Duration::from_millis(150)); // > stale_gap + h.tx.send(opus_frame()).unwrap(); + wait_until("discard on gap", || h.discards.load(Ordering::SeqCst) >= 1); + drop(h.tx); + h.join.join().unwrap(); + } +}