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.