diff --git a/crates/rutster-media/src/lib.rs b/crates/rutster-media/src/lib.rs index 00bfef8..948d077 100644 --- a/crates/rutster-media/src/lib.rs +++ b/crates/rutster-media/src/lib.rs @@ -27,15 +27,20 @@ //! - [`pcm`] — `PcmFrame` + `AudioSource`/`AudioSink` traits (the tap //! seam) + `EchoAudioPipe` (slice-1 wiring). //! - [`opus_codec`] — `OpusDecoder`/`OpusEncoder` wrappers. +//! - [`packet_io`] — `PacketIo` trait: the test seam for `RtcSession`'s +//! socket (str0m sans-IO posture retrofitted per the slice-1 review +//! test-gap finding). //! - [`loop_driver`] (Task 4) — the str0m poll loop on tokio. //! - [`rtc_session`] (Task 4) — `RtcSession`, the per-peer owner. pub mod loop_driver; pub mod opus_codec; +pub mod packet_io; pub mod pcm; pub mod rtc_session; pub use opus_codec::{OpusDecoder, OpusEncoder}; +pub use packet_io::PacketIo; pub use pcm::{AudioPipe, AudioSink, AudioSource, EchoAudioPipe, PcmFrame, SAMPLES_PER_FRAME}; pub use rtc_session::{RtcSession, RtcSessionError}; diff --git a/crates/rutster-media/src/packet_io.rs b/crates/rutster-media/src/packet_io.rs new file mode 100644 index 0000000..e7742c4 --- /dev/null +++ b/crates/rutster-media/src/packet_io.rs @@ -0,0 +1,64 @@ +//! # Packet I/O trait — the test seam for `RtcSession`'s socket +//! +//! Production uses `std::net::UdpSocket`; tests use a `Vec`-backed +//! fake to drive `loop_driver::drive()` without a real network round-trip. +//! This is the sans-IO test pattern str0m itself is built around, retrofitted +//! onto `RtcSession` per the slice-1 review's "test-coverage gap" finding. +//! +//! ## Why a trait (not a concrete `UdpSocket`) +//! +//! Adversarial review F1/F4 regressed against a missing `loop_driver::drive()` +//! test harness: the pre-refactor `drive()` called `session.socket.recv_from` +//! on the *concrete* `UdpSocket`, so unit tests couldn't feed synthetic STUN / +//! RTP / DTLS bytes without spinning up a real UDP loopback pair + hand-rolled +//! valid packets on the wire. The trait is the seam that lets tests inject +//! arbitrary bytes through `Box`. +//! +//! ## Dispatch cost +//! +//! `Box` is a vtable call (~1 ns) on a 20 ms tick at +//! single-digit packets per cycle. The spec's 200 ms latency bar (~10,000× +//! the dispatch cost) absorbs this trivially; the slice-1 hot path's +//! privileged work is the Opus encode/decode + media-time arithmetic, not +//! the I/O dispatch. If the step-4 dedicated timing thread ever measures +//! the vtable as meaningful, swap `Box` → a generic `RtcSession` +//! parameterization; the trait interface stays. +//! +//! ## Send requirement +//! +//! `RtcSession` lives behind `Arc>` in the binary's +//! `DashMap` and is driven from a tokio task — it must be `Send`. The +//! `+ Send` bound on `Box` propagates that requirement +//! to the test fake too (which is fine: a `VecDeque` is `Send`). + +use std::io; +use std::net::SocketAddr; + +/// The bidirectional UDP-like I/O surface `loop_driver::drive()` consumes. +/// +/// Production: `std::net::UdpSocket` configured non-blocking. Tests: a fake +/// queuing inbound `(bytes, source)` tuples and recording outbound sends. +pub trait PacketIo: Send { + /// Non-blocking recv. Returns `Err(io::ErrorKind::WouldBlock)` when + /// the queue is empty, mirroring `UdpSocket::set_nonblocking(true)`. + fn recv_from(&mut self, buf: &mut [u8]) -> io::Result<(usize, SocketAddr)>; + + /// Send a datagram. Failures other than `WouldBlock` are logged and + /// dropped by the caller — the hot-path "drop + observe" policy + /// (spec §3.8) applies here too; we do NOT `?`-propagate on the loop. + fn send_to(&mut self, buf: &[u8], dst: SocketAddr) -> io::Result; +} + +/// Blanket forwarding to the inherent `UdpSocket` methods. The methods on +/// `std::net::UdpSocket` take `&self` (not `&mut self`); the trait requires +/// `&mut self` for symmetry with test fakes that mutate on `recv_from` +/// (popping a `VecDeque`). The shared-reborrow from `&mut self → &self` +/// makes both shapes work. +impl PacketIo for std::net::UdpSocket { + fn recv_from(&mut self, buf: &mut [u8]) -> io::Result<(usize, SocketAddr)> { + std::net::UdpSocket::recv_from(self, buf) + } + fn send_to(&mut self, buf: &[u8], dst: SocketAddr) -> io::Result { + std::net::UdpSocket::send_to(self, buf, dst) + } +} diff --git a/crates/rutster-media/src/rtc_session.rs b/crates/rutster-media/src/rtc_session.rs index 931acd3..f7c3769 100644 --- a/crates/rutster-media/src/rtc_session.rs +++ b/crates/rutster-media/src/rtc_session.rs @@ -24,6 +24,7 @@ use str0m::Rtc; use thiserror::Error; use crate::opus_codec::{OpusDecoder, OpusEncoder}; +use crate::packet_io::PacketIo; use crate::pcm::{AudioPipe, EchoAudioPipe}; /// Per-session idle timeout (spec §4.5): 60 s of no RTP from the peer @@ -89,7 +90,13 @@ pub struct RtcSession { /// receives `Input::Receive` packets from. Bound to an ephemeral /// port at construction; the local candidate passed to str0m at /// offer-accept time uses this address. - pub(crate) socket: std::net::UdpSocket, + /// + /// `Box` instead of a concrete `UdpSocket`: + /// the trait is the test seam `loop_driver::drive()` needs to feed + /// synthetic STUN/RTP/DTLS bytes without a real UDP loopback pair + /// (slice-1 review "test-coverage gap" finding). Production + /// constructs via `new()`; tests via `new_with_socket()`. + pub(crate) socket: Box, /// Local socket address — cached because `local_addr()` is a syscall. pub(crate) local_addr: SocketAddr, /// Mid of the audio m-line we accepted. Set during `accept_offer`. @@ -120,7 +127,48 @@ impl RtcSession { /// `for_server` split; the body is identical (binding a UDP socket /// on `0.0.0.0:0`, constructing the `Rtc` + codecs). pub fn new() -> Result { - Self::new_internal() + // Bind a real non-blocking UDP socket. The `Box` is + // a vtable wrapper (~1 ns dispatch) — see `packet_io.rs` for the + // rationale (slice-1 review test-harness retrofit). + // + // We bind to 127.0.0.1 (not 0.0.0.0) because the local candidate + // we hand to str0m uses this socket's address, and str0m's + // `Candidate::host` rejects the unspecified address 0.0.0.0 as + // invalid. Slice 1 is loopback-only; production binding (when + // the binary lands in Task 5) can rebind to a real interface + // address before `accept_offer` constructs the candidate. + let socket = std::net::UdpSocket::bind("127.0.0.1:0")?; + socket.set_nonblocking(true)?; + let local_addr = socket.local_addr()?; + Self::new_with_socket(Box::new(socket), local_addr) + } + + /// Test constructor: inject a fake `PacketIo` (e.g. `Vec`-backed) + /// and a synthetic `local_addr`. The rest of the construction mirrors + /// `new()` exactly. Production paths must use `new()`. + /// + /// Returns `Result` because codec init can fail; tests that don't + /// need a fake socket should use `RtcSession::new()?`. + #[doc(hidden)] + pub fn new_with_socket( + socket: Box, + local_addr: SocketAddr, + ) -> Result { + let rtc = Rtc::new(Instant::now()); + Ok(Self { + channel: Channel::new_inbound(), + rtc, + decoder: OpusDecoder::new()?, + encoder: OpusEncoder::new()?, + pipe: Box::new(EchoAudioPipe::new()), + socket, + local_addr, + audio_mid: None, + next_timeout: None, + last_rx: Instant::now(), + last_outbound_at: Instant::now(), + next_media_time: str0m::media::MediaTime::ZERO, + }) } /// Replace the default `EchoAudioPipe` with a real tap pipe (slice-2, @@ -169,41 +217,6 @@ impl RtcSession { self.pipe.clear_playout_ring(); } - fn new_internal() -> Result { - // Bind an ephemeral UDP socket. We use std::net::UdpSocket and - // drive it non-blocking from tokio rather than tokio's UdpSocket: - // str0m operates on raw `Receive` values and yields `Transmit` - // values, both of which are plain structs — no async needed. - // Setting non-blocking lets us `recv_from` without blocking. - // - // We bind to 127.0.0.1 (not 0.0.0.0) because the local candidate - // we hand to str0m uses this socket's address, and str0m's - // `Candidate::host` rejects the unspecified address 0.0.0.0 as - // invalid. Slice 1 is loopback-only; production binding (when - // the binary lands in Task 5) can rebind to a real interface - // address before `accept_offer` constructs the candidate. - let socket = std::net::UdpSocket::bind("127.0.0.1:0")?; - socket.set_nonblocking(true)?; - let local_addr = socket.local_addr()?; - - let rtc = Rtc::new(Instant::now()); - - Ok(Self { - channel: Channel::new_inbound(), - rtc, - decoder: OpusDecoder::new()?, - encoder: OpusEncoder::new()?, - pipe: Box::new(EchoAudioPipe::new()), - socket, - local_addr, - audio_mid: None, - next_timeout: None, - last_rx: Instant::now(), - last_outbound_at: Instant::now(), - next_media_time: str0m::media::MediaTime::ZERO, - }) - } - /// Returns the session's `ChannelId` — also the REST API session id /// surfaced at `/v1/sessions/{id}` (spec §4.5). pub fn channel_id(&self) -> ChannelId {