From 6286f91f89ab9804263285a8b5c0f4517c1c4e07 Mon Sep 17 00:00:00 2001 From: enricobuehler Date: Fri, 31 Jul 2026 21:50:29 +0200 Subject: [PATCH 1/4] feat(host/audio): the mic uplink keeps its sequence, and a backlog heals itself MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit The 0xCB datagrams always carried a seq + pts, and the ingest threw both away one line after decoding them — the pump saw an anonymous byte pile. Frames now travel as MicFrame {seq, pts_ns, opus} so the de-jitter that follows can reorder, conceal and measure. The shared queue also stops being a latency reservoir: cap 64 → 12, and a pump that wakes to a backlog deeper than 6 frames jumps to the newest 4 instead of replaying the pile — before this, a scheduling stall could park up to 1.28 s of standing mic delay that only a >600 ms silence gap ever flushed. Co-Authored-By: Claude Fable 5 --- crates/punktfunk-host/src/audio.rs | 2 +- crates/punktfunk-host/src/audio/mic_pump.rs | 127 ++++++++++++++------ crates/punktfunk-host/src/native.rs | 12 +- 3 files changed, 101 insertions(+), 40 deletions(-) diff --git a/crates/punktfunk-host/src/audio.rs b/crates/punktfunk-host/src/audio.rs index f1997202..ceef8618 100644 --- a/crates/punktfunk-host/src/audio.rs +++ b/crates/punktfunk-host/src/audio.rs @@ -153,4 +153,4 @@ mod wasapi_mic; pub(crate) mod wiring_plan; mod mic_pump; -pub use mic_pump::MicPump; +pub use mic_pump::{MicFrame, MicPump}; diff --git a/crates/punktfunk-host/src/audio/mic_pump.rs b/crates/punktfunk-host/src/audio/mic_pump.rs index e09b9e12..f65c49c0 100644 --- a/crates/punktfunk-host/src/audio/mic_pump.rs +++ b/crates/punktfunk-host/src/audio/mic_pump.rs @@ -10,10 +10,29 @@ 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; + +/// One client mic frame off the wire (`0xCB` — `punktfunk_core::quic::decode_mic_datagram`). +/// The datagram's seq + pts used to be decoded at ingest and thrown away; the pump's de-jitter +/// needs them (reorder window, loss concealment, cadence/drift), so they ride the queue with +/// the Opus payload. +pub struct MicFrame { + pub seq: u32, + pub pts_ns: u64, + pub opus: Vec, +} + +/// Bound for the shared mic frame queue (drop-newest at the producer 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). 12 × 5–20 ms frames ≈ +/// 60–240 ms of slack — enough for a scheduling hiccup, and the consumer heals the rest (see +/// [`DRAIN_ABOVE`]). The old cap of 64 could park 1.28 s of standing latency here that only a +/// >600 ms silence gap ever flushed. +const MIC_QUEUE_CAP: usize = 12; +/// Consumer self-healing: a pump that wakes to a backlog deeper than this drops down to the +/// newest [`DRAIN_KEEP`] frames before decoding — the oldest frames are already late, and +/// playing them would turn a one-off stall into permanent mic delay. +const DRAIN_ABOVE: usize = 6; +const DRAIN_KEEP: usize = 4; /// Tuning for [`MicPump`]'s open/reopen/flush behaviour — parameterized so the tests can run the /// real pump loop in milliseconds instead of seconds. @@ -62,14 +81,14 @@ const PUMP_TUNING: PumpTuning = PumpTuning { /// (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>, + 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 (tx, rx) = std::sync::mpsc::sync_channel::(MIC_QUEUE_CAP); let spawned = std::thread::Builder::new() .name("punktfunk-mic-pump".into()) .spawn(move || { @@ -90,7 +109,7 @@ impl MicPump { /// 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> { + pub fn sender(&self) -> std::sync::mpsc::SyncSender { self.tx.clone() } } @@ -99,7 +118,7 @@ impl MicPump { /// 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 { +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 { @@ -118,7 +137,7 @@ fn drain_sleep(rx: &std::sync::mpsc::Receiver>, dur: std::time::Duration /// 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) +fn pump_thread(rx: std::sync::mpsc::Receiver, opener: O, tuning: PumpTuning) where O: Fn() -> Result>, { @@ -160,34 +179,61 @@ where // Pump phase — runs until the backend dies (break) or the host shuts down (return). let mut decode_fails: u64 = 0; + let mut drain_drops: u64 = 0; let mut pcm = vec![0f32; 5760 * MIC_CHANNELS as usize]; // up to 120 ms scratch let mut last_push = Instant::now(); - loop { + let mut batch: Vec = Vec::new(); + 'pump: loop { match rx.recv_timeout(tuning.heartbeat) { - Ok(frame) => { - if frame.is_empty() { - continue; // DTX silence — the source underruns to silence on its own + Ok(first) => { + // Take everything already queued in one gulp: normally that's just `first`, + // but after a stall the backlog IS standing mic latency — heal by jumping + // to the newest few frames instead of replaying the pile. + batch.clear(); + batch.push(first); + while batch.len() <= MIC_QUEUE_CAP { + match rx.try_recv() { + Ok(f) => batch.push(f), + Err(_) => break, + } + } + if batch.len() > DRAIN_ABOVE { + let drop_n = batch.len() - DRAIN_KEEP; + drain_drops += drop_n as u64; + if drain_drops.is_power_of_two() { + tracing::debug!( + dropped = drop_n, + total = drain_drops, + "mic pump backlog — dropped oldest frames to recover latency" + ); + } + batch.drain(..drop_n); } 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; + for frame in batch.drain(..) { + if frame.opus.is_empty() { + continue; // DTX silence — the source underruns to silence on its own } - 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"); + match decoder.decode_float(&frame.opus, &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 'pump; + } + 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"); + } } } } @@ -252,7 +298,7 @@ mod pump_tests { } struct Harness { - tx: std::sync::mpsc::SyncSender>, + tx: std::sync::mpsc::SyncSender, opens: Arc, alive: Arc>>>, // latest instance's kill switch pushed: Arc, @@ -265,7 +311,7 @@ mod pump_tests { /// 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 (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)); @@ -336,6 +382,15 @@ mod pump_tests { out } + /// A wire-shaped frame: `seq` with the matching 20 ms pts, payload from [`opus_frame`]. + fn mic_frame(seq: u32) -> MicFrame { + MicFrame { + seq, + pts_ns: seq as u64 * 20_000_000, + opus: opus_frame(), + } + } + /// Eager: the backend opens (after transient failures) with NO frame ever sent. #[test] fn opens_eagerly_with_backoff() { @@ -352,7 +407,7 @@ mod pump_tests { fn decodes_and_pushes() { let h = start(0); wait_until("open", || h.alive.lock().unwrap().is_some()); - h.tx.send(opus_frame()).unwrap(); + h.tx.send(mic_frame(0)).unwrap(); wait_until("pcm pushed", || h.pushed.load(Ordering::SeqCst) > 0); drop(h.tx); h.join.join().unwrap(); @@ -388,9 +443,9 @@ mod pump_tests { .as_ref() .unwrap() .store(false, Ordering::Release); - h.tx.send(opus_frame()).unwrap(); // push sees death → reopen + h.tx.send(mic_frame(0)).unwrap(); // push sees death → reopen wait_until("reopen", || h.opens.load(Ordering::SeqCst) >= 2); - h.tx.send(opus_frame()).unwrap(); + h.tx.send(mic_frame(1)).unwrap(); wait_until("pcm after reopen", || h.pushed.load(Ordering::SeqCst) > 0); drop(h.tx); h.join.join().unwrap(); @@ -421,10 +476,10 @@ mod pump_tests { fn discards_after_gap() { let h = start(0); wait_until("instance", || h.alive.lock().unwrap().is_some()); - h.tx.send(opus_frame()).unwrap(); + h.tx.send(mic_frame(0)).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(); + h.tx.send(mic_frame(1)).unwrap(); wait_until("discard on gap", || h.discards.load(Ordering::SeqCst) >= 1); drop(h.tx); h.join.join().unwrap(); diff --git a/crates/punktfunk-host/src/native.rs b/crates/punktfunk-host/src/native.rs index 05807e7c..7dad3025 100644 --- a/crates/punktfunk-host/src/native.rs +++ b/crates/punktfunk-host/src/native.rs @@ -760,7 +760,7 @@ async fn serve_session( opts: &Punktfunk1Options, audio_cap: &AudioCapSlot, inj_tx: std::sync::mpsc::Sender, - mic_tx: std::sync::mpsc::SyncSender>, + mic_tx: std::sync::mpsc::SyncSender, host_fp: &[u8; 32], np: &NativePairing, last_pairing: &std::sync::Mutex>, @@ -1178,11 +1178,17 @@ async fn serve_session( tokio::spawn(async move { let (mut input_count, mut mic_count, mut rich_count) = (0u64, 0u64, 0u64); while let Ok(d) = input_conn.read_datagram().await { - if let Some((_seq, _pts, opus)) = punktfunk_core::quic::decode_mic_datagram(&d) { + if let Some((seq, pts, opus)) = punktfunk_core::quic::decode_mic_datagram(&d) { mic_count += 1; // Host-lifetime mic service (bounded queue): `try_send` drops the frame when the // service is full or gone, never blocking this datagram loop (security-review S6). - let _ = mic_tx.try_send(opus.to_vec()); + // seq + pts ride along — the pump's de-jitter reorders, conceals losses and + // tracks cadence with them (they used to be decoded here and thrown away). + let _ = mic_tx.try_send(crate::audio::MicFrame { + seq, + pts_ns: pts, + opus: opus.to_vec(), + }); } else if let Some(rich) = punktfunk_core::quic::RichInput::decode(&d) { rich_count += 1; if rich_tx.send(ClientInput::Rich(rich)).is_err() { From efac33fd338456f8b9bdbfcbde4e32813b81da62 Mon Sep 17 00:00:00 2001 From: enricobuehler Date: Fri, 31 Jul 2026 22:11:08 +0200 Subject: [PATCH 2/4] feat(host/audio): the virtual mic buffers for the client it has, not the worst one ever MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit The mic pump grows a real de-jitter (audio/mic_jitter.rs): a two-frame reorder window in front of the decoder, libopus concealment on sequence gaps (up to 5 frames — a lost datagram no longer drains the ring into a silence + re-prime crackle), and an adaptive target depth measured from inter-arrival jitter, clamped to 10–60 ms. Both backend rings now prime at one consumer quantum + that target: the old bursty Mac client still measures ~42 ms and lands where the fixed 48 ms prime protected it, a modern 10 ms-cadence client settles at ~25–35 ms, and a 2048-frame recorder on Linux stops buying 128 ms of latency from the 3-quanta clamp. Depth stuck above target sheds near-silent frames a few ms per 100 ms — never speech, never a hard clear. PUNKTFUNK_MIC_LEGACY_BUFFER=1 (documented) is the one-release escape hatch back to the fixed constants, and a "mic uplink health" line every 30 s (depth/target, cadence, gaps, conceals, reorders, drops, re-primes) finally says which side of the link a bad mic lives on. Co-Authored-By: Claude Fable 5 --- crates/punktfunk-host/src/audio.rs | 42 ++ crates/punktfunk-host/src/audio/linux/mod.rs | 89 +++- crates/punktfunk-host/src/audio/mic_jitter.rs | 471 ++++++++++++++++++ crates/punktfunk-host/src/audio/mic_pump.rs | 213 ++++++-- .../src/audio/windows/wasapi_mic.rs | 151 ++++-- docs-site/content/docs/configuration.md | 1 + 6 files changed, 891 insertions(+), 76 deletions(-) create mode 100644 crates/punktfunk-host/src/audio/mic_jitter.rs diff --git a/crates/punktfunk-host/src/audio.rs b/crates/punktfunk-host/src/audio.rs index ceef8618..0df8b571 100644 --- a/crates/punktfunk-host/src/audio.rs +++ b/crates/punktfunk-host/src/audio.rs @@ -115,6 +115,47 @@ pub trait VirtualMic: Send { fn channels(&self) -> u32 { CHANNELS as u32 } + + /// The adaptive de-jitter target (per-channel samples) the pump measured from uplink + /// arrival jitter (see `mic_jitter`). A backend with a jitter ring primes around this PLUS + /// one of its own consumer quanta: the ring must absorb arrival burstiness (the pump's + /// number) and pull granularity (the backend's own), and neither may buy the other's depth + /// — a 2048-frame recorder gets its one quantum, not three. Never called ⇒ the backend + /// keeps its legacy fixed constants, which is also how `PUNKTFUNK_MIC_LEGACY_BUFFER=1` + /// works (the pump simply never drives the target). Default: no ring, ignored. + fn set_target_depth(&self, _samples_per_ch: usize) {} + + /// `(buffered, prime_target)` of the backend's jitter ring in per-channel samples — read + /// by the pump's creep trim + telemetry. `None` while unknown (consumer not yet running, + /// or a backend without a ring). + fn depth(&self) -> Option<(usize, usize)> { + None + } + + /// Reset-on-read telemetry counters (see [`MicBackendStats`]). Default: all zero. + fn take_stats(&self) -> MicBackendStats { + MicBackendStats::default() + } +} + +/// Reset-on-read counters a [`VirtualMic`] backend reports into the pump's periodic mic +/// telemetry line ("mic uplink health"). +#[derive(Debug, Default, Clone, Copy)] +pub struct MicBackendStats { + /// Full-drain re-prime arms: the ring emptied and gates on silence until the target depth + /// rebuilds. One per talk spurt is normal; several per second mid-speech is the crackle. + pub reprimes: u64, + /// Per-channel samples dropped by the ring's overflow cap (drop-oldest). + pub overflow_dropped: u64, +} + +/// One-release escape hatch (docs: configuration → Audio / microphone): +/// `PUNKTFUNK_MIC_LEGACY_BUFFER=1` keeps the pre-adaptive fixed mic buffering — the pump never +/// drives the backend target (so the rings stay on their legacy constants: 48 ms prime / +/// 120 ms cap on Windows, the 3-quanta clamp on Linux) and never creep-trims depth. +pub(crate) fn mic_legacy_buffer() -> bool { + static ON: std::sync::OnceLock = std::sync::OnceLock::new(); + *ON.get_or_init(|| std::env::var_os("PUNKTFUNK_MIC_LEGACY_BUFFER").is_some_and(|v| v != "0")) } /// Open a virtual microphone with `channels` interleaved channels (1 or 2). Linux: a PipeWire @@ -152,5 +193,6 @@ mod wasapi_mic; #[path = "audio/wiring_plan.rs"] pub(crate) mod wiring_plan; +mod mic_jitter; mod mic_pump; pub use mic_pump::{MicFrame, MicPump}; diff --git a/crates/punktfunk-host/src/audio/linux/mod.rs b/crates/punktfunk-host/src/audio/linux/mod.rs index 534cc29a..b6fbd0cd 100644 --- a/crates/punktfunk-host/src/audio/linux/mod.rs +++ b/crates/punktfunk-host/src/audio/linux/mod.rs @@ -29,10 +29,10 @@ mod stream_sink; -use super::{AudioCapturer, VirtualMic, SAMPLE_RATE}; +use super::{AudioCapturer, MicBackendStats, VirtualMic, SAMPLE_RATE}; use anyhow::{anyhow, Context, Result}; use std::collections::VecDeque; -use std::sync::atomic::{AtomicBool, Ordering}; +use std::sync::atomic::{AtomicBool, AtomicU64, AtomicUsize, Ordering}; use std::sync::mpsc::{sync_channel, Receiver, RecvTimeoutError}; use std::sync::Arc; use std::thread; @@ -218,6 +218,26 @@ pub struct PwMicSource { alive: Arc, /// One-shot flush request, consumed by the process callback (clears the jitter ring). flush: Arc, + /// Ring policy/telemetry shared with the RT process callback (see [`MicRingShared`]). + ring: Arc, +} + +/// Atomics shared between [`PwMicSource`] (the pump's side) and the RT process callback: the +/// pump's adaptive de-jitter target in, ring depth/prime gauges + reset-on-read counters out. +/// All `Relaxed` — a slowly-moving target and telemetry, not synchronization. +#[derive(Default)] +struct MicRingShared { + /// Pump-set jitter target (per-channel samples). `0` = the pump never spoke (legacy mode, + /// or its first estimate hasn't landed) → the callback keeps the historical 3-quanta clamp. + target: AtomicUsize, + /// Ring depth after the last callback (per-channel samples). + depth: AtomicUsize, + /// Effective prime target of the last callback (per-channel samples). + prime: AtomicUsize, + /// Full-drain re-prime arms (see [`MicBackendStats`]). + reprimes: AtomicU64, + /// Per-channel samples dropped by the overflow cap. + overflow: AtomicU64, } impl PwMicSource { @@ -234,11 +254,13 @@ impl PwMicSource { // service started before the user session) must surface as an open ERROR — engaging the // pump's backoff — not as an instantly-dead instance the pump would churn on. let (ready_tx, ready_rx) = sync_channel::>(1); - let (alive_t, flush_t) = (alive.clone(), flush.clone()); + let ring = Arc::new(MicRingShared::default()); + let (alive_t, flush_t, ring_t) = (alive.clone(), flush.clone(), ring.clone()); thread::Builder::new() .name("punktfunk-pw-mic".into()) .spawn(move || { - if let Err(e) = mic_pw_thread(pcm_rx, quit_rx, channels, flush_t, ready_tx) { + if let Err(e) = mic_pw_thread(pcm_rx, quit_rx, channels, flush_t, ring_t, ready_tx) + { // Reaching here is always a setup/open failure (once the mainloop runs it exits // Ok) — and it was already reported to the pump via the ready handshake, which // owns the throttled operator-facing warn. Keep only a debug breadcrumb. @@ -255,6 +277,7 @@ impl PwMicSource { quit: quit_tx, alive, flush, + ring, }), Ok(Err(e)) => Err(e), Err(_) => Err(anyhow!("pipewire virtual-mic init timed out")), @@ -291,6 +314,20 @@ impl VirtualMic for PwMicSource { fn channels(&self) -> u32 { self.channels } + fn set_target_depth(&self, samples_per_ch: usize) { + self.ring.target.store(samples_per_ch, Ordering::Relaxed); + } + fn depth(&self) -> Option<(usize, usize)> { + let prime = self.ring.prime.load(Ordering::Relaxed); + // 0 = the process callback hasn't run yet (no consumer) — nothing meaningful to report. + (prime > 0).then(|| (self.ring.depth.load(Ordering::Relaxed), prime)) + } + fn take_stats(&self) -> MicBackendStats { + MicBackendStats { + reprimes: self.ring.reprimes.swap(0, Ordering::Relaxed), + overflow_dropped: self.ring.overflow.swap(0, Ordering::Relaxed), + } + } } /// Producer-side state for the virtual-mic loop: incoming decoded PCM and a small ring buffer @@ -303,6 +340,8 @@ struct MicUserData { primed: bool, /// One-shot flush request from [`PwMicSource::discard`] (stale-audio drop after a gap). flush: Arc, + /// Pump-driven ring policy + telemetry (see [`MicRingShared`]). + shared: Arc, /// When the process callback last ran — a long gap means the ring content predates the /// current consumer (the stream idles with no recorder attached) and must be dropped. last_run: Option, @@ -318,6 +357,7 @@ fn mic_pw_thread( quit_rx: pipewire::channel::Receiver, channels: u32, flush: Arc, + shared: Arc, ready: std::sync::mpsc::SyncSender>, ) -> Result<()> { use pipewire as pw; @@ -400,6 +440,7 @@ fn mic_pw_thread( channels: channels as usize, primed: false, flush, + shared, last_run: None, }; @@ -473,15 +514,31 @@ fn mic_pw_thread( ); } - // Adaptive jitter buffer. The client pushes 5 ms frames; the recorder pulls a - // whole *quantum* (often 20–43 ms) from an independent clock. A drain of one - // quantum must not outrun what's buffered, or every call underruns to silence - // (the original ~58% gaps). So prime to ~3 quanta before producing, hold there, - // and re-prime only after a genuine full drain (the client went quiet). The ring - // is capped at a few quanta so latency stays bounded. - let target = (3 * want).clamp(720 * ud.channels, 9600 * ud.channels); + // Jitter buffer. The client pushes frames on its own clock; the recorder + // pulls a whole *quantum* (often 20–43 ms) from an independent one. A drain + // of one quantum must not outrun what's buffered, or every call underruns + // to silence (the original ~58% gaps). Prime target = one quantum (the pull + // granularity) + the pump's measured-jitter target (arrival burstiness — + // see `mic_jitter`), re-priming only after a genuine full drain (the client + // went quiet). Until the pump's first estimate lands — and forever under + // PUNKTFUNK_MIC_LEGACY_BUFFER — the historical 3-quanta clamp applies; that + // formula scaled with the RECORDING APP's quantum, so a 2048-frame recorder + // bought 128+ ms of standing mic latency for jitter it never had. + let pump_target = ud.shared.target.load(Ordering::Relaxed) * ud.channels; + let target = if pump_target == 0 { + (3 * want).clamp(720 * ud.channels, 9600 * ud.channels) + } else { + want + pump_target + }; + let mut dropped = 0usize; while ud.ring.len() > target.max(want) + want { ud.ring.pop_front(); // bound latency: drop the oldest beyond ~1 quantum slack + dropped += 1; + } + if dropped > 0 { + ud.shared + .overflow + .fetch_add((dropped / ud.channels) as u64, Ordering::Relaxed); } if !ud.primed && ud.ring.len() >= target { ud.primed = true; @@ -501,9 +558,17 @@ fn mic_pw_thread( } else { 0 }; - if ud.ring.is_empty() { + if ud.primed && ud.ring.is_empty() { ud.primed = false; // fully drained — re-prime before producing again + ud.shared.reprimes.fetch_add(1, Ordering::Relaxed); } + // Publish depth/prime for the pump's creep trim + telemetry. + ud.shared + .depth + .store(ud.ring.len() / ud.channels, Ordering::Relaxed); + ud.shared + .prime + .store(target / ud.channels, Ordering::Relaxed); let chunk = data.chunk_mut(); *chunk.offset_mut() = 0; *chunk.stride_mut() = stride as _; diff --git a/crates/punktfunk-host/src/audio/mic_jitter.rs b/crates/punktfunk-host/src/audio/mic_jitter.rs new file mode 100644 index 00000000..1ae08d4e --- /dev/null +++ b/crates/punktfunk-host/src/audio/mic_jitter.rs @@ -0,0 +1,471 @@ +//! Backend-agnostic de-jitter for the client-mic uplink. The pump feeds every received `0xCB` +//! frame (with its datagram sequence — see [`MicFrame`]) through [`MicDejitter`], which puts a +//! two-frame reorder window in front of the Opus decoder, asks for libopus concealment on true +//! sequence gaps (so a lost datagram doesn't drain the backend ring into a silence + re-prime +//! cycle), and measures inter-arrival jitter into the adaptive target depth the backend rings +//! prime against. Mirrors the client downlink's `AudioGapTracker` (punktfunk-core `audio.rs`): +//! the same wrapping-sequence gap rules, grown the reorder hold and the jitter estimator the +//! uplink needs — the downlink's jitter rings own their depth; the uplink's live in the +//! backends and take their target from here. + +use super::MicFrame; +use std::time::{Duration, Instant}; + +/// What the pump should do next, in playback order. `Frame` is a real frame to decode; +/// `Conceal` is one frame of libopus packet-loss concealment (`decode_float(&[], …)`, sized to +/// the last real frame) standing in for a lost one. +pub(super) enum Deliver { + Frame(MicFrame), + Conceal, +} + +/// Most concealment frames one gap may synthesize (~100 ms at the common 20 ms frames) — the +/// client downlink's `AudioGapTracker` cap, scaled to the uplink's bigger frames. libopus PLC +/// fades to silence after a few frames anyway; past the cap the ring's underrun/re-prime path +/// takes over, as before. +const MAX_CONCEAL_FRAMES: u32 = 5; + +/// How long one out-of-order frame may wait for its missing predecessor before the gap is +/// concealed and the held frame plays: ~one 20 ms frame interval plus scheduling slack. With +/// the incoming frame that is a two-frame window — anything needing more is treated as loss, +/// exactly like the client downlink (which holds nothing at all). +const HOLD_MAX: Duration = Duration::from_millis(30); + +/// Arrival gaps above this are talk-spurt pauses (DTX, mute), not network jitter — folding +/// them into the estimate would balloon the target after every pause. +const PAUSE_GAP: Duration = Duration::from_millis(250); + +/// The sliding jitter window: [`BUCKETS`] × [`BUCKET_LEN`] of per-bucket max inter-arrival gap. +/// Long enough that the old bursty Mac cadence (~two 20 ms frames every ~42 ms — the 2026-07-03 +/// "crackling mic", see `wasapi_mic.rs`) is always represented; short enough that a one-off +/// spike ages out in a few seconds. +const BUCKETS: usize = 4; +const BUCKET_LEN: Duration = Duration::from_millis(1000); + +/// Target before the estimator has a full [`WARMUP`] of evidence: roughly the old fixed 48 ms +/// prime minus the one consumer quantum the backends add — the crackle-safe assumption that +/// the client might be an old bursty one. Modern 10 ms-cadence clients drop to ~15–25 ms once +/// measured (after their first talk pause re-primes the ring at the lower target). +const DEFAULT_TARGET_MS: u32 = 40; +/// Headroom over the measured worst burst gap. +const TARGET_PAD_MS: u32 = 5; +/// How long the estimator distrusts its own (young) window and keeps the default as a floor. +const WARMUP: Duration = Duration::from_secs(1); + +/// Reset-on-read snapshot for the pump's telemetry line. Counters cover the window since the +/// last read; `cadence_ms`/`frame_ms` are gauges (EWMAs). +#[derive(Debug, Default, Clone, Copy)] +pub(super) struct DejitterStats { + pub seq_gaps: u64, + pub concealed: u64, + pub reorders: u64, + pub late_drops: u64, + /// EWMA inter-arrival gap — the uplink's burst cadence as seen by the pump. + pub cadence_ms: f32, + /// Nominal frame duration from consecutive frames' pts deltas (what the client encodes). + pub frame_ms: f32, +} + +pub(super) struct MicDejitter { + /// Sequence the uplink owes next; `None` before the first frame / after + /// [`reset_stream`](Self::reset_stream). + expected: Option, + /// The reorder window: at most ONE newer frame parked ≤ [`HOLD_MAX`] for its predecessor. + held: Option<(Instant, MicFrame)>, + // --- adaptive target (inter-arrival jitter) --- + first_arrival: Option, + last_arrival: Option, + bucket_start: Option, + bucket_idx: usize, + gap_max: [f32; BUCKETS], + ewma_gap_ms: f32, + // --- pts bookkeeping --- + /// `(seq, pts_ns)` of the newest ingested frame — a pts delta over exactly one seq step is + /// the client's true frame duration. + last_pts: Option<(u32, u64)>, + frame_ms: f32, + // --- counters (reset on read) --- + seq_gaps: u64, + concealed: u64, + reorders: u64, + late_drops: u64, +} + +impl MicDejitter { + pub(super) fn new() -> Self { + MicDejitter { + expected: None, + held: None, + first_arrival: None, + last_arrival: None, + bucket_start: None, + bucket_idx: 0, + gap_max: [0.0; BUCKETS], + ewma_gap_ms: 0.0, + last_pts: None, + frame_ms: 0.0, + seq_gaps: 0, + concealed: 0, + reorders: 0, + late_drops: 0, + } + } + + /// Feed one received frame; playback-ordered work is appended to `out`. + pub(super) fn ingest(&mut self, now: Instant, frame: MicFrame, out: &mut Vec) { + self.record_arrival(now, frame.seq, frame.pts_ns); + self.accept(now, frame, out); + } + + fn accept(&mut self, now: Instant, frame: MicFrame, out: &mut Vec) { + let Some(expected) = self.expected else { + // First frame (or first after a reset): the chain starts here. + self.expected = Some(frame.seq.wrapping_add(1)); + out.push(Deliver::Frame(frame)); + return; + }; + let delta = frame.seq.wrapping_sub(expected); + if delta == 0 { + self.expected = Some(expected.wrapping_add(1)); + out.push(Deliver::Frame(frame)); + self.release_held_in_order(out); + } else if delta > u32::MAX / 2 { + // Duplicate of, or older than, something already played — the window moved on. + self.late_drops += 1; + } else if let Some(held_seq) = self.held.as_ref().map(|(_, h)| h.seq) { + if held_seq == frame.seq { + self.late_drops += 1; // duplicate of the held frame + return; + } + // Two newer-than-expected frames in hand — waiting longer can't fill the gap + // within the window. Play them in order, concealing what's genuinely missing. + let (_, held) = self.held.take().expect("checked above"); + let (first, second) = if frame.seq.wrapping_sub(held.seq) > u32::MAX / 2 { + (frame, held) + } else { + (held, frame) + }; + self.give_up_and_play(first, out); + self.accept(now, second, out); // depth ≤ 1: `held` is None here + } else { + self.held = Some((now, frame)); // wait ≤ HOLD_MAX for the predecessor + } + } + + /// If the held frame's predecessor just played, the hold is over — the reorder repaired. + fn release_held_in_order(&mut self, out: &mut Vec) { + if let Some((_, h)) = &self.held { + if Some(h.seq) == self.expected { + let (_, h) = self.held.take().expect("checked above"); + self.expected = Some(h.seq.wrapping_add(1)); + self.reorders += 1; + out.push(Deliver::Frame(h)); + } + } + } + + /// Stop waiting for `frame`'s predecessors: conceal the gap (capped) and play it. + fn give_up_and_play(&mut self, frame: MicFrame, out: &mut Vec) { + let gap = frame + .seq + .wrapping_sub(self.expected.expect("callers checked")); + if gap > 0 { + self.seq_gaps += 1; + let conceal = gap.min(MAX_CONCEAL_FRAMES); + self.concealed += u64::from(conceal); + for _ in 0..conceal { + out.push(Deliver::Conceal); + } + } + self.expected = Some(frame.seq.wrapping_add(1)); + out.push(Deliver::Frame(frame)); + } + + /// When the current hold (if any) must be given up — the pump's next wake deadline. + pub(super) fn hold_deadline(&self) -> Option { + self.held.as_ref().map(|(t, _)| *t + HOLD_MAX) + } + + /// Give up an expired hold (the predecessor never came): conceal + play. Called on the + /// pump's timeout wakes. + pub(super) fn flush_expired_hold(&mut self, now: Instant, out: &mut Vec) { + if self.hold_deadline().is_some_and(|d| now >= d) { + let (_, h) = self.held.take().expect("deadline implies held"); + self.give_up_and_play(h, out); + } + } + + /// Uplink pause / backend reopen: whatever is parked predates the gap, and the next frame + /// restarts the chain (a pause is not loss — it must not conceal or count a gap). + pub(super) fn reset_stream(&mut self) { + self.expected = None; + self.held = None; + self.last_pts = None; + } + + // ---- adaptive target ------------------------------------------------------------------- + + fn record_arrival(&mut self, now: Instant, seq: u32, pts_ns: u64) { + if let Some(last) = self.last_arrival { + let gap = now.duration_since(last); + if gap <= PAUSE_GAP { + let ms = gap.as_secs_f32() * 1000.0; + self.ewma_gap_ms = if self.ewma_gap_ms == 0.0 { + ms + } else { + self.ewma_gap_ms * 0.9 + ms * 0.1 + }; + self.rotate_buckets(now); + self.gap_max[self.bucket_idx] = self.gap_max[self.bucket_idx].max(ms); + } + } + self.first_arrival.get_or_insert(now); + self.last_arrival = Some(now); + if let Some((last_seq, last_pts)) = self.last_pts { + if seq == last_seq.wrapping_add(1) && pts_ns > last_pts { + let d = (pts_ns - last_pts) as f32 / 1_000_000.0; + if (1.0..=120.0).contains(&d) { + self.frame_ms = if self.frame_ms == 0.0 { + d + } else { + self.frame_ms * 0.9 + d * 0.1 + }; + } + } + } + self.last_pts = Some((seq, pts_ns)); + } + + fn rotate_buckets(&mut self, now: Instant) { + let Some(mut start) = self.bucket_start else { + self.bucket_start = Some(now); + return; + }; + let mut advanced = 0; + while now.duration_since(start) >= BUCKET_LEN { + self.bucket_idx = (self.bucket_idx + 1) % BUCKETS; + self.gap_max[self.bucket_idx] = 0.0; + start += BUCKET_LEN; + advanced += 1; + if advanced >= BUCKETS { + // Everything aged out (long pause) — snap the window to now. + self.gap_max = [0.0; BUCKETS]; + start = now; + break; + } + } + self.bucket_start = Some(start); + } + + /// The jitter component of the backend ring's prime depth, in ms — the worst burst gap in + /// the window plus a pad, clamped to [10, 60]. Backends add ONE consumer quantum on top + /// (see [`VirtualMic::set_target_depth`](super::VirtualMic::set_target_depth)). Old bursty + /// clients measure ~42 ms and land near the historical 48 ms prime; modern 10 ms-cadence + /// clients settle at ~15–25 ms. + pub(super) fn target_ms(&self, now: Instant) -> u32 { + let measured = + self.gap_max.iter().fold(0f32, |a, &b| a.max(b)).ceil() as u32 + TARGET_PAD_MS; + let warm = self + .first_arrival + .is_some_and(|t| now.duration_since(t) >= WARMUP); + let t = if warm { + measured + } else { + measured.max(DEFAULT_TARGET_MS) + }; + t.clamp(10, 60) + } + + /// Reset-on-read counters + gauges for the pump's telemetry line. + pub(super) fn take_stats(&mut self) -> DejitterStats { + let s = DejitterStats { + seq_gaps: self.seq_gaps, + concealed: self.concealed, + reorders: self.reorders, + late_drops: self.late_drops, + cadence_ms: self.ewma_gap_ms, + frame_ms: self.frame_ms, + }; + self.seq_gaps = 0; + self.concealed = 0; + self.reorders = 0; + self.late_drops = 0; + s + } +} + +#[cfg(test)] +mod tests { + use super::*; + + fn f(seq: u32) -> MicFrame { + MicFrame { + seq, + pts_ns: u64::from(seq) * 20_000_000, + opus: vec![1], + } + } + + /// Compact delivery shape: frame seqs as-is, concealment as -1. + fn seqs(out: &[Deliver]) -> Vec { + out.iter() + .map(|d| match d { + Deliver::Frame(fr) => i64::from(fr.seq), + Deliver::Conceal => -1, + }) + .collect() + } + + #[test] + fn in_order_passthrough() { + let mut dj = MicDejitter::new(); + let t = Instant::now(); + let mut out = Vec::new(); + for s in 5..8 { + dj.ingest(t, f(s), &mut out); + } + assert_eq!(seqs(&out), [5, 6, 7]); + let st = dj.take_stats(); + assert_eq!( + (st.seq_gaps, st.concealed, st.reorders, st.late_drops), + (0, 0, 0, 0) + ); + } + + #[test] + fn swapped_pair_is_repaired_not_concealed() { + let mut dj = MicDejitter::new(); + let t = Instant::now(); + let mut out = Vec::new(); + dj.ingest(t, f(5), &mut out); + dj.ingest(t, f(7), &mut out); // held, waiting for 6 + assert_eq!(seqs(&out), [5]); + dj.ingest(t, f(6), &mut out); // the missing one arrives late → both play, in order + assert_eq!(seqs(&out), [5, 6, 7]); + let st = dj.take_stats(); + assert_eq!(st.reorders, 1); + assert_eq!(st.concealed, 0); + } + + #[test] + fn loss_is_concealed_when_the_window_moves_on() { + let mut dj = MicDejitter::new(); + let t = Instant::now(); + let mut out = Vec::new(); + dj.ingest(t, f(5), &mut out); + dj.ingest(t, f(7), &mut out); // held, waiting for 6 + dj.ingest(t, f(8), &mut out); // 6 is lost: one PLC frame, then 7 and 8 + assert_eq!(seqs(&out), [5, -1, 7, 8]); + let st = dj.take_stats(); + assert_eq!(st.seq_gaps, 1); + assert_eq!(st.concealed, 1); + } + + #[test] + fn expired_hold_flushes_with_concealment() { + let mut dj = MicDejitter::new(); + let t = Instant::now(); + let mut out = Vec::new(); + dj.ingest(t, f(5), &mut out); + dj.ingest(t, f(7), &mut out); + assert!(dj.hold_deadline().is_some()); + dj.flush_expired_hold(t + HOLD_MAX, &mut out); + assert_eq!(seqs(&out), [5, -1, 7]); + assert!(dj.hold_deadline().is_none()); + } + + #[test] + fn big_gap_conceal_is_capped() { + let mut dj = MicDejitter::new(); + let t = Instant::now(); + let mut out = Vec::new(); + dj.ingest(t, f(5), &mut out); + dj.ingest(t, f(100), &mut out); // held + dj.flush_expired_hold(t + HOLD_MAX, &mut out); + let conceals = out.iter().filter(|d| matches!(d, Deliver::Conceal)).count(); + assert_eq!(conceals, MAX_CONCEAL_FRAMES as usize); + assert_eq!(dj.take_stats().seq_gaps, 1); + } + + #[test] + fn duplicates_and_stale_late_frames_drop() { + let mut dj = MicDejitter::new(); + let t = Instant::now(); + let mut out = Vec::new(); + dj.ingest(t, f(5), &mut out); + dj.ingest(t, f(6), &mut out); + dj.ingest(t, f(6), &mut out); // duplicate + dj.ingest(t, f(3), &mut out); // ancient + dj.ingest(t, f(8), &mut out); // held + dj.ingest(t, f(8), &mut out); // duplicate of the held frame + assert_eq!(seqs(&out), [5, 6]); + assert_eq!(dj.take_stats().late_drops, 3); + } + + #[test] + fn wraparound_is_a_reorder_not_an_eternity() { + let mut dj = MicDejitter::new(); + let t = Instant::now(); + let mut out = Vec::new(); + dj.ingest(t, f(u32::MAX), &mut out); + dj.ingest(t, f(0), &mut out); // in order across the wrap + assert_eq!(seqs(&out), [i64::from(u32::MAX), 0]); + dj.ingest(t, f(u32::MAX), &mut out); // pre-wrap replay: late, not a 2^31 gap + assert_eq!(dj.take_stats().late_drops, 1); + } + + #[test] + fn reset_restarts_the_chain_without_concealing() { + let mut dj = MicDejitter::new(); + let t = Instant::now(); + let mut out = Vec::new(); + dj.ingest(t, f(5), &mut out); + dj.reset_stream(); + dj.ingest(t, f(500), &mut out); // a pause is not loss + assert_eq!(seqs(&out), [5, 500]); + assert_eq!(dj.take_stats().concealed, 0); + } + + #[test] + fn bursty_uplink_grows_the_target_modern_uplink_shrinks_it() { + // Old Mac cadence: two frames back-to-back every ~42 ms. + let mut dj = MicDejitter::new(); + let t0 = Instant::now(); + let mut out = Vec::new(); + let mut now = t0; + let mut seq = 0u32; + for burst in 0u64..50 { + now = t0 + Duration::from_millis(burst * 42); + for _ in 0..2 { + dj.ingest(now, f(seq), &mut out); + seq += 1; + } + } + let target = dj.target_ms(now); + assert!((42..=60).contains(&target), "bursty target {target}"); + + // Modern 10 ms cadence: past warmup the target follows the small gaps down. + let mut dj = MicDejitter::new(); + let mut now = t0; + for i in 0u64..200 { + now = t0 + Duration::from_millis(i * 10); + dj.ingest(now, f(i as u32), &mut out); + } + let target = dj.target_ms(now); + assert!((10..=20).contains(&target), "steady target {target}"); + } + + #[test] + fn warmup_holds_the_crackle_safe_default() { + let mut dj = MicDejitter::new(); + let t0 = Instant::now(); + let mut out = Vec::new(); + dj.ingest(t0, f(0), &mut out); + dj.ingest(t0 + Duration::from_millis(10), f(1), &mut out); + // 10 ms of evidence must not shrink the target yet — the client might be bursty. + assert_eq!( + dj.target_ms(t0 + Duration::from_millis(20)), + DEFAULT_TARGET_MS + ); + } +} diff --git a/crates/punktfunk-host/src/audio/mic_pump.rs b/crates/punktfunk-host/src/audio/mic_pump.rs index f65c49c0..912ddf45 100644 --- a/crates/punktfunk-host/src/audio/mic_pump.rs +++ b/crates/punktfunk-host/src/audio/mic_pump.rs @@ -5,6 +5,7 @@ //! trait, its factory ([`open_virtual_mic`](super::open_virtual_mic)) and the audio-plane sample //! rate stay in `super`. +use super::mic_jitter::{Deliver, MicDejitter}; use super::{VirtualMic, SAMPLE_RATE}; use anyhow::Result; @@ -34,6 +35,20 @@ const MIC_QUEUE_CAP: usize = 12; const DRAIN_ABOVE: usize = 6; const DRAIN_KEEP: usize = 4; +/// Cadence of the "mic uplink health" telemetry line (emitted only while frames flow). +const TELEMETRY_EVERY: std::time::Duration = std::time::Duration::from_secs(30); + +/// Creep trim: the backend ring must sit more than this over its prime target, for +/// [`TRIM_AFTER`], before the pump starts shedding — and then only near-silent frames, at most +/// one per [`TRIM_SPACING`] (a 20 ms frame per 300 ms ≈ 7 ms per 100 ms). Depth built by a +/// burst otherwise only ever comes back down at a full drain (the device consumes exactly what +/// it plays). +const TRIM_MARGIN_MS: usize = 15; +const TRIM_AFTER: std::time::Duration = std::time::Duration::from_secs(2); +const TRIM_SPACING: std::time::Duration = std::time::Duration::from_millis(300); +/// Peak below which a decoded frame counts as a silence stretch (≈ −48 dBFS). +const TRIM_SILENCE_PEAK: f32 = 0.004; + /// 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)] @@ -75,6 +90,10 @@ const PUMP_TUNING: PumpTuning = PumpTuning { /// 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`]). +/// - **De-jittered**: frames pass a sequence-aware reorder window + loss concealment, and an +/// adaptive target-depth estimator drives the backend rings' priming — see +/// [`MicDejitter`](super::mic_jitter) (`PUNKTFUNK_MIC_LEGACY_BUFFER=1` restores the fixed +/// pre-adaptive buffering). /// /// 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 @@ -178,13 +197,33 @@ where let opened_at = Instant::now(); // Pump phase — runs until the backend dies (break) or the host shuts down (return). + let legacy = super::mic_legacy_buffer(); let mut decode_fails: u64 = 0; let mut drain_drops: u64 = 0; + let mut trimmed: u64 = 0; + let mut frames_seen: u64 = 0; let mut pcm = vec![0f32; 5760 * MIC_CHANNELS as usize]; // up to 120 ms scratch let mut last_push = Instant::now(); let mut batch: Vec = Vec::new(); + let mut deliveries: Vec = Vec::new(); + let mut jitter = MicDejitter::new(); + // Concealment must match the last real frame's duration — libopus sizes PLC from the + // output slice. One 20 ms frame until something decodes. + let mut plc_samples: usize = 960; + let mut applied_target_ms: u32 = 0; + let mut over_since: Option = None; + let mut last_trim = Instant::now(); + let mut last_log = Instant::now(); 'pump: loop { - match rx.recv_timeout(tuning.heartbeat) { + // Wake for the next frame, the heartbeat, or a parked reorder hold aging out — + // whichever is soonest. + let timeout = jitter + .hold_deadline() + .map(|d| d.saturating_duration_since(Instant::now())) + .unwrap_or(tuning.heartbeat) + .min(tuning.heartbeat); + deliveries.clear(); + match rx.recv_timeout(timeout) { Ok(first) => { // Take everything already queued in one gulp: normally that's just `first`, // but after a stall the backlog IS standing mic latency — heal by jumping @@ -200,45 +239,23 @@ where if batch.len() > DRAIN_ABOVE { let drop_n = batch.len() - DRAIN_KEEP; drain_drops += drop_n as u64; - if drain_drops.is_power_of_two() { - tracing::debug!( - dropped = drop_n, - total = drain_drops, - "mic pump backlog — dropped oldest frames to recover latency" - ); - } batch.drain(..drop_n); } + frames_seen += batch.len() as u64; if last_push.elapsed() > tuning.stale_gap { mic.discard(); + jitter.reset_stream(); } + let now = Instant::now(); for frame in batch.drain(..) { - if frame.opus.is_empty() { - continue; // DTX silence — the source underruns to silence on its own - } - match decoder.decode_float(&frame.opus, &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 'pump; - } - 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"); - } - } - } + jitter.ingest(now, frame, &mut deliveries); } + // A hold whose window expired while traffic kept flowing (only late + // duplicates arriving) must still flush — the timeout arm never fires then. + jitter.flush_expired_hold(now, &mut deliveries); } Err(RecvTimeoutError::Timeout) => { + jitter.flush_expired_hold(Instant::now(), &mut deliveries); if !mic.alive() { tracing::warn!("virtual mic backend died while idle — reopening"); break; @@ -249,6 +266,123 @@ where return; } } + + for d in deliveries.drain(..) { + let samples_per_ch = match d { + Deliver::Frame(frame) => { + if frame.opus.is_empty() { + continue; // DTX silence — the source underruns to silence on its own + } + match decoder.decode_float(&frame.opus, &mut pcm, false) { + Ok(n) => { + plc_samples = n.max(120); // ≥ 2.5 ms keeps PLC well-formed + decode_fails = 0; + n + } + Err(e) => { + // Malformed/garbage frame: drop it, keep the shared mic + + // decoder (see the struct docs). Throttled log (1, 2, 4, …). + decode_fails += 1; + if decode_fails.is_power_of_two() { + tracing::warn!(error = %e, fails = decode_fails, + "mic opus decode failed — dropping frame"); + } + continue; + } + } + } + Deliver::Conceal => { + // A lost datagram: an empty-input decode synthesizes libopus PLC for + // exactly the slice's duration, so the backend ring doesn't starve + // into a silence + re-prime cycle over one missing frame. + let want = (plc_samples * MIC_CHANNELS as usize).min(pcm.len()); + match decoder.decode_float(&[], &mut pcm[..want], false) { + Ok(n) => n, + Err(_) => continue, // nothing decoded yet — nothing to extend + } + } + }; + let total = (samples_per_ch * MIC_CHANNELS as usize).min(pcm.len()); + // Creep trim: ring depth persistently over target sheds a near-silent frame at + // a time (a few ms per 100 ms) — never speech, never a hard clear. Without it, + // depth built by a burst only ever comes back down at a full drain. + let mut shed = false; + if !legacy { + match mic.depth() { + Some((buffered, target)) + if buffered > target + TRIM_MARGIN_MS * SAMPLE_RATE as usize / 1000 => + { + let since = *over_since.get_or_insert_with(Instant::now); + if since.elapsed() >= TRIM_AFTER + && last_trim.elapsed() >= TRIM_SPACING + && pcm[..total].iter().all(|s| s.abs() < TRIM_SILENCE_PEAK) + { + shed = true; + last_trim = Instant::now(); + trimmed += 1; + } + } + _ => over_since = None, + } + } + if shed { + last_push = Instant::now(); // the uplink is live — a trim is not a stale gap + continue; + } + if !mic.push(&pcm[..total]) { + tracing::warn!("virtual mic backend died — reopening"); + break 'pump; + } + last_push = Instant::now(); + } + + // Drive the backend ring's prime target from the measured jitter. Legacy mode + // never calls this, which keeps the backend on its fixed constants (see + // `VirtualMic::set_target_depth`). + if !legacy { + let t = jitter.target_ms(Instant::now()); + if t != applied_target_ms { + applied_target_ms = t; + mic.set_target_depth(t as usize * SAMPLE_RATE as usize / 1000); + } + } + + // One structured health line per interval while the mic is live (reset-on-read). + if last_log.elapsed() >= TELEMETRY_EVERY { + if frames_seen > 0 { + let js = jitter.take_stats(); + let bs = mic.take_stats(); + let (depth_ms, target_ms) = mic + .depth() + .map(|(d, t)| { + ( + d * 1000 / SAMPLE_RATE as usize, + t * 1000 / SAMPLE_RATE as usize, + ) + }) + .unwrap_or((0, 0)); + tracing::info!( + depth_ms, + target_ms, + cadence_ms = (js.cadence_ms * 10.0).round() / 10.0, + frame_ms = (js.frame_ms * 10.0).round() / 10.0, + frames = frames_seen, + gaps = js.seq_gaps, + concealed = js.concealed, + reorders = js.reorders, + late = js.late_drops, + drained = drain_drops, + trimmed, + reprimes = bs.reprimes, + overflow_ms = bs.overflow_dropped * 1000 / SAMPLE_RATE as u64, + "mic uplink health" + ); + } + last_log = Instant::now(); + frames_seen = 0; + drain_drops = 0; + trimmed = 0; + } } // Death triage: an instance that lived is a one-off (PipeWire/audio-engine restart) — @@ -471,6 +605,23 @@ mod pump_tests { h.join.join().unwrap(); } + /// A lost datagram is concealed: seq 0 then 2 (1 missing) must still push ~3 frames of PCM + /// (decode, PLC, decode) once the reorder window expires — the backend ring never starves + /// over a single missing frame. + #[test] + fn seq_gap_is_concealed() { + let h = start(0); + wait_until("instance", || h.alive.lock().unwrap().is_some()); + h.tx.send(mic_frame(0)).unwrap(); + h.tx.send(mic_frame(2)).unwrap(); + // 0 plays immediately; 2 is held ≤ 30 ms for 1, then the gap is concealed and 2 plays. + wait_until("conceal + late frame pushed", || { + h.pushed.load(Ordering::SeqCst) >= 3 * 960 * 2 + }); + drop(h.tx); + h.join.join().unwrap(); + } + /// An uplink gap discards buffered-stale audio before the next frame plays. #[test] fn discards_after_gap() { diff --git a/crates/punktfunk-host/src/audio/windows/wasapi_mic.rs b/crates/punktfunk-host/src/audio/windows/wasapi_mic.rs index cac735a8..dbc15aa0 100644 --- a/crates/punktfunk-host/src/audio/windows/wasapi_mic.rs +++ b/crates/punktfunk-host/src/audio/windows/wasapi_mic.rs @@ -19,19 +19,21 @@ //! returns `false` and the pump reopens (re-planning, so endpoint churn re-resolves). Before this //! existed, the first device change silently killed mic passthrough for the rest of the host's life. //! -//! `push` enqueues decoded interleaved-f32 PCM into a bounded ring (drop-oldest beyond ~120 ms so -//! mic latency stays bounded); a dedicated COM-apartment thread renders it event-driven through an -//! adaptive jitter buffer (prime → hold → re-prime, see the render loop — clients arrive in bursts, -//! the device pulls per-period), filling silence when the client isn't talking. WASAPI objects are -//! `!Send`, so they live entirely on that thread (mirrors `WasapiLoopbackCapturer`). +//! `push` enqueues decoded interleaved-f32 PCM into a bounded ring (drop-oldest so mic latency +//! stays bounded — the bound follows the adaptive prime threshold, or the legacy ~120 ms until +//! the pump drives it); a dedicated COM-apartment thread renders it event-driven through a +//! jitter buffer (prime → hold → re-prime, see the render loop — clients arrive in bursts, the +//! device pulls per-period) whose prime depth the mic pump sets from measured uplink jitter +//! ([`VirtualMic::set_target_depth`]), filling silence when the client isn't talking. WASAPI +//! objects are `!Send`, so they live entirely on that thread (mirrors `WasapiLoopbackCapturer`). // Every `unsafe` block in this file carries a `// SAFETY:` proof; enforce it. #![deny(clippy::undocumented_unsafe_blocks)] -use super::{audio_control, VirtualMic, SAMPLE_RATE}; +use super::{audio_control, MicBackendStats, VirtualMic, SAMPLE_RATE}; use anyhow::{anyhow, Context, Result}; use std::collections::VecDeque; -use std::sync::atomic::{AtomicBool, Ordering}; +use std::sync::atomic::{AtomicBool, AtomicU64, AtomicUsize, Ordering}; use std::sync::mpsc::{sync_channel, SyncSender}; use std::sync::{Arc, Mutex}; use std::thread::{self, JoinHandle}; @@ -41,26 +43,53 @@ use wasapi::{Direction, SampleType, StreamMode, WaveFormat}; const CHANNELS: u32 = 2; /// 48 kHz stereo f32: 2 channels * 4 bytes. const BLOCK_ALIGN: usize = 2 * 4; -/// Jitter-buffer priming depth (~48 ms): the render loop emits pure silence until this much PCM -/// is queued, then plays from the cushion. Clients deliver mic audio in BURSTS (the Mac client's -/// input tap yields ~two 20 ms Opus packets every ~42 ms) while WASAPI pulls a small block every -/// device period (~10 ms) — with no cushion the queue sits near-empty and most periods insert -/// mid-stream silence: the "crackling mic" (heard live, Mac → Windows host 2026-07-03; the Linux -/// backend's process callback primes the same way and the identical stream was clean there). The -/// depth must cover the worst inter-burst gap (~42 ms), so ~48 ms with re-prime on a full drain. +/// LEGACY jitter-buffer priming depth (~48 ms): the render loop emits pure silence until this +/// much PCM is queued, then plays from the cushion. Old clients deliver mic audio in BURSTS +/// (the Mac client's input tap yields ~two 20 ms Opus packets every ~42 ms) while WASAPI pulls +/// a small block every device period (~10 ms) — with no cushion the queue sits near-empty and +/// most periods insert mid-stream silence: the "crackling mic" (heard live, Mac → Windows host +/// 2026-07-03; the Linux backend's process callback primes the same way and the identical +/// stream was clean there). The depth had to cover the worst inter-burst gap (~42 ms), so +/// ~48 ms with re-prime on a full drain. Today the pump MEASURES that gap and drives the prime +/// threshold per client ([`VirtualMic::set_target_depth`] — bursty clients still get ~48 ms, +/// modern 10 ms-cadence ones ~25–35 ms); this constant remains the fallback until the pump's +/// first estimate, and forever under `PUNKTFUNK_MIC_LEGACY_BUFFER=1`. const PRIME_BYTES: usize = (SAMPLE_RATE as usize * 48 / 1000) * BLOCK_ALIGN; -/// Bound the inject queue at ~120 ms so the passed-through mic stays low-latency (drop oldest -/// beyond): the priming cushion (~48 ms) plus arrival-burst headroom. +/// LEGACY bound for the inject queue at ~120 ms (drop oldest beyond): the fixed priming +/// cushion plus arrival-burst headroom. Applies only while the pump isn't driving the target; +/// adaptive mode bounds the queue at prime + [`CAP_HEADROOM_BYTES`] instead (≈ 50–105 ms). const MAX_QUEUE_BYTES: usize = (SAMPLE_RATE as usize * 120 / 1000) * BLOCK_ALIGN; +/// Producer-side overflow headroom (~32 ms) over the render loop's prime threshold when the +/// adaptive target drives the ring. +const CAP_HEADROOM_BYTES: usize = (SAMPLE_RATE as usize * 32 / 1000) * BLOCK_ALIGN; pub struct WasapiVirtualMic { queue: Arc>>, stop: Arc, /// False once the render thread has exited (device error or stop) — the pump's reopen signal. alive: Arc, + /// Ring policy/telemetry shared with the render thread (see [`RingShared`]). + ring: Arc, join: Option>, } +/// Atomics shared between the pump-facing handle and the render thread: the pump's adaptive +/// de-jitter target in, the effective prime threshold + reset-on-read counters out. All +/// `Relaxed` — a slowly-moving target and telemetry, not synchronization. +#[derive(Default)] +struct RingShared { + /// Pump-set jitter target in bytes. `0` = the pump never spoke (legacy mode, or its first + /// estimate hasn't landed) → the render loop keeps the fixed [`PRIME_BYTES`] and `push` + /// keeps the fixed [`MAX_QUEUE_BYTES`]. + target_bytes: AtomicUsize, + /// Effective prime threshold (bytes) of the last render iteration. + prime_bytes: AtomicUsize, + /// Full-drain re-prime arms (see [`MicBackendStats`]). + reprimes: AtomicU64, + /// Per-channel samples dropped by the overflow cap. + overflow: AtomicU64, +} + impl WasapiVirtualMic { pub fn open(channels: u32) -> Result { anyhow::ensure!( @@ -70,14 +99,15 @@ impl WasapiVirtualMic { let queue = Arc::new(Mutex::new(VecDeque::::new())); let stop = Arc::new(AtomicBool::new(false)); let alive = Arc::new(AtomicBool::new(true)); + let ring = Arc::new(RingShared::default()); // Bring-up handshake: report the resolved device (or the error) before returning, so a missing // virtual-mic device surfaces as Err (the caller retries with backoff) not a silent dead thread. let (ready_tx, ready_rx) = sync_channel::>(1); - let (q, st, al) = (queue.clone(), stop.clone(), alive.clone()); + let (q, st, rg, al) = (queue.clone(), stop.clone(), ring.clone(), alive.clone()); let join = thread::Builder::new() .name("punktfunk-wasapi-mic".into()) .spawn(move || { - if let Err(e) = render_thread(q, st, ready_tx) { + if let Err(e) = render_thread(q, st, rg, ready_tx) { tracing::error!(error = %format!("{e:#}"), "wasapi virtual-mic thread failed"); } // Normal stop or device error alike: this instance is done — the pump reopens. @@ -92,6 +122,7 @@ impl WasapiVirtualMic { queue, stop, alive, + ring, join: Some(join), }) } @@ -122,10 +153,25 @@ impl VirtualMic for WasapiVirtualMic { for &s in pcm { q.extend(s.to_le_bytes()); } - // Drop-oldest to keep latency bounded (mic is real-time; stale audio is worse than dropped). - if q.len() > MAX_QUEUE_BYTES { - let excess = q.len() - MAX_QUEUE_BYTES; + // Drop-oldest to keep latency bounded (mic is real-time; stale audio is worse than + // dropped). With the pump driving the target, the bound follows the render loop's prime + // threshold + headroom; otherwise (legacy / no estimate yet) the fixed 120 ms applies. + let cap = if self.ring.target_bytes.load(Ordering::Relaxed) == 0 { + MAX_QUEUE_BYTES + } else { + // `max(PRIME_BYTES)` covers the one render period before the loop first publishes. + self.ring + .prime_bytes + .load(Ordering::Relaxed) + .max(PRIME_BYTES) + + CAP_HEADROOM_BYTES + }; + if q.len() > cap { + let excess = q.len() - cap; q.drain(..excess); + self.ring + .overflow + .fetch_add((excess / BLOCK_ALIGN) as u64, Ordering::Relaxed); } true } @@ -143,6 +189,28 @@ impl VirtualMic for WasapiVirtualMic { fn channels(&self) -> u32 { CHANNELS } + + fn set_target_depth(&self, samples_per_ch: usize) { + self.ring + .target_bytes + .store(samples_per_ch * BLOCK_ALIGN, Ordering::Relaxed); + } + + fn depth(&self) -> Option<(usize, usize)> { + let prime = self.ring.prime_bytes.load(Ordering::Relaxed); + if prime == 0 { + return None; // render loop hasn't run yet + } + let q = self.queue.lock().ok()?; + Some((q.len() / BLOCK_ALIGN, prime / BLOCK_ALIGN)) + } + + fn take_stats(&self) -> MicBackendStats { + MicBackendStats { + reprimes: self.ring.reprimes.swap(0, Ordering::Relaxed), + overflow_dropped: self.ring.overflow.swap(0, Ordering::Relaxed), + } + } } /// Resolve the mic inject target from the wiring plan, auto-installing the Steam Streaming pair @@ -273,6 +341,7 @@ fn try_install_steam_audio(inf_name: &str) -> bool { fn render_thread( queue: Arc>>, stop: Arc, + shared: Arc, ready: SyncSender>, ) -> Result<()> { if let Err(e) = wasapi::initialize_mta() @@ -284,7 +353,7 @@ fn render_thread( } // Open + start the render stream. The WASAPI objects must outlive the loop, so build them here and // keep them (a closure that *returned* them would drop them); on any failure report Err and exit. - let setup = (|| -> Result<(wasapi::AudioClient, wasapi::AudioRenderClient, wasapi::Handle, String)> { + let setup = (|| -> Result<(wasapi::AudioClient, wasapi::AudioRenderClient, wasapi::Handle, i64, String)> { let (device, name) = resolve_target()?; let mut audio_client = device.get_iaudioclient().context("IAudioClient")?; // 48 kHz stereo f32; autoconvert lets WASAPI shared-mode SRC match the device mix format. @@ -312,9 +381,9 @@ fn render_thread( let buf_frames = audio_client.get_buffer_size().context("buffer size")? as usize; let _ = render_client.write_to_device(buf_frames, &vec![0u8; buf_frames * BLOCK_ALIGN], None); audio_client.start_stream().context("start render stream")?; - Ok((audio_client, render_client, h_event, name)) + Ok((audio_client, render_client, h_event, default_period, name)) })(); - let (audio_client, render_client, h_event, name) = match setup { + let (audio_client, render_client, h_event, default_period, name) = match setup { Ok(t) => t, Err(e) => { let _ = ready.send(Err(anyhow!("{e:#}"))); @@ -322,18 +391,26 @@ fn render_thread( } }; let _ = ready.send(Ok(name)); + // One device period in bytes (period is in 100 ns units; floor 10 ms if it reads absurd) — + // the device-side pull granularity the adaptive prime threshold builds on. + let period_bytes = ((default_period.max(0) as usize * SAMPLE_RATE as usize / 10_000_000) + .max(SAMPLE_RATE as usize / 100)) + * BLOCK_ALIGN; // Any error below (endpoint invalidated/removed, engine restart) propagates out of the loop, // ending the thread — the `alive` flag flips in the spawn wrapper and the pump reopens. // - // Adaptive jitter buffer (mirrors the Linux backend's process callback): clients push mic - // audio in bursts on their own clock while the device pulls a block every period from an - // independent clock, so a greedy per-period drain leaves the queue near-empty and pads most - // periods with mid-stream silence — audible as constant crackling. Instead: emit silence - // until [`PRIME_BYTES`] is buffered, then play from the cushion (zero-filling only a - // momentary shortfall), and re-prime only after a genuine FULL drain (the client went quiet — - // between talk spurts the cushion rebuilds, and [`VirtualMic::discard`] resets it across - // session gaps). + // Jitter buffer (mirrors the Linux backend's process callback): clients push mic audio in + // bursts on their own clock while the device pulls a block every period from an independent + // clock, so a greedy per-period drain leaves the queue near-empty and pads most periods + // with mid-stream silence — audible as constant crackling. Instead: emit silence until the + // prime threshold is buffered, then play from the cushion (zero-filling only a momentary + // shortfall), and re-prime only after a genuine FULL drain (the client went quiet — between + // talk spurts the cushion rebuilds, and [`VirtualMic::discard`] resets it across session + // gaps). The threshold = one device period + the pump's measured-jitter target + // ([`VirtualMic::set_target_depth`]); the fixed [`PRIME_BYTES`] until the pump's first + // estimate, and forever under `PUNKTFUNK_MIC_LEGACY_BUFFER=1` (the pump then never drives + // the target). let mut buf: Vec = Vec::new(); let mut primed = false; while !stop.load(Ordering::Relaxed) { @@ -351,11 +428,18 @@ fn render_thread( if buf.len() < need { buf.resize(need, 0); } + let target = shared.target_bytes.load(Ordering::Relaxed); + let prime = if target == 0 { + PRIME_BYTES + } else { + period_bytes + target + }; + shared.prime_bytes.store(prime, Ordering::Relaxed); // Silence base; overwrite with queued mic PCM once the cushion is primed. buf[..need].fill(0); { let mut q = queue.lock().unwrap(); - if !primed && q.len() >= PRIME_BYTES { + if !primed && q.len() >= prime { primed = true; } if primed { @@ -365,6 +449,7 @@ fn render_thread( } if q.is_empty() { primed = false; // fully drained — re-prime before producing again + shared.reprimes.fetch_add(1, Ordering::Relaxed); } } } diff --git a/docs-site/content/docs/configuration.md b/docs-site/content/docs/configuration.md index b45ef24e..db20455f 100644 --- a/docs-site/content/docs/configuration.md +++ b/docs-site/content/docs/configuration.md @@ -150,6 +150,7 @@ See your desktop page ([KDE](/docs/kde), [GNOME](/docs/gnome)) for when to set t |---|---|---| | `PUNKTFUNK_AUDIO_GAIN` | float (default `1.0`) | **(Moonlight/GameStream sessions only)** Linear gain applied to captured desktop audio — bump it for a quiet source. The native `punktfunk/1` path ignores it; adjust the source's own volume there instead. | | `PUNKTFUNK_MIC_DEVICE` | name substring | **(Windows)** Target mic-uplink device by friendly-name substring (first match wins). | +| `PUNKTFUNK_MIC_LEGACY_BUFFER` | `1` | Restore the fixed pre-adaptive mic buffering (a ~48 ms prime and ~120 ms cap on Windows; a buffer scaled to the recording app's audio quantum on Linux) instead of the adaptive per-client jitter target. One-release escape hatch: if the microphone coming out of the host only sounds right *with* this set, that's a bug — please report it. | | `PUNKTFUNK_NO_MIC_INSTALL` | set | **(Windows)** Skip installing the virtual-mic driver (e.g. when the host runs as SYSTEM). | | `PUNKTFUNK_HOST_AUDIO` | set | **(Windows)** Also play the stream's audio on the host's own speakers. While a session is capturing desktop audio the host parks the default playback device on a silent sink, so sound comes out of the *client* only — that's why the PC goes quiet when a stream starts. Set this to prefer a real output device instead (audible on both ends). The default is put back when the capture closes. | | `PUNKTFUNK_KEEP_DEFAULT` | set | **(Windows)** Never touch the default playback/recording devices at all — the host leaves whatever you chose in Sound settings in place. The mic uplink still picks a target device; you may then have to select it yourself. | From 79856d2c5024eef55114e8607ae6163624513cb7 Mon Sep 17 00:00:00 2001 From: enricobuehler Date: Fri, 31 Jul 2026 22:11:17 +0200 Subject: [PATCH 3/4] fix(host/audio): a VoiceMeeter box can no longer loop the mic into the stream MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit wiring_plan could hand the mic Voicemeeter Input and the loopback Voicemeeter Aux Input — two strips of the same internal mixer, i.e. a digital feedback loop with no acoustic path to break it, because leftover() never asked virtualish() and the exclusion list only knew "cable" and the Steam Speakers. Now every VoiceMeeter or generically "virtual" render is excluded from loopback, and the last-resort tier refuses anything virtual: a box with only mixer endpoints gets loopback=None, honest like the cable-only case. Co-Authored-By: Claude Fable 5 --- .../punktfunk-host/src/audio/wiring_plan.rs | 80 +++++++++++++++++-- 1 file changed, 73 insertions(+), 7 deletions(-) diff --git a/crates/punktfunk-host/src/audio/wiring_plan.rs b/crates/punktfunk-host/src/audio/wiring_plan.rs index 172a4915..10314419 100644 --- a/crates/punktfunk-host/src/audio/wiring_plan.rs +++ b/crates/punktfunk-host/src/audio/wiring_plan.rs @@ -69,11 +69,18 @@ fn capture_for(mic_render_lname: &str) -> &'static [&'static str] { } /// A render endpoint no loopback should capture: the VB-CABLE (reserved for the mic even when it -/// isn't the chosen target — capturing a cable someone else feeds echoes too) and the Steam -/// Streaming Speakers, whose loopback is silent (validated live). Also the capture-side -/// watchdog's test for "the operator's new default can never work — snap back to the plan". +/// isn't the chosen target — capturing a cable someone else feeds echoes too), the Steam +/// Streaming Speakers, whose loopback is silent (validated live), and any VoiceMeeter or +/// generically-"virtual" endpoint. VoiceMeeter's strips share one internal mixer: capturing ANY +/// of its render endpoints re-captures what the mic wrote into another one — a digital feedback +/// loop with no acoustic path to break it (mic=`Voicemeeter Input`, loopback=`Voicemeeter Aux +/// Input` used to pass the old name checks). Also the capture-side watchdog's test for "the +/// operator's new default can never work — snap back to the plan". pub(crate) fn excluded_from_loopback(lname: &str) -> bool { - lname.contains("cable") || lname.contains("steam streaming speakers") + lname.contains("cable") + || lname.contains("steam streaming speakers") + || lname.contains("voicemeeter") + || lname.contains("virtual") } /// A render endpoint that is SILENT on the host but loopback-capturable — the client-only audio @@ -140,10 +147,14 @@ pub(crate) fn plan( .iter() .find(|(n, id)| not_mic(id) && silent_sink(&n.to_lowercase())) }; + // `virtualish` here too: a virtual endpoint that slipped past `excluded_from_loopback`'s + // name list (a future cable/mixer sibling) is an internal-feedback loop waiting to happen — + // no loopback is the honest answer, exactly like the cable-only case. let leftover = || { - renders - .iter() - .find(|(n, id)| not_mic(id) && !excluded_from_loopback(&n.to_lowercase())) + renders.iter().find(|(n, id)| { + let ln = n.to_lowercase(); + not_mic(id) && !excluded_from_loopback(&ln) && !virtualish(&ln) + }) }; let loopback_render = if host_audio { real_hw().or_else(silent).or_else(leftover) @@ -359,4 +370,59 @@ mod tests { assert!(w.mic_render.is_none()); assert_eq!(w.loopback_render.unwrap().0, "Speakers (Realtek HD Audio)"); } + + /// VoiceMeeter box with real hardware: no cable, so the mic takes `Voicemeeter Input` — and + /// the loopback must NEVER be `Voicemeeter Aux Input` (same internal mixer: capturing any + /// VoiceMeeter render re-captures what the mic wrote — a digital feedback loop). Real + /// hardware wins the loopback in both modes. + #[test] + fn voicemeeter_aux_never_pairs_with_voicemeeter_mic() { + let renders = [ + ep("Voicemeeter Input (VB-Audio Voicemeeter VAIO)"), + ep("Voicemeeter Aux Input (VB-Audio Voicemeeter AUX VAIO)"), + ep("Speakers (Realtek HD Audio)"), + ]; + let captures = [ep("Voicemeeter Out B1 (VB-Audio Voicemeeter VAIO)")]; + for host_audio in [false, true] { + let w = plan(&renders, &captures, None, host_audio); + assert_eq!( + w.mic_render.as_ref().unwrap().0, + "Voicemeeter Input (VB-Audio Voicemeeter VAIO)", + "host_audio={host_audio}" + ); + assert_eq!( + w.loopback_render.as_ref().unwrap().0, + "Speakers (Realtek HD Audio)", + "host_audio={host_audio}" + ); + } + } + + /// Only VoiceMeeter endpoints (headless mixer box): mic wins one, the loopback is honestly + /// absent — like the cable-only case, never another strip of the same mixer. + #[test] + fn voicemeeter_only_no_loopback() { + let renders = [ + ep("Voicemeeter Input (VB-Audio Voicemeeter VAIO)"), + ep("Voicemeeter Aux Input (VB-Audio Voicemeeter AUX VAIO)"), + ]; + for host_audio in [false, true] { + let w = plan(&renders, &[], None, host_audio); + assert!(w.mic_render.is_some(), "host_audio={host_audio}"); + assert!(w.loopback_render.is_none(), "host_audio={host_audio}"); + } + } + + /// A generically-"virtual" leftover (unknown vendor cable) is refused too: `leftover()` + /// applies `virtualish`, so a virtual endpoint that slips past the name list can't become + /// the loopback. + #[test] + fn unknown_virtual_never_loopback() { + let renders = [ + ep("CABLE Input (VB-Audio Virtual Cable)"), + ep("Speakers (Some Virtual Audio Device)"), + ]; + let w = plan(&renders, &[], None, false); + assert!(w.loopback_render.is_none()); + } } From e861565e27e0f3dc99da72a494a798f3518baf01 Mon Sep 17 00:00:00 2001 From: enricobuehler Date: Fri, 31 Jul 2026 22:11:19 +0200 Subject: [PATCH 4/4] =?UTF-8?q?docs:=20"Why=20do=20I=20hear=20myself"=20?= =?UTF-8?q?=E2=80=94=20the=20four=20echo=20loops,=20and=20which=20knob=20s?= =?UTF-8?q?tops=20each?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Co-Authored-By: Claude Fable 5 --- docs-site/content/docs/echo.md | 58 ++++++++++++++++++++++++++++++++ docs-site/content/docs/meta.json | 1 + 2 files changed, 59 insertions(+) create mode 100644 docs-site/content/docs/echo.md diff --git a/docs-site/content/docs/echo.md b/docs-site/content/docs/echo.md new file mode 100644 index 00000000..14102c58 --- /dev/null +++ b/docs-site/content/docs/echo.md @@ -0,0 +1,58 @@ +--- +title: Why do I hear myself +description: Echo while streaming — the four places it can come from (client speakers, Windows monitoring loops, host speakers, virtual mixers) and how to stop each one. +--- + +You talk into your device's microphone and hear your own voice come back a beat later. The echo +is almost never "in the stream" itself — it's a loop in one of four well-known places. Work +through them in order; the first two cover nearly every report. + +## Your device's speakers + +If the stream's audio plays out of the **speakers of the device you're streaming on** — a +phone, tablet or laptop without headphones — its microphone picks that sound back up and sends +it to the host along with your voice. Everyone in your voice chat hears the game twice, and you +hear yourself whenever anything routes the mic back. + +**Fix: use headphones on the device you're streaming on.** Newer clients cancel this +automatically (acoustic echo cancellation is rolling out per platform); headphones are the +reliable fix everywhere today. + +## "Listen to this device" and app monitoring (Windows hosts) + +Windows can play a microphone straight out of the speakers. If **Listen to this device** is +ticked for the Punktfunk mic (usually *CABLE Output*), your voice plays on the host's output — +which the stream then captures and sends right back to you. + +Open **Sound settings → More sound settings → Recording**, double-click *CABLE Output*, and on +the **Listen** tab untick *Listen to this device*. + +The same loop hides in apps: **Discord's** *Mic Test* / input monitoring, **OBS's** *Monitor +audio* on a mic source, and similar monitoring features in other tools all play your mic into +the host's output. Turn the monitoring off rather than the mic. + +## The host's own speakers + +If you set `PUNKTFUNK_HOST_AUDIO` (Windows) so the stream's sound also plays in the room, and +you're streaming **from that same room**, your device's mic hears the host's speakers. Remove +the setting while you stream from nearby, or turn the host's volume down. + +## Virtual mixers (VoiceMeeter and friends) + +VoiceMeeter's virtual devices all share one internal mixer. Older Punktfunk hosts could pick one +VoiceMeeter strip as the microphone target and *another* as the audio capture — a feedback loop +with no acoustics involved at all. Current hosts refuse to capture VoiceMeeter or other virtual +endpoints for desktop audio, so this fixes itself with an update. If you route audio through +VoiceMeeter on purpose, make sure no strip that hears the Punktfunk mic feeds the output being +streamed. + +## What the host log can tell you + +While your microphone is in use, the host writes a **`mic uplink health`** line to its log every +30 seconds (web console → **Logs**). It shows how much of your voice is buffered on the host +(`depth_ms` vs `target_ms`), how much the network lost (`gaps`, `concealed`), and how steadily +your client is delivering audio (`cadence_ms`). It won't point at an echo loop directly — echo +is a routing problem, not a network one — but if your voice also sounds choppy or delayed, +include that line in a bug report. The startup log also names exactly which devices the host +picked for the microphone and for audio capture, which is the quickest way to spot a monitoring +loop like the ones above. diff --git a/docs-site/content/docs/meta.json b/docs-site/content/docs/meta.json index c24a52d3..548edd9a 100644 --- a/docs-site/content/docs/meta.json +++ b/docs-site/content/docs/meta.json @@ -47,6 +47,7 @@ "wake-on-lan", "---Troubleshooting---", "troubleshooting", + "echo", "stats", "forgot-password", "---Project---",