1f0f64c1c6
Four routes on axum 0.7 per spec §4.1: POST /v1/sessions (mint), POST /v1/sessions/:id/offer (str0m-native SDP accept), DELETE /v1/sessions/:id (close), GET / (static HTML client). Session store is a DashMap<ChannelId, Arc<Mutex<RtcSession>>> (spec §4.5); one tokio task drives all session poll loops — per-session tasks would pre-pave the wrong pattern for step 4's dedicated thread. Graceful shutdown drops the DashMap on Ctrl-C / SIGTERM. Integration test exercises the REST surface; manual browser e2e per README §6.5.
301 lines
13 KiB
Rust
301 lines
13 KiB
Rust
//! # `RtcSession` — the per-peer media owner (spec §3.1, §4.5)
|
||
//!
|
||
//! Owns a `str0m::Rtc` instance + an Opus decoder/encoder pair + an
|
||
//! `EchoAudioPipe` wiring inbound to outbound + the per-peer UDP socket.
|
||
//! One per WebRTC peer. The `ChannelId` (from `rutster-call-model`) is
|
||
//! the session id surfaced in the REST API.
|
||
//!
|
||
//! ## What str0m does for us (so we don't)
|
||
//!
|
||
//! str0m 0.21's `Rtc::sdp_api().accept_offer(offer)` produces the SDP
|
||
//! answer natively: DTLS fingerprint (from the cert str0m generates), ICE
|
||
//! ufrag/pwd, and codec negotiation (Opus, the only codec we registered).
|
||
//! Slice 1 does NOT hand-roll an SDP munger — str0m's path is the spec's
|
||
//! "embryo of the future SIP SDP path" (§3.7). When step 5 brings SIP/SDP
|
||
//! negotiation into `rutster-signaling-sip`, that crate may extract shared
|
||
//! SDP helpers from str0m or build its own. Slice 1's WebRTC-ICE-coupled
|
||
//! SDP lives entirely in str0m.
|
||
|
||
use std::net::SocketAddr;
|
||
use std::time::{Duration, Instant};
|
||
|
||
use rutster_call_model::{Channel, ChannelId, ChannelState};
|
||
use str0m::Rtc;
|
||
use thiserror::Error;
|
||
|
||
use crate::opus_codec::{OpusDecoder, OpusEncoder};
|
||
use crate::pcm::EchoAudioPipe;
|
||
|
||
/// Per-session idle timeout (spec §4.5): 60 s of no RTP from the peer
|
||
/// → close. RTC quiet periods are normal but 60 s of dead air means
|
||
/// "the browser tab is dead."
|
||
pub(crate) const IDLE_TIMEOUT: Duration = Duration::from_secs(60);
|
||
|
||
/// Cold-path errors for `RtcSession` construction and SDP negotiation.
|
||
///
|
||
/// Hot-path failures (decode, encode, UDP recv) follow the
|
||
/// match-and-continue "drop + observe" policy from spec §3.8 — they
|
||
/// never reach this enum.
|
||
#[derive(Debug, Error)]
|
||
pub enum RtcSessionError {
|
||
/// Two-stage failure from str0m's SDP path: `SdpOffer::from_sdp_string`
|
||
/// can fail to parse, OR `accept_offer` can reject the parsed offer.
|
||
/// Both surface as `str0m::RtcError` / `str0m::sdp::SdpError`; we
|
||
/// collapse them via a display-format `String` since both are
|
||
/// display-format-only at the axum boundary (HTTP 400 in `routes.rs`).
|
||
/// Using `String` (not `#[source]`) keeps the variants uniform and the
|
||
/// axum layer doesn't need to downcast — it logs + 400s.
|
||
#[error("SDP offer parse or accept failed: {0}")]
|
||
SdpOffer(String),
|
||
#[error("opus codec init failed: {0}")]
|
||
Codec(#[from] opus::Error),
|
||
#[error("UDP socket bind failed: {0}")]
|
||
Socket(#[from] std::io::Error),
|
||
}
|
||
|
||
use str0m::change::SdpOffer;
|
||
use str0m::media::Mid;
|
||
|
||
/// The per-peer media owner (spec §3.1, §4.5).
|
||
///
|
||
/// # Ownership / sharing
|
||
///
|
||
/// An `RtcSession` lives behind an `Arc<Mutex<RtcSession>>` in the
|
||
/// binary's `DashMap<ChannelId, RtcSession>` (Task 5). The mutex is
|
||
/// short-held: each tokio poll iteration locks, runs `run_poll_once`,
|
||
/// unlocks. We do NOT hold the lock across `tokio::time::sleep` — that
|
||
/// would defeat the `DashMap`'s sharded concurrency and pre-pave the
|
||
/// wrong pattern for step 4's dedicated thread.
|
||
///
|
||
/// # Why `Arc<Mutex<...>>` (not `Arc<RwLock<...>>`)
|
||
///
|
||
/// Every access of an `RtcSession` mutates it (str0m's `&mut self`
|
||
/// contract on `handle_input` + `poll_output`). `RwLock`'s read-mode
|
||
/// would be useless because str0m takes `&mut Rtc`. `Mutex` it is.
|
||
pub struct RtcSession {
|
||
pub channel: Channel,
|
||
pub(crate) rtc: Rtc,
|
||
pub(crate) decoder: OpusDecoder,
|
||
pub(crate) encoder: OpusEncoder,
|
||
pub(crate) pipe: EchoAudioPipe,
|
||
/// Local UDP socket str0m sends `Transmit` packets out on and
|
||
/// 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,
|
||
/// 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`.
|
||
/// Slice 1 has exactly one m-line; multi-m-line arrives with video.
|
||
pub(crate) audio_mid: Option<Mid>,
|
||
/// Last deadline from `Rtc::poll_output` — the next time the loop
|
||
/// should wake the rtc with `Input::Timeout`.
|
||
pub(crate) next_timeout: Option<Instant>,
|
||
/// Last Instant we received an RTP packet from the peer. Used for
|
||
/// the 60 s idle timeout (spec §4.5).
|
||
pub(crate) last_rx: Instant,
|
||
/// Last Instant we wrote an outbound Opus frame. Used to pace the
|
||
/// 20 ms encode tick for the echo path (slice-1 read of spec §3.2).
|
||
pub(crate) last_outbound_at: Instant,
|
||
/// Outbound RTP media-time clock. For Opus audio on the wire str0m's
|
||
/// negotiated clock is 48 kHz; a 20 ms frame advances the RTP
|
||
/// timestamp by 48 000 × 0.020 = 960 ticks. We increment the
|
||
/// `MediaTime` by `MediaTime::from(Duration)` per frame, since
|
||
/// `MediaTime` has no `add(Duration)` method on str0m 0.21 — the
|
||
/// `From<Duration>` impl interprets the duration against the
|
||
/// underlying clock frequency.
|
||
pub(crate) next_media_time: str0m::media::MediaTime,
|
||
}
|
||
|
||
impl RtcSession {
|
||
/// Construct a new session — used by both the binary's `AppState`
|
||
/// (production) and the tests. Single constructor — no `for_test` /
|
||
/// `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()
|
||
}
|
||
|
||
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: 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,
|
||
})
|
||
}
|
||
|
||
pub fn channel_id(&self) -> ChannelId {
|
||
self.channel.id
|
||
}
|
||
|
||
pub fn channel_state(&self) -> ChannelState {
|
||
self.channel.state
|
||
}
|
||
|
||
pub fn is_closed(&self) -> bool {
|
||
matches!(self.channel.state, ChannelState::Closed)
|
||
}
|
||
|
||
/// Accept a browser SDP offer; return the SDP answer (spec §4.1).
|
||
///
|
||
/// str0m 0.21's `sdp_api().accept_offer()` does the heavy lifting:
|
||
/// parses the offer, picks compatible codecs (Opus, the only one we
|
||
/// register by default), generates the DTLS fingerprint from its
|
||
/// self-signed cert, and produces ICE ufrag/pwd. We add our local
|
||
/// host candidate (the UDP socket we just bound) *before* calling
|
||
/// `accept_offer` so the answer carries it.
|
||
pub fn accept_offer(&mut self, offer_sdp: &str) -> Result<String, RtcSessionError> {
|
||
assert!(self.audio_mid.is_none(), "accept_offer called twice");
|
||
|
||
// Register our local UDP socket as a host candidate. str0m includes
|
||
// this candidate's address + the ICE creds it generates in the SDP
|
||
// answer. `add_local_candidate` returns `Option<&Candidate>` —
|
||
// `None` means str0m rejected it (log + continue; not fatal).
|
||
let candidate = str0m::Candidate::host(self.local_addr, "udp")
|
||
.expect("host candidate from bound UDP socket");
|
||
// ^-- expect is acceptable here: this is construction (cold path),
|
||
// not the hot path. A bound UDP socket always yields a valid
|
||
// host candidate; only an absurd Protocol parse fails.
|
||
if self.rtc.add_local_candidate(candidate).is_none() {
|
||
tracing::warn!(channel_id = %self.channel.id, "str0m rejected local candidate");
|
||
}
|
||
|
||
// str0m's SDP API parses + accepts the offer natively. There is NO
|
||
// `from_str_unchecked` — `SdpOffer::from_sdp_string` returns
|
||
// `Result` and is the canonical entry point. `accept_offer` takes
|
||
// the owned `SdpOffer` and returns the `SdpAnswer`.
|
||
let parsed_offer = SdpOffer::from_sdp_string(offer_sdp)
|
||
.map_err(|e| RtcSessionError::SdpOffer(format!("parse: {e}")))?;
|
||
let answer = self
|
||
.rtc
|
||
.sdp_api()
|
||
.accept_offer(parsed_offer)
|
||
.map_err(|e| RtcSessionError::SdpOffer(format!("accept: {e}")))?;
|
||
|
||
// The first audio mid we accepted. Used to get the Writer for
|
||
// outbound Opus frames in `run_poll_once`. A single audio m-line
|
||
// is slice 1's whole world; multi-m-line arrives with video.
|
||
//
|
||
// `SdpAnswer` has no `mid()` accessor in str0m 0.21 (the brief's
|
||
// sketch assumed it). It derefs to `Sdp`, whose `media_lines`
|
||
// is `Vec<MediaLine>`; each `MediaLine::mid()` returns the mid.
|
||
// Slice 1's answer has exactly one m-line; we read its mid.
|
||
self.audio_mid = answer.media_lines.first().map(|m| m.mid());
|
||
|
||
self.channel.state = ChannelState::Connecting;
|
||
Ok(answer.to_sdp_string())
|
||
}
|
||
|
||
/// Drive one iteration of the sans-IO poll loop (spec §3.2, §3.4).
|
||
///
|
||
/// Returns the `Duration` until the next `Input::Timeout` should be
|
||
/// fed back to str0m, or `None` if the peer is closed. The caller
|
||
/// (Task 5's tokio task) sleeps this duration then calls again.
|
||
///
|
||
/// DEV-DEVIATION: tokio polling accepted for slice 1; step 4
|
||
/// replaces with dedicated timing thread per ARCHITECTURE.md.
|
||
pub fn run_poll_once(&mut self, now: Instant) -> Option<Duration> {
|
||
if self.is_closed() {
|
||
return None;
|
||
}
|
||
crate::loop_driver::drive(self, now)
|
||
}
|
||
}
|
||
|
||
#[cfg(test)]
|
||
mod tests {
|
||
use super::*;
|
||
|
||
/// A captured Chrome SDP offer for an audio-only Opus m-line. Real
|
||
/// browser-style offer with host ICE candidates — the simplest valid
|
||
/// offer str0m 0.21 will accept and produce an answer for. Slice 1 has
|
||
/// no video m-line; multi-m-line arrives with the escalation rung.
|
||
///
|
||
/// NOTE: the brief's fixture omitted the `a=group:BUNDLE 0` line
|
||
/// (str0m 0.21 rejects SDP without a session-level group attribute as
|
||
/// inconsistent — `sdp::Sdp::assert_consistency`). Real Chrome offers
|
||
/// always include BUNDLE; restored here so the offer str0m receives
|
||
/// matches what a browser actually sends.
|
||
const BROWSER_SDP_OFFER: &str = "\
|
||
v=0\r
|
||
o=- 4593482934 2 IN IP4 127.0.0.1\r
|
||
s=-\r
|
||
t=0 0\r
|
||
a=group:BUNDLE 0\r
|
||
m=audio 9 UDP/TLS/RTP/SAVPF 111\r
|
||
c=IN IP4 0.0.0.0\r
|
||
a=rtcp:9 IN IP4 0.0.0.0\r
|
||
a=ice-ufrag:abcd\r
|
||
a=ice-pwd:abcdefghijklmnopqrstuvwxyz0123456789\r
|
||
a=fingerprint:sha-256 AB:CD:EF:00:11:22:33:44:55:66:77:88:99:AA:BB:CC:DD:EE:FF:00:11:22:33:44:55:66:77:88:99:AA:BB:CC:DD\r
|
||
a=setup:actpass\r
|
||
a=mid:0\r
|
||
a=sendrecv\r
|
||
a=rtpmap:111 opus/48000/2\r
|
||
a=fmtp:111 minptime=10;useinbandfec=1\r
|
||
a=candidate:1 1 UDP 2113667327 192.168.1.42 50000 typ host\r
|
||
";
|
||
|
||
#[test]
|
||
fn accept_offer_returns_sdp_answer_with_opus() {
|
||
let mut session = RtcSession::new().expect("session");
|
||
let answer = session.accept_offer(BROWSER_SDP_OFFER).expect("SDP answer");
|
||
// Answer contains an audio m-line, an Opus payload, a fingerprint,
|
||
// and ICE credentials (str0m fills these natively in 0.21).
|
||
assert!(answer.contains("m=audio"), "answer has an audio m-line");
|
||
assert!(answer.contains("opus/48000"), "answer advertises Opus");
|
||
assert!(
|
||
answer.contains("a=fingerprint:sha-256 "),
|
||
"DTLS fingerprint"
|
||
);
|
||
assert!(answer.contains("a=ice-ufrag:"), "ICE ufrag present");
|
||
assert!(answer.contains("a=ice-pwd:"), "ICE pwd present");
|
||
}
|
||
|
||
#[test]
|
||
fn channel_id_matches_session_id() {
|
||
let session = RtcSession::new().expect("session");
|
||
let id = session.channel_id();
|
||
// The ChannelId IS the session id surfaced in the REST API (spec §4.5).
|
||
assert_eq!(format!("{}", id).len(), 36);
|
||
}
|
||
|
||
#[test]
|
||
fn accept_offer_transitions_channel_to_connecting() {
|
||
// The spec §5.4 state machine: New → Connecting on offer receive.
|
||
// This test pins the transition callers depend on; the impl sets
|
||
// it at the end of `accept_offer`.
|
||
let mut session = RtcSession::new().expect("session");
|
||
assert_eq!(session.channel_state(), ChannelState::New);
|
||
let _ = session.accept_offer(BROWSER_SDP_OFFER).expect("answer");
|
||
assert_eq!(session.channel_state(), ChannelState::Connecting);
|
||
}
|
||
}
|