refactor(slice-1): PacketIo trait — test seam for RtcSession.socket

Adversarial review's test-coverage gap finding: no test drives
`loop_driver::drive()`, the central function of the slice. F1, F2,
and F4 would have been caught by one. The obstacle was that `drive()`
called `session.socket.recv_from` on a concrete `UdpSocket`, so
unit tests couldn't feed synthetic STUN/RTP/DTLS bytes without a real
UDP loopback pair + hand-rolled valid packets on the wire.

Retrofit a sans-IO test seam shaped the way str0m itself is built:

- New `crates/rutster-media/src/packet_io.rs` defines `PacketIo`
  (`recv_from`, `send_to`) with a blanket impl for `UdpSocket`.
- `RtcSession.socket` goes from `std::net::UdpSocket` to
  `Box<dyn PacketIo + Send>`. Dispatch cost: a vtable call (~1 ns)
  on a 20 ms tick at single-digit pps; ~10,000× headroom vs the 200 ms
  latency bar. If step 4's dedicated timing thread measures it, swap
  `Box<dyn>` → a generic `RtcSession<S>` parameterization (same
  trait interface).
- `RtcSession::new()` keeps its behavior; new hidden `new_with_socket`
  is the test constructor (injects a fake + a synthetic local_addr).

This is the infrastructural refactor only — no behavior change. F4
and F1+F3 land on top of it (separate commits, TDD-style: failing
test first, then fix).

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

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<u8>`-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<dyn PacketIo + Send>`.
+//!
+//! ## Dispatch cost
+//!
+//! `Box<dyn PacketIo + Send>` 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<dyn>` → a generic `RtcSession<S>`
+//! parameterization; the trait interface stays.
+//!
+//! ## Send requirement
+//!
+//! `RtcSession` lives behind `Arc<Mutex<RtcSession>>` in the binary's
+//! `DashMap` and is driven from a tokio task — it must be `Send`. The
+//! `+ Send` bound on `Box<dyn PacketIo + Send>` 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<usize>;
+}
+
+/// 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<usize> {
+        std::net::UdpSocket::send_to(self, buf, dst)
+    }
+}

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<dyn PacketIo + Send>` 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<dyn PacketIo + Send>,
     /// 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, RtcSessionError> {
-        Self::new_internal()
+        // Bind a real non-blocking UDP socket. The `Box<dyn PacketIo>` 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<u8>`-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<dyn PacketIo + Send>,
+        local_addr: SocketAddr,
+    ) -> Result<Self, RtcSessionError> {
+        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<Self, RtcSessionError> {
-        // 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 {