diff --git a/crates/punktfunk-host/src/audio.rs b/crates/punktfunk-host/src/audio.rs index f1997202..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::MicPump; +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 e09b9e12..912ddf45 100644 --- a/crates/punktfunk-host/src/audio/mic_pump.rs +++ b/crates/punktfunk-host/src/audio/mic_pump.rs @@ -5,15 +5,49 @@ //! 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; /// 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; + +/// 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. @@ -56,20 +90,24 @@ 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 /// (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 +128,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 +137,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 +156,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>, { @@ -159,40 +197,65 @@ 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(); - loop { - match rx.recv_timeout(tuning.heartbeat) { - Ok(frame) => { - if frame.is_empty() { - continue; // DTX silence — the source underruns to silence on its own + 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 { + // 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 + // 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; + batch.drain(..drop_n); + } + frames_seen += batch.len() as u64; if last_push.elapsed() > tuning.stale_gap { mic.discard(); + jitter.reset_stream(); } - 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"); - } - } + let now = Instant::now(); + for frame in batch.drain(..) { + 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; @@ -203,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) — @@ -252,7 +432,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 +445,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 +516,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 +541,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 +577,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(); @@ -416,15 +605,32 @@ 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() { 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/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/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()); + } } 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() { 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. | 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---",