Merge branch 'audio/core-desktop-uplink' into audio/mic-latency-echo

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<Terminate>,
     thread: Option<std::thread::JoinHandle<()>>,
@@ -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<NativeClient>,
     ring: VecDeque<f32>,
@@ -371,6 +375,55 @@ struct MicData {
     out: Vec<u8>,
 }
 
+/// `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<String> {
+    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<String> {
+    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<NativeClient>,
     quit_rx: pipewire::channel::Receiver<Terminate>,
@@ -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<f32> = 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<f32> = 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(),

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<Vec<f32>>,
@@ -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<AtomicBool>,
     thread: Option<std::thread::JoinHandle<()>>,
@@ -259,18 +267,37 @@ fn mic_thread(connector: &Arc<NativeClient>, stop: Arc<AtomicBool>) -> 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<NativeClient>, stop: Arc<AtomicBool>) -> Result<()
                 Err(e) => return Err(anyhow!("get_next_packet_size: {e}")),
             }
         }
-        let whole = (bytes.len() / 4) * 4;
-        for c in bytes.drain(..whole).collect::<Vec<u8>>().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::<Vec<u8>>()
+            .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<f32> = ring.drain(..MIC_FRAME * CHANNELS).collect();
+        // Ship every complete 10 ms mono frame.
+        while ring.len() >= MIC_FRAME {
+            let pcm: Vec<f32> = ring.drain(..MIC_FRAME).collect();
             match encoder.encode_float(&pcm, &mut out) {
                 Ok(len) => {
                     let pts = std::time::SystemTime::now()

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<Instant> = 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();

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() {

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<Receiver<crate::quic::CursorState>>,
     input_tx: tokio::sync::mpsc::UnboundedSender<InputEvent>,
     /// 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<u8>)>,
+    /// Mic uplink counters (sent / dropped per stage) — see [`NativeClient::mic_stats`].
+    mic_stats: Arc<MicUplinkCounters>,
     /// 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(())
             }

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);
         }
     });
 

crates/punktfunk-core/src/client/worker.rs

@@ -66,6 +66,9 @@ pub(crate) struct WorkerArgs {
     pub(crate) probe: Arc<Mutex<ProbeState>>,
     pub(crate) frames_dropped: Arc<AtomicU64>,
     pub(crate) fec_recovered: Arc<AtomicU64>,
+    /// 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<MicUplinkCounters>,
     pub(crate) hot_tids: Arc<Mutex<Vec<i32>>>,
     /// 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.