diff --git a/crates/rutster-tap/src/lib.rs b/crates/rutster-tap/src/lib.rs index 080d422..7015f16 100644 --- a/crates/rutster-tap/src/lib.rs +++ b/crates/rutster-tap/src/lib.rs @@ -27,18 +27,22 @@ //! — that would invert the canonical-home of `PcmFrame` and pull the //! loopback peer into the tap story. +pub mod metrics; pub mod protocol; +pub mod tap_audio_pipe; // Re-export PcmFrame so `rutster-tap` consumers (the binary, the echo brain, // future brains) get it from one canonical home (spec §3.1). pub use rutster_media::PcmFrame; +pub use metrics::{MetricsSnapshot, TapMetrics, TAP_PLAYOUT_FRAMES}; pub use protocol::{ decode_envelope, decode_pcm, encode_audio_in, encode_audio_out, encode_bye, encode_error, encode_hello, encode_pcm, encode_session_end, AudioPayload, DecodedFrame, DecodedPayload, Envelope, ErrorPayload, FrameKind, HelloPayload, Payload, ReasonPayload, SessionEndPayload, TapProtoError, PROTOCOL_VERSION, SAMPLES_PER_FRAME as WIRE_SAMPLES_PER_FRAME, }; +pub use tap_audio_pipe::TapAudioPipe; #[cfg(test)] mod tests { diff --git a/crates/rutster-tap/src/metrics.rs b/crates/rutster-tap/src/metrics.rs new file mode 100644 index 0000000..90c7e8e --- /dev/null +++ b/crates/rutster-tap/src/metrics.rs @@ -0,0 +1,62 @@ +//! # TapMetrics — the "observe" half of "drop + observe" (slice-1 §3.8) +//! +//! Atomic counters shared between `TapAudioPipe` (hot path — inbound/outbound +//! drops, playout overflow/underflow) and `TapClient` (seq gaps, unknown +//! frames, malformed frames, reconnect attempts). All ops are +//! `fetch_add(_, Ordering::Relaxed)` — we're using the counts for +//! observability, not synchronization, so relaxed ordering is correct + +//! cheapest. A snapshot read is taken at log time (e.g. once per second or +//! on tap-disconnect). + +use std::sync::atomic::{AtomicU64, Ordering}; +use std::sync::Arc; + +/// Bounded playout ring capacity (spec §4.1: 5 frames = 100 ms at 20 ms/frame). +/// Tunable *constant* (no runtime config in slice-2; a future-rung concern). +pub const TAP_PLAYOUT_FRAMES: usize = 5; + +#[derive(Debug, Default)] +pub struct TapMetrics { + pub inbound_dropped: AtomicU64, + pub outbound_dropped: AtomicU64, + pub playout_overflow: AtomicU64, + pub playout_underflow: AtomicU64, + pub seq_gaps: AtomicU64, + pub unknown_frames: AtomicU64, + pub malformed_frames: AtomicU64, + pub reconnect_attempts: AtomicU64, +} + +impl TapMetrics { + pub fn new() -> Arc { + Arc::new(Self::default()) + } + + /// Read a consistent-ish snapshot. Not atomic across fields (each is + /// independently read), but for log/observability purposes that's fine — + /// we're reporting "approximately this many X happened," not synchronizing. + pub fn snapshot(&self) -> MetricsSnapshot { + MetricsSnapshot { + inbound_dropped: self.inbound_dropped.load(Ordering::Relaxed), + outbound_dropped: self.outbound_dropped.load(Ordering::Relaxed), + playout_overflow: self.playout_overflow.load(Ordering::Relaxed), + playout_underflow: self.playout_underflow.load(Ordering::Relaxed), + seq_gaps: self.seq_gaps.load(Ordering::Relaxed), + unknown_frames: self.unknown_frames.load(Ordering::Relaxed), + malformed_frames: self.malformed_frames.load(Ordering::Relaxed), + reconnect_attempts: self.reconnect_attempts.load(Ordering::Relaxed), + } + } +} + +#[derive(Debug, Clone, Default)] +pub struct MetricsSnapshot { + pub inbound_dropped: u64, + pub outbound_dropped: u64, + pub playout_overflow: u64, + pub playout_underflow: u64, + pub seq_gaps: u64, + pub unknown_frames: u64, + pub malformed_frames: u64, + pub reconnect_attempts: u64, +} diff --git a/crates/rutster-tap/src/tap_audio_pipe.rs b/crates/rutster-tap/src/tap_audio_pipe.rs new file mode 100644 index 0000000..93705f8 --- /dev/null +++ b/crates/rutster-tap/src/tap_audio_pipe.rs @@ -0,0 +1,215 @@ +//! # TapAudioPipe — the seam object (spec §4.1) +//! +//! The sync object `RtcSession` holds and `loop_driver` calls via the +//! `AudioSource`/`AudioSink` trait seam. It's a thin wrapper over two +//! `tokio::sync::mpsc` channels + a bounded `VecDeque` playout ring: +//! +//! ```text +//! peer mic → Opus decode → on_pcm_frame() → tx_pcm_in → TapClient → audio_in (WS) +//! brain audio_out (WS) → TapClient → rx_audio_out → [playout ring] → next_pcm_frame() → Opus encode → peer +//! ``` +//! +//! # Why `VecDeque` for the playout ring (not `mpsc`?) +//! The playout ring has a specific policy on overflow (drop *oldest*, not +//! newest — spec §4.1). `mpsc` drops the *newest* on `try_send` when +//! full (the receiver would see a gap). For a real-time media path, we +//! want stale (old) frames dropped, not late (new) frames — drop-oldest +//! keeps the buffer at-or-behind real-time. A `VecDeque` under our +//! manual `push_back_bounded` policy is the smallest structure that fits. + +use std::collections::VecDeque; +use std::sync::atomic::Ordering; +use std::sync::Arc; + +use rutster_media::{AudioSink, AudioSource, PcmFrame}; +use tokio::sync::mpsc; +use tracing::{debug, trace}; + +use crate::metrics::{TapMetrics, TAP_PLAYOUT_FRAMES}; + +pub struct TapAudioPipe { + // Core → brain (inbound decoded PCM from peer): + tx_pcm_in: mpsc::Sender, + // Brain → core (playout buffer): + rx_audio_out: mpsc::Receiver, + // Bounded ring between rx_audio_out and next_pcm_frame (spec §4.1). + playout_ring: VecDeque, + metrics: Arc, +} + +impl TapAudioPipe { + pub fn new( + tx_pcm_in: mpsc::Sender, + rx_audio_out: mpsc::Receiver, + metrics: Arc, + ) -> Self { + Self { + tx_pcm_in, + rx_audio_out, + playout_ring: VecDeque::with_capacity(TAP_PLAYOUT_FRAMES), + metrics, + } + } + + /// Drain the inbound mpsc into the playout ring. Called by + /// `next_pcm_frame` (one drain per source tick). Returns the count + /// drained. **Hot-path policy:** every drop is counted (drop + observe). + fn drain_inbound(&mut self) -> usize { + let mut drained = 0; + loop { + match self.rx_audio_out.try_recv() { + Ok(frame) => { + // push_back_bounded: drop oldest if full (drop-oldest policy). + if self.playout_ring.len() >= TAP_PLAYOUT_FRAMES { + self.playout_ring.pop_front(); + self.metrics + .playout_overflow + .fetch_add(1, Ordering::Relaxed); + debug!(overflow = true, "playout ring overflow; dropped oldest"); + } + self.playout_ring.push_back(frame); + drained += 1; + } + Err(mpsc::error::TryRecvError::Empty) => break, + Err(mpsc::error::TryRecvError::Disconnected) => { + // Engine task gone. Silence until TapClient reconnects + // via a fresh mpsc (handled by TapEngine's reconnect loop). + trace!("rx_audio_out disconnected; silence until reconnect"); + break; + } + } + } + drained + } +} + +impl AudioSource for TapAudioPipe { + /// Take the next brain-proposed PCM frame to send to the peer. + /// `None` = silence (loop_driver emits Opus silence on None — slice-1). + fn next_pcm_frame(&mut self) -> Option { + // First drain any frames the engine task has queued; then pop one. + self.drain_inbound(); + match self.playout_ring.pop_front() { + Some(frame) => Some(frame), + None => { + // Underflow: brain slower than the 20ms tick. Silence. + self.metrics + .playout_underflow + .fetch_add(1, Ordering::Relaxed); + None + } + } + } +} + +impl AudioSink for TapAudioPipe { + /// Receive a decoded PCM frame from the peer. Must not block + /// (slice-1 §3.3 contract — `on_pcm_frame` runs in the 20ms loop). + /// `try_send` (not `send`) — if the channel is full (engine task slow + /// or gone), drop + count (hot-path "drop + observe, don't crash"). + fn on_pcm_frame(&mut self, frame: PcmFrame) { + if self.tx_pcm_in.try_send(frame).is_err() { + self.metrics.inbound_dropped.fetch_add(1, Ordering::Relaxed); + trace!("inbound PCM dropped (channel full)"); + } + } +} + +#[cfg(test)] +mod tests { + use super::*; + + #[allow(clippy::type_complexity)] // test helper: 5-tuple of channel ends; not worth a struct. + fn channels() -> ( + mpsc::Sender, + mpsc::Receiver, + mpsc::Sender, + mpsc::Receiver, + Arc, + ) { + // tx_pcm_in (inbound PCM → engine), rx_audio_out (engine → playout ring) + let (tx_pcm_in, rx_pcm_in) = mpsc::channel(64); + let (tx_audio_out, rx_audio_out) = mpsc::channel(64); + let metrics = TapMetrics::new(); + (tx_pcm_in, rx_pcm_in, tx_audio_out, rx_audio_out, metrics) + } + + #[test] + fn source_returns_none_on_empty_ring() { + let (_, _, tx_audio_out, rx_audio_out, metrics) = channels(); + let mut pipe = TapAudioPipe::new(tx_audio_out, rx_audio_out, metrics); + assert!(pipe.next_pcm_frame().is_none()); + } + + #[test] + fn source_returns_frame_pushed_by_engine() { + let (_, _, tx_audio_out, rx_audio_out, metrics) = channels(); + let mut pipe = TapAudioPipe::new(tx_audio_out.clone(), rx_audio_out, metrics.clone()); + // Engine task (simulated): push an audio_out frame. + let mut frame = PcmFrame::zeroed(); + frame.samples[0] = 42; + tx_audio_out.blocking_send(frame).unwrap(); + // Source should now drain + return it. + let got = pipe.next_pcm_frame().expect("frame present"); + assert_eq!(got.samples[0], 42); + assert_eq!(metrics.playout_underflow.load(Ordering::Relaxed), 0); + } + + #[test] + fn overflow_drops_oldest_not_newest() { + let (_, _, tx_audio_out, rx_audio_out, metrics) = channels(); + let mut pipe = TapAudioPipe::new(tx_audio_out.clone(), rx_audio_out, metrics.clone()); + // Push TAP_PLAYOUT_FRAMES + 3 frames. The first 3 should be dropped + // (oldest), keeping the last TAP_PLAYOUT_FRAMES. + for i in 0..(TAP_PLAYOUT_FRAMES + 3) { + let mut f = PcmFrame::zeroed(); + f.samples[0] = i as i16; + tx_audio_out.blocking_send(f).unwrap(); + } + // Drain into the ring via next_pcm_frame's drain step (returns the + // first popped, so we call N+1 times to fully drain + assert). + let first = pipe.next_pcm_frame().expect("first frame after drain"); + assert_eq!(first.samples[0], 3); // first 3 (0,1,2) were dropped + assert_eq!(metrics.playout_overflow.load(Ordering::Relaxed), 3); + } + + #[test] + fn underflow_increments_counter() { + let (_, _, tx_audio_out, rx_audio_out, metrics) = channels(); + let mut pipe = TapAudioPipe::new(tx_audio_out, rx_audio_out, metrics.clone()); + pipe.next_pcm_frame(); // None -> underflow + pipe.next_pcm_frame(); // None -> underflow + assert_eq!(metrics.playout_underflow.load(Ordering::Relaxed), 2); + } + + #[test] + fn sink_full_drops_and_counts() { + // Tiny inbound channel to force overflow. + let (tx_pcm_in, _rx_pcm_in, _tx_audio_out, rx_audio_out, metrics) = channels(); + let mut pipe = TapAudioPipe::new(tx_pcm_in, rx_audio_out, metrics.clone()); + // Channel capacity 64 — push 100 frames. + for _ in 0..100 { + pipe.on_pcm_frame(PcmFrame::zeroed()); + } + assert!(metrics.inbound_dropped.load(Ordering::Relaxed) > 0); + } + + #[test] + fn disconnected_engine_returns_none() { + let (_, _, tx_audio_out, rx_audio_out, metrics) = channels(); + let mut pipe = TapAudioPipe::new(tx_audio_out, rx_audio_out, metrics); + // Drop the sender → rx_audio_out sees Disconnected → None (silence). + // (Implicit drop at scope exit here; explicit next_pcm_frame below + // reads Disconnected.) + assert!(pipe.next_pcm_frame().is_none()); + } + + #[test] + fn disconnected_engine_returns_none_after_close() { + let (tx_pcm_in, _rx_pcm_in, tx_audio_out, rx_audio_out, metrics) = channels(); + let mut pipe = TapAudioPipe::new(tx_pcm_in, rx_audio_out, metrics); + drop(tx_audio_out); // close the engine→playout-ring direction + // Now next_pcm_frame should return None (silence) — Disconnected path. + assert!(pipe.next_pcm_frame().is_none()); + } +}