From 15392bd7071db81cba7fe0a8adca93442eb4bb92 Mon Sep 17 00:00:00 2001 From: enricobuehler Date: Fri, 31 Jul 2026 21:46:02 +0200 Subject: [PATCH 1/3] fix(core/client): the mic queue stops hoarding a second of your voice MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit MIC_QUEUE claimed 64 was ~320 ms of 5 ms frames; the frames were 20 ms, so it really allowed 1.28 s — and since a full tokio mpsc can only refuse the FRESH frame, one worker stall turned the whole backlog into permanent standing mic latency. The queue shrinks to 12 and the pump's mic task now sheds oldest-first past a ~60 ms backlog, so a stall costs a short dropout and heals itself. New per-stage counters (sent / dropped-full / dropped-stale) surface through NativeClient::mic_stats for the stats HUDs. Co-Authored-By: Claude Fable 5 --- crates/punktfunk-core/src/client/mod.rs | 73 +++++++++++++++++++--- crates/punktfunk-core/src/client/pump.rs | 10 +++ crates/punktfunk-core/src/client/worker.rs | 3 + 3 files changed, 77 insertions(+), 9 deletions(-) diff --git a/crates/punktfunk-core/src/client/mod.rs b/crates/punktfunk-core/src/client/mod.rs index 59472a0e..eda3b9e5 100644 --- a/crates/punktfunk-core/src/client/mod.rs +++ b/crates/punktfunk-core/src/client/mod.rs @@ -63,11 +63,45 @@ fn join_host_port(host: &str, port: u16) -> String { } } -/// Outbound mic uplink queue depth: 5 ms Opus frames, so 64 is ~320 ms of audio — far beyond -/// any worker stall a live mic session survives anyway. On overflow the FRESH frame is dropped -/// (a tokio mpsc can't shed from the head; by the time 320 ms are queued the stream is broken -/// either way, and the bound is about memory, not audio quality) and logged at debug. -const MIC_QUEUE: usize = 64; +/// Outbound mic uplink queue depth: desktop clients send 10 ms Opus frames (mobile still 20 ms), +/// so 12 is ~120–240 ms of audio — the hard memory cap, not the working depth. (The old 64 was +/// mislabeled "~320 ms of 5 ms frames"; the frames were 20 ms, so it really allowed 1.28 s, and +/// since a filled tokio mpsc can only drop the FRESH frame, every queued frame became permanent +/// standing latency once a stall filled it.) The pump's mic task now sheds the OLDEST frames +/// whenever more than [`MIC_BACKLOG_MAX`] are still waiting, so a stall costs a short dropout +/// and heals the moment the worker catches up. The producer stays non-blocking: on overflow it +/// still drops the fresh frame (logged at debug) — the shed loop keeps that a rare event. +const MIC_QUEUE: usize = 12; + +/// The mic backlog the pump tolerates before shedding oldest-first: ~60 ms of 10 ms frames — +/// enough slack to ride out an encode/send hiccup, small enough that voice stays conversational +/// and never accrues a session-long standing delay. +pub(crate) const MIC_BACKLOG_MAX: usize = 6; + +/// Mic uplink counters, shared between the producer side ([`NativeClient::send_mic`]) and the +/// pump's mic task. All monotonic for the session; a stats HUD windows them by diffing +/// successive [`NativeClient::mic_stats`] snapshots (the `frames_dropped` pattern). +#[derive(Debug, Default)] +pub(crate) struct MicUplinkCounters { + /// Frames handed to the QUIC datagram send (past every client-side queue). + pub(crate) sent: AtomicU64, + /// Frames shed at enqueue — the worker queue was full ([`MIC_QUEUE`]). + pub(crate) dropped_full: AtomicU64, + /// Frames shed by the pump's backlog governor (stale-oldest past [`MIC_BACKLOG_MAX`]). + pub(crate) dropped_stale: AtomicU64, +} + +/// A [`NativeClient::mic_stats`] snapshot: cumulative mic uplink frame counts per stage. +#[derive(Clone, Copy, Debug, Default)] +pub struct MicUplinkStats { + /// Frames handed to the QUIC datagram send. + pub sent: u64, + /// Frames shed at enqueue (worker queue full). + pub dropped_full: u64, + /// Frames shed by the pump's backlog governor (stale-oldest — see [`MIC_QUEUE`]'s + /// self-healing note). + pub dropped_stale: u64, +} /// Outbound control-request queue depth. The requests are sparse (mode switches, keyframe /// requests, ~1.3 loss reports/s, clock re-syncs) — 32 is hours of headroom; a full queue means @@ -101,9 +135,12 @@ pub struct NativeClient { cursor_state: Mutex>, input_tx: tokio::sync::mpsc::UnboundedSender, /// Outbound mic frames `(seq, pts_ns, opus)` → encoded as 0xCB datagrams by the worker. - /// Bounded ([`MIC_QUEUE`]): a wedged worker drops fresh frames (logged) instead of queueing - /// audio-latency (and memory) without limit — mic is best-effort end to end. + /// Bounded ([`MIC_QUEUE`]): the pump sheds stale frames oldest-first and a full queue drops + /// the fresh one (logged) instead of queueing audio-latency (and memory) without limit — + /// mic is best-effort end to end, and a standing backlog is worse than a dropout. mic_tx: tokio::sync::mpsc::Sender<(u32, u64, Vec)>, + /// Mic uplink counters (sent / dropped per stage) — see [`NativeClient::mic_stats`]. + mic_stats: Arc, /// Outbound 0xCC rich-input plane, PRE-ENCODED datagrams: [`RichInput`] touchpad/motion /// (encoded in [`NativeClient::send_rich_input`]) and stylus [`crate::quic::PenBatch`]es /// (encoded in [`NativeClient::send_pen`]) share the channel — the worker's task just @@ -396,6 +433,7 @@ impl NativeClient { let probe = Arc::new(Mutex::new(ProbeState::default())); let frames_dropped = Arc::new(AtomicU64::new(0)); let fec_recovered = Arc::new(AtomicU64::new(0)); + let mic_stats = Arc::new(MicUplinkCounters::default()); let hot_tids = Arc::new(Mutex::new(Vec::new())); let clock_offset = Arc::new(AtomicI64::new(0)); let decode_lat = Arc::new(Mutex::new(DecodeLatAcc::default())); @@ -408,6 +446,7 @@ impl NativeClient { let probe_w = probe.clone(); let frames_dropped_w = frames_dropped.clone(); let fec_recovered_w = fec_recovered.clone(); + let mic_stats_w = mic_stats.clone(); let hot_tids_w = hot_tids.clone(); let clock_offset_w = clock_offset.clone(); let decode_lat_w = decode_lat.clone(); @@ -472,6 +511,7 @@ impl NativeClient { probe: probe_w, frames_dropped: frames_dropped_w, fec_recovered: fec_recovered_w, + mic_stats: mic_stats_w, hot_tids: hot_tids_w, clock_offset: clock_offset_w, decode_lat: decode_lat_w, @@ -506,6 +546,7 @@ impl NativeClient { cursor_state: Mutex::new(cursor_state_rx), input_tx, mic_tx, + mic_stats, rich_input_tx, ctrl_tx, clip: Mutex::new(clip_event_rx), @@ -709,6 +750,18 @@ impl NativeClient { self.fec_recovered.load(Ordering::Relaxed) } + /// Cumulative mic uplink frame counts per stage: handed to the QUIC datagram send, shed at + /// enqueue (worker queue full), and shed by the pump's backlog governor (stale-oldest — see + /// [`MIC_QUEUE`]'s self-healing note). All monotonic for the session; a stats HUD windows + /// them by diffing successive reads, like [`frames_dropped`](Self::frames_dropped). + pub fn mic_stats(&self) -> MicUplinkStats { + MicUplinkStats { + sent: self.mic_stats.sent.load(Ordering::Relaxed), + dropped_full: self.mic_stats.dropped_full.load(Ordering::Relaxed), + dropped_stale: self.mic_stats.dropped_stale.load(Ordering::Relaxed), + } + } + /// Whether the underlying QUIC session has ended — the worker's connection-close watcher set the /// shutdown flag (`conn.closed()` fired: a host suspend / crash / network drop idle-timed the /// connection out, or the host closed it), or a deliberate [`disconnect_quit`](Self::disconnect_quit) @@ -1124,8 +1177,10 @@ impl NativeClient { match self.mic_tx.try_send((seq, pts_ns, opus)) { Ok(()) => Ok(()), Err(TrySendError::Full(_)) => { - // Bounded queue full = the worker stalled for ~MIC_QUEUE x 5 ms. Shed this - // frame (mic is best-effort end to end) instead of queueing latency/memory. + // Bounded queue full = the worker stalled long enough to outrun even the + // pump's oldest-first shed. Drop this frame (mic is best-effort end to end) + // instead of queueing latency/memory; the counter keeps the loss visible. + self.mic_stats.dropped_full.fetch_add(1, Ordering::Relaxed); tracing::debug!("mic uplink queue full — dropping frame"); Ok(()) } diff --git a/crates/punktfunk-core/src/client/pump.rs b/crates/punktfunk-core/src/client/pump.rs index f3774164..73b922a2 100644 --- a/crates/punktfunk-core/src/client/pump.rs +++ b/crates/punktfunk-core/src/client/pump.rs @@ -68,6 +68,7 @@ pub(super) async fn run_pump(args: WorkerArgs) { probe, frames_dropped, fec_recovered, + mic_stats, hot_tids, clock_offset, decode_lat, @@ -92,11 +93,20 @@ pub(super) async fn run_pump(args: WorkerArgs) { tokio::spawn(input_task::run(conn.clone(), input_rx, gamepad_snapshots)); // Mic task: embedder Opus mic frames → 0xCB uplink datagrams (best-effort, dropped on loss). + // Self-healing latency bound: every frame still queued once this task catches up is standing + // mic delay from then on, so a frame with more than [`MIC_BACKLOG_MAX`] successors already + // waiting is shed as stale instead of sent — a stall costs a short dropout, not a + // session-long lag (see [`MIC_QUEUE`]). The counters feed [`NativeClient::mic_stats`]. let mic_conn = conn.clone(); tokio::spawn(async move { while let Some((seq, pts_ns, opus)) = mic_rx.recv().await { + if mic_rx.len() > MIC_BACKLOG_MAX { + mic_stats.dropped_stale.fetch_add(1, Ordering::Relaxed); + continue; + } let d = crate::quic::encode_mic_datagram(seq, pts_ns, &opus); let _ = mic_conn.send_datagram(d.into()); + mic_stats.sent.fetch_add(1, Ordering::Relaxed); } }); diff --git a/crates/punktfunk-core/src/client/worker.rs b/crates/punktfunk-core/src/client/worker.rs index 07b1f961..fd02c44b 100644 --- a/crates/punktfunk-core/src/client/worker.rs +++ b/crates/punktfunk-core/src/client/worker.rs @@ -66,6 +66,9 @@ pub(crate) struct WorkerArgs { pub(crate) probe: Arc>, pub(crate) frames_dropped: Arc, pub(crate) fec_recovered: Arc, + /// Mic uplink counters (see [`NativeClient::mic_stats`]): the pump's mic task counts wire + /// sends and its own stale-shed drops here; the producer counts queue-full drops. + pub(crate) mic_stats: Arc, pub(crate) hot_tids: Arc>>, /// The live clock offset (see [`NativeClient::clock_offset`]): the worker seeds it with the /// connect-time estimate; the control task's mid-stream re-syncs update it. From d17e942db0b1699cd8ec351cc9fad41f746335bc Mon Sep 17 00:00:00 2001 From: enricobuehler Date: Fri, 31 Jul 2026 21:49:05 +0200 Subject: [PATCH 2/3] feat(client/audio): the mic speaks 10 ms mono and asks the OS to cancel the echo MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Uplink format on both desktops: 10 ms mono frames (was 20 ms stereo), Opus Voip at 48 kbps with in-band FEC against 10 % assumed loss — half the frame-fill latency, half the samples, and a lost datagram's audio now rides in its successor. One datagram per frame, unchanged wire. Linux: the capture stream finally asks for its own quantum (NODE_LATENCY 480/48000) instead of inheriting the graph's 1024-2048 sample bursts, and when the user picked no mic it prefers an existing echo-cancel source over the default (PUNKTFUNK_NO_AEC=1 opts out; loading module-echo-cancel ourselves needs a load_module the pipewire crate doesn't expose yet). Windows: the capture client declares AudioCategory_Communications before Initialize so an endpoint's communications APO (the system AEC) can engage; capture stays stereo via autoconvert — the proven path — and downmixes to mono in code. Co-Authored-By: Claude Fable 5 --- crates/pf-client-core/src/audio.rs | 100 +++++++++++++++++----- crates/pf-client-core/src/audio_wasapi.rs | 71 +++++++++++---- 2 files changed, 134 insertions(+), 37 deletions(-) diff --git a/crates/pf-client-core/src/audio.rs b/crates/pf-client-core/src/audio.rs index 6e62912b..d40a185d 100644 --- a/crates/pf-client-core/src/audio.rs +++ b/crates/pf-client-core/src/audio.rs @@ -14,9 +14,12 @@ use std::sync::mpsc::{Receiver, SyncSender, TrySendError}; use std::sync::Arc; const SAMPLE_RATE: u32 = 48_000; -const CHANNELS: usize = 2; -/// Mic frames are 20 ms (960 samples/channel) — any size ≤ 120 ms is fine host-side. -const MIC_FRAME: usize = 960; +/// Mic capture is MONO: voice is mono at the source, the host accepts any Opus channel +/// layout (its stereo decoder upmixes), and half the samples halve the encode + wire cost. +const MIC_CHANNELS: usize = 1; +/// Mic frames are 10 ms (480 mono samples) — any size ≤ 120 ms is fine host-side; 10 ms +/// halves the frame-fill share of mouth-to-ear latency vs the old 20 ms. +const MIC_FRAME: usize = 480; struct Terminate; @@ -327,8 +330,9 @@ fn pw_thread( Ok(()) } -/// The microphone uplink: capture the default input device, Opus-encode 20 ms chunks, -/// ship them as 0xCB datagrams into the host's virtual PipeWire source. +/// The microphone uplink: capture the default input device (or the picked / echo-cancelled +/// source), Opus-encode 10 ms mono chunks, ship them as 0xCB datagrams into the host's +/// virtual PipeWire source. pub struct MicStreamer { quit_tx: pipewire::channel::Sender, thread: Option>, @@ -361,8 +365,8 @@ impl Drop for MicStreamer { } } -/// Capture-side state: accumulated PCM and the Opus encoder (encoding a 20 ms frame is -/// ~100 µs — fine inside the process callback). +/// Capture-side state: accumulated PCM and the Opus encoder (encoding a 10 ms frame is +/// well under 100 µs — fine inside the process callback). struct MicData { connector: Arc, ring: VecDeque, @@ -371,6 +375,55 @@ struct MicData { out: Vec, } +/// `PUNKTFUNK_NO_AEC=1` — the opt-out for every mic echo-cancellation hook (here: the +/// echo-cancelled-source preference; the WASAPI twin gates its Communications category on it). +fn aec_disabled() -> bool { + std::env::var("PUNKTFUNK_NO_AEC").is_ok_and(|v| !v.is_empty() && v != "0") +} + +/// The capture stream's `target.object`, in preference order: the Settings microphone pick +/// (`Settings::mic_device` via session main's `PUNKTFUNK_AUDIO_SOURCE`) verbatim, else — so a +/// desktop that already runs `module-echo-cancel` stops feeding its own downlink audio back +/// into the host's virtual mic — the first echo-cancelled source in the graph. `None` = the +/// user picked nothing and no such source exists: PipeWire's default routing, as before. +/// +/// Preference-only by design: loading `libpipewire-module-echo-cancel` ourselves needs +/// `pw_context_load_module`, which the pipewire crate (0.9) doesn't expose safely — until it +/// does, we only ever target processing the user (or their session) already set up. +fn mic_capture_target() -> Option { + if let Ok(target) = std::env::var("PUNKTFUNK_AUDIO_SOURCE") { + if !target.is_empty() { + return Some(target); + } + } + if aec_disabled() { + return None; + } + let name = echo_cancel_source()?; + tracing::info!( + source = %name, + "mic capture targets the echo-cancelled source (PUNKTFUNK_NO_AEC=1 to disable)" + ); + Some(name) +} + +/// Find an existing echo-cancelled capture node: the first `Audio/Source` whose `node.name` +/// or description says echo-cancel (`module-echo-cancel`'s convention — `echo-cancel-*` +/// nodes, "Echo-Cancel …" descriptions; PulseAudio-compat setups match too). One registry +/// roundtrip via [`devices`]; any failure reads as "none". +fn echo_cancel_source() -> Option { + let (_, sources) = devices().ok()?; + sources.into_iter().find_map(|d| { + let name = d.name.to_ascii_lowercase(); + let desc = d.description.to_ascii_lowercase(); + (name.contains("echo-cancel") + || name.contains("echo_cancel") + || desc.contains("echo-cancel") + || desc.contains("echo cancel")) + .then_some(d.name) + }) +} + fn mic_thread( connector: &Arc, quit_rx: pipewire::channel::Receiver, @@ -384,9 +437,14 @@ fn mic_thread( PW_INIT.call_once(pw::init); let mut encoder = - opus::Encoder::new(SAMPLE_RATE, opus::Channels::Stereo, opus::Application::Voip) + opus::Encoder::new(SAMPLE_RATE, opus::Channels::Mono, opus::Application::Voip) .map_err(|e| anyhow::anyhow!("opus encoder: {e}"))?; - let _ = encoder.set_bitrate(opus::Bitrate::Bits(64_000)); + // Voice tuning: 48 kbps mono is transparent for speech; in-band FEC + an assumed 10 % + // loss let the host's decoder rebuild a lost 0xCB datagram from its successor instead + // of concealing (datagrams are fire-and-forget — this FEC is the only redundancy). + let _ = encoder.set_bitrate(opus::Bitrate::Bits(48_000)); + let _ = encoder.set_inband_fec(true); + let _ = encoder.set_packet_loss_perc(10); let mainloop = pw::main_loop::MainLoopRc::new(None).context("pw mic MainLoop")?; let context = pw::context::ContextRc::new(&mainloop, None).context("pw mic Context")?; @@ -405,14 +463,15 @@ fn mic_thread( *pw::keys::MEDIA_ROLE => "Communication", *pw::keys::NODE_NAME => "punktfunk-mic-capture", *pw::keys::NODE_DESCRIPTION => "Punktfunk Microphone", + // ~10 ms quantum (one mic frame). Without it the capture stream inherits the graph + // quantum — commonly 1024–2048 samples, so the mic arrived in 21–43 ms bursts that + // sat ahead of the encoder as latency (the playback stream always asked for 5 ms). + *pw::keys::NODE_LATENCY => "480/48000", }; - // The Settings microphone pick (`Settings::mic_device` via session main). - if let Ok(target) = std::env::var("PUNKTFUNK_AUDIO_SOURCE") { - if !target.is_empty() { - // Raw key: the `keys::TARGET_OBJECT` constant is feature-gated on a newer - // libpipewire than we require; the wire name is stable. - props.insert("target.object", target); - } + if let Some(target) = mic_capture_target() { + // Raw key: the `keys::TARGET_OBJECT` constant is feature-gated on a newer + // libpipewire than we require; the wire name is stable. + props.insert("target.object", target); } let stream = pw::stream::StreamBox::new(&core, "punktfunk-mic-capture", props) .context("pw mic Stream")?; @@ -447,9 +506,9 @@ fn mic_thread( .push_back(f32::from_le_bytes([s[0], s[1], s[2], s[3]])); } } - // Ship every complete 20 ms stereo frame. - while ud.ring.len() >= MIC_FRAME * CHANNELS { - let pcm: Vec = ud.ring.drain(..MIC_FRAME * CHANNELS).collect(); + // Ship every complete 10 ms mono frame. + while ud.ring.len() >= MIC_FRAME * MIC_CHANNELS { + let pcm: Vec = ud.ring.drain(..MIC_FRAME * MIC_CHANNELS).collect(); match ud.encoder.encode_float(&pcm, &mut ud.out) { Ok(len) => { let pts = std::time::SystemTime::now() @@ -473,7 +532,8 @@ fn mic_thread( let mut info = AudioInfoRaw::new(); info.set_format(AudioFormat::F32LE); info.set_rate(SAMPLE_RATE); - info.set_channels(CHANNELS as u32); + // Mono: the stream's adapter downmixes whatever layout the source really has. + info.set_channels(MIC_CHANNELS as u32); let obj = pw::spa::pod::Object { type_: pw::spa::utils::SpaTypes::ObjectParamFormat.as_raw(), id: pw::spa::param::ParamType::EnumFormat.as_raw(), diff --git a/crates/pf-client-core/src/audio_wasapi.rs b/crates/pf-client-core/src/audio_wasapi.rs index 5fb1de9f..4b191df9 100644 --- a/crates/pf-client-core/src/audio_wasapi.rs +++ b/crates/pf-client-core/src/audio_wasapi.rs @@ -23,14 +23,22 @@ use std::sync::atomic::{AtomicBool, Ordering}; use std::sync::mpsc::{Receiver, SyncSender, TrySendError}; use std::sync::Arc; use std::time::Duration; -use wasapi::{DeviceEnumerator, Direction, SampleType, StreamMode, WaveFormat}; +use wasapi::{ + AudioClientProperties, DeviceEnumerator, Direction, SampleType, StreamCategory, StreamMode, + WaveFormat, +}; const SAMPLE_RATE: usize = 48_000; -/// The microphone uplink stays stereo (the host's virtual mic is stereo). The render path is -/// multichannel — its channel count + block align are runtime, driven by the host-resolved layout. -const CHANNELS: usize = 2; -/// Mic frames are 20 ms (960 samples/channel) — any size ≤ 120 ms is fine host-side. -const MIC_FRAME: usize = 960; +/// Mic capture requests STEREO from WASAPI (autoconvert matrixes any endpoint layout down to +/// it — the proven path; `read_from_device_to_deque` then delivers our requested format) and +/// downmixes to MONO in code before the encoder: voice is mono at the source, the host accepts +/// any Opus channel layout (its stereo decoder upmixes), and half the samples halve the encode +/// + wire cost. The render path is multichannel — its channel count + block align are runtime, +/// driven by the host-resolved layout. +const CAPT_CHANNELS: usize = 2; +/// Mic frames are 10 ms (480 mono samples) — any size ≤ 120 ms is fine host-side; 10 ms +/// halves the frame-fill share of mouth-to-ear latency vs the old 20 ms. +const MIC_FRAME: usize = 480; pub struct AudioPlayer { pcm_tx: SyncSender>, @@ -217,8 +225,8 @@ fn render_thread( res } -/// The microphone uplink: capture the default input device, Opus-encode 20 ms chunks, ship -/// them as 0xCB datagrams into the host's virtual mic source. +/// The microphone uplink: capture the default input device, Opus-encode 10 ms mono chunks, +/// ship them as 0xCB datagrams into the host's virtual mic source. pub struct MicStreamer { stop: Arc, thread: Option>, @@ -259,18 +267,37 @@ fn mic_thread(connector: &Arc, stop: Arc) -> Result<() let mut encoder = opus::Encoder::new( SAMPLE_RATE as u32, - opus::Channels::Stereo, + opus::Channels::Mono, opus::Application::Voip, ) .map_err(|e| anyhow!("opus encoder: {e}"))?; - let _ = encoder.set_bitrate(opus::Bitrate::Bits(64_000)); + // Voice tuning: 48 kbps mono is transparent for speech; in-band FEC + an assumed 10 % + // loss let the host's decoder rebuild a lost 0xCB datagram from its successor instead + // of concealing (datagrams are fire-and-forget — this FEC is the only redundancy). + let _ = encoder.set_bitrate(opus::Bitrate::Bits(48_000)); + let _ = encoder.set_inband_fec(true); + let _ = encoder.set_packet_loss_perc(10); let device = DeviceEnumerator::new() .context("DeviceEnumerator")? .get_default_device(&Direction::Capture) .context("default capture endpoint (no microphone?)")?; let mut audio_client = device.get_iaudioclient().context("IAudioClient")?; - let desired = WaveFormat::new(32, 32, &SampleType::Float, SAMPLE_RATE, CHANNELS, None); + // Communications category → the endpoint's communications signal-processing chain. A + // driver/APO stack with an echo canceller only engages it for communications-category + // streams; the default (Other) category never did, so the downlink audio playing on + // this box fed straight back into the host's virtual mic. Must precede Initialize + // (SetClientProperties is a pre-init call; the wasapi crate QIs IAudioClient2 inside). + // Best-effort: an endpoint without IAudioClient2 just keeps the default category. + // PUNKTFUNK_NO_AEC=1 opts out (same lever as the Linux echo-cancel-source preference). + if !std::env::var("PUNKTFUNK_NO_AEC").is_ok_and(|v| !v.is_empty() && v != "0") { + if let Err(e) = audio_client.set_properties( + AudioClientProperties::new().set_category(StreamCategory::Communications), + ) { + tracing::debug!(error = %e, "mic capture: Communications category not set"); + } + } + let desired = WaveFormat::new(32, 32, &SampleType::Float, SAMPLE_RATE, CAPT_CHANNELS, None); let (default_period, _min_period) = audio_client.get_device_period().context("device period")?; let mode = StreamMode::EventsShared { @@ -308,13 +335,23 @@ fn mic_thread(connector: &Arc, stop: Arc) -> Result<() Err(e) => return Err(anyhow!("get_next_packet_size: {e}")), } } - let whole = (bytes.len() / 4) * 4; - for c in bytes.drain(..whole).collect::>().chunks_exact(4) { - ring.push_back(f32::from_le_bytes([c[0], c[1], c[2], c[3]])); + // One stereo capture frame (8 bytes) → one mono sample: average L/R. Autoconvert + // already matrixed the endpoint's real layout (mono/stereo/array mic) into the + // stereo stream we initialized, so this is the only downmix left to do. + let stereo_frame = 4 * CAPT_CHANNELS; + let whole = (bytes.len() / stereo_frame) * stereo_frame; + for c in bytes + .drain(..whole) + .collect::>() + .chunks_exact(stereo_frame) + { + let l = f32::from_le_bytes([c[0], c[1], c[2], c[3]]); + let r = f32::from_le_bytes([c[4], c[5], c[6], c[7]]); + ring.push_back((l + r) * 0.5); } - // Ship every complete 20 ms stereo frame. - while ring.len() >= MIC_FRAME * CHANNELS { - let pcm: Vec = ring.drain(..MIC_FRAME * CHANNELS).collect(); + // Ship every complete 10 ms mono frame. + while ring.len() >= MIC_FRAME { + let pcm: Vec = ring.drain(..MIC_FRAME).collect(); match encoder.encode_float(&pcm, &mut out) { Ok(len) => { let pts = std::time::SystemTime::now() From 55e01b4dcbc824ce62cc46cdc65f0f138572973a Mon Sep 17 00:00:00 2001 From: enricobuehler Date: Fri, 31 Jul 2026 21:49:12 +0200 Subject: [PATCH 3/3] feat(client): the stats overlay learns whether your mic frames arrive MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit NativeClient::mic_stats reaches the per-second stream window: sent and dropped (queue-full + stale-shed) frame deltas join the tracing line and the Stats event, and the Detailed OSD tier renders a mic line while the uplink is live — a healthy 10 ms-frame mic reads ~100 f/s, and a drop term means the client is shedding backlog, not the network eating audio. Co-Authored-By: Claude Fable 5 --- crates/pf-client-core/src/session.rs | 19 ++++++++++++++++ crates/pf-presenter/src/run.rs | 34 ++++++++++++++++++++++++++++ 2 files changed, 53 insertions(+) diff --git a/crates/pf-client-core/src/session.rs b/crates/pf-client-core/src/session.rs index 93c6dc38..0773e116 100644 --- a/crates/pf-client-core/src/session.rs +++ b/crates/pf-client-core/src/session.rs @@ -121,6 +121,12 @@ pub struct Stats { /// received+lost (%). The OSD renders the counter line only when nonzero. pub lost: u32, pub lost_pct: f32, + /// Mic uplink frames this window: handed to the QUIC datagram send, and shed anywhere + /// client-side (queue-full at the producer + the pump's stale-oldest backlog governor — + /// see [`NativeClient::mic_stats`]). Both stay 0 while the mic is off, so the OSD + /// renders the mic line only when the uplink is live. + pub mic_sent: u32, + pub mic_dropped: u32, /// The decode path frames actually took this window (`"vaapi"`/`"software"`, empty /// until the first frame) — the OSD's trailing tag; tracks a mid-session fallback. pub decoder: &'static str, @@ -434,6 +440,9 @@ fn pump( let mut dec_path: &'static str = ""; // The stats window keeps its own drop cursor — the OSD shows the per-window delta. let mut window_dropped = connector.frames_dropped(); + // Mic uplink cursor (same per-window diffing): a healthy 10 ms-frame mic reads ~100 + // sent/s; a nonzero drop delta is the queue shedding backlog (see NativeClient::mic_stats). + let mut window_mic = connector.mic_stats(); let mut last_kf_req: Option = None; // Freeze-until-reanchor: the shared post-loss gate ([`punktfunk_core::reanchor::ReanchorGate`]). // Armed on any loss signal (frame-index gap, dropped-count climb, decoder wedge/demotion), it @@ -789,6 +798,12 @@ fn pump( let (pace_p50, _) = window_percentiles(&mut pace_us_win); let lost = dropped.saturating_sub(window_dropped) as u32; window_dropped = dropped; + let mic_now = connector.mic_stats(); + let mic_sent = mic_now.sent.saturating_sub(window_mic.sent) as u32; + let mic_dropped = (mic_now.dropped_full + mic_now.dropped_stale) + .saturating_sub(window_mic.dropped_full + window_mic.dropped_stale) + as u32; + window_mic = mic_now; tracing::debug!( fps = frames_n, hostnet_p50_us = hn_p50, @@ -800,6 +815,8 @@ fn pump( pace_p50_us = pace_p50, decode_p50_us = dec_p50, lost, + mic_sent, + mic_dropped, total_frames, "stream window" ); @@ -822,6 +839,8 @@ fn pump( } else { 0.0 }, + mic_sent, + mic_dropped, decoder: dec_path, })); window_start = Instant::now(); diff --git a/crates/pf-presenter/src/run.rs b/crates/pf-presenter/src/run.rs index 50c62fad..123b5a53 100644 --- a/crates/pf-presenter/src/run.rs +++ b/crates/pf-presenter/src/run.rs @@ -2017,6 +2017,14 @@ fn stats_text( if s.lost > 0 { text.push_str(&format!("\nlost {} ({:.1}%)", s.lost, s.lost_pct)); } + // The mic uplink line renders only while the mic is live (a healthy 10 ms-frame uplink + // reads ~100 f/s) and only in Detailed — drops here are the client shedding backlog. + if detailed && (s.mic_sent > 0 || s.mic_dropped > 0) { + text.push_str(&format!("\nmic {} f/s", s.mic_sent)); + if s.mic_dropped > 0 { + text.push_str(&format!(" · dropped {}", s.mic_dropped)); + } + } text } @@ -2262,6 +2270,8 @@ mod tests { decode_ms: 1.8, lost: 3, lost_pct: 0.4, + mic_sent: 0, + mic_dropped: 0, decoder: "vulkan", }, PresentedWindow { @@ -2303,6 +2313,30 @@ mod tests { ); } + /// The mic uplink line: Detailed-only, and only while the uplink is live. + #[test] + fn stats_text_mic_line() { + let (mut s, p) = sample(); + let text = |s: &Stats, v| stats_text(v, "m", s, &p, false, false, None); + assert!( + !text(&s, StatsVerbosity::Detailed).contains("mic"), + "no mic line while the mic is off" + ); + s.mic_sent = 100; + let detailed = text(&s, StatsVerbosity::Detailed); + assert!(detailed.contains("\nmic 100 f/s")); + assert!( + !detailed.contains("dropped"), + "a healthy uplink shows no drop term" + ); + assert!( + !text(&s, StatsVerbosity::Normal).contains("mic"), + "mic line is Detailed-only" + ); + s.mic_dropped = 7; + assert!(text(&s, StatsVerbosity::Detailed).contains("mic 100 f/s · dropped 7")); + } + /// Compact omits the latency term until the presenter's first e2e window lands. #[test] fn compact_waits_for_e2e() {