diff --git a/crates/rutster-media/Cargo.toml b/crates/rutster-media/Cargo.toml index 3aef900..fd096bd 100644 --- a/crates/rutster-media/Cargo.toml +++ b/crates/rutster-media/Cargo.toml @@ -10,6 +10,7 @@ description = "Media core: str0m WebRTC + Opus⇄PCM boundary (slice 1)." [dependencies] rutster-call-model = { path = "../rutster-call-model" } opus = { workspace = true } +str0m = { workspace = true } thiserror = { workspace = true } tracing = { workspace = true } diff --git a/crates/rutster-media/src/lib.rs b/crates/rutster-media/src/lib.rs index 90867fa..4fc93f4 100644 --- a/crates/rutster-media/src/lib.rs +++ b/crates/rutster-media/src/lib.rs @@ -30,11 +30,14 @@ //! - [`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 pcm; +pub mod rtc_session; pub use opus_codec::{OpusDecoder, OpusEncoder}; pub use pcm::{AudioSink, AudioSource, EchoAudioPipe, PcmFrame, SAMPLES_PER_FRAME}; +pub use rtc_session::{RtcSession, RtcSessionError}; use thiserror::Error; diff --git a/crates/rutster-media/src/loop_driver.rs b/crates/rutster-media/src/loop_driver.rs new file mode 100644 index 0000000..48ee2f5 --- /dev/null +++ b/crates/rutster-media/src/loop_driver.rs @@ -0,0 +1,258 @@ +//! # str0m poll loop (spec §3.2, §3.4) +//! +//! The heart of the media core. Drives the `str0m::Rtc` instance forward +//! on each call: drains `poll_output()` until `Output::Timeout`, handling +//! each `Output::Transmit` (send on our UDP socket) and `Output::Event` +//! (inbound `MediaData` → Opus decode → sink; inbound RTP count for the +//! idle timeout). When the drain returns `Timeout`, the caller sleeps +//! that duration and calls back with `Input::Timeout`. +//! +//! # Why this lives in a separate module +//! +//! `run_poll_once` takes `&mut RtcSession` — a single function with +//! the full poll logic would make `RtcSession::run_poll_once` 100+ lines +//! of non-trivial control flow. Splitting the loop into a module makes +//! the sans-IO pattern obvious: the loop driver takes a `&mut RtcSession`, +//! reads str0m outputs, and writes str0m inputs. Nothing else. +//! +//! # DEV-DEVIATION +//! +//! Slice 1 runs the poll on a tokio task. ARCHITECTURE.md mandates a +//! dedicated timing thread; we defer that to step 4 (barge-in) because +//! slice 1 has no reflex to time against. The poll function's shape +//! (single `&mut self`, no I/O inside) makes the step-4 swap localized. + +use std::io::ErrorKind; +use std::time::{Duration, Instant}; + +use str0m::net::Protocol; +use str0m::{Input, Output}; + +use crate::pcm::{AudioSink as _, AudioSource as _}; +use crate::rtc_session::{RtcSession, IDLE_TIMEOUT}; + +/// 20 ms tick for outbound encoding (matches the PCM frame size, spec §3.9: +/// 480 samples @ 24 kHz = 20 ms). On each tick, we pull one frame from the +/// source pipe and write the encoded Opus via str0m's `Writer::write`. +const OUTBOUND_TICK: Duration = Duration::from_millis(20); + +/// One iteration of the str0m poll loop. +/// +/// 1. Read any pending UDP packets (non-blocking) and feed each to str0m +/// as `Input::Receive`. A WouldBlock means no packets this cycle — fine. +/// 2. Drain `poll_output()` until `Timeout`: +/// - `Transmit` → send on our UDP socket. +/// - `Event::MediaData` → decode Opus → push to the echo pipe (sink). +/// - `Event::IceConnectionStateChange` → state transition + tracing. +/// - We don't break out of the drain on any of these: str0m's contract +/// is mutate → drain to `Timeout` → mutate (see str0m 0.21 lib docs). +/// 3. **Outbound encode tick:** if ≥20 ms of wallclock passed since the +/// last outbound frame, pull one `PcmFrame` from the source, encode to +/// Opus, and write via `Rtc::writer(mid)->Writer::write`. Then re-drain +/// `poll_output` (the Writer write is a mutation → must drain per str0m). +/// 4. Check the idle timeout: if `now - last_rx > IDLE_TIMEOUT`, transition +/// to `Closed`. +/// 5. Return the `Duration` to the next `Timeout`. +pub fn drive(session: &mut RtcSession, now: Instant) -> Option { + // === Step 1: drain our UDP socket non-blocking, feed str0m. === + let mut buf = [0u8; 2000]; + loop { + match session.socket.recv_from(&mut buf) { + Ok((n, source)) => { + // `Receive::new` parses the raw datagram into one of + // STUN/DTLS/RTP/RTCP (str0m's demultiplexer). It returns + // `Err(NetError)` for things str0m can't classify; we drop + // those, don't crash (hot-path policy, spec §3.8). + let recv = match str0m::net::Receive::new( + Protocol::Udp, + source, + session.local_addr, + &buf[..n], + ) { + Ok(r) => r, + Err(e) => { + tracing::warn!(error = ?e, "str0m datagram parse failed; dropping"); + continue; + } + }; + // str0m's `Input::Receive` carries the receive timestamp as + // the first tuple element (str0m 0.21 API — the brief's + // sketch had `Input::Receive(recv)` and `handle_input(now, ...)`, + // both adjusted to the actual surface: a single `Input` + // argument, with the Instant packed into the variant). + if session.rtc.handle_input(Input::Receive(now, recv)).is_err() { + // Hot-path policy: drop + observe, don't crash. + tracing::warn!("str0m rejected input packet; dropping"); + } + session.last_rx = now; + } + // WouldBlock (unix) / TimedOut (windows) — no packets this cycle. + Err(e) if matches!(e.kind(), ErrorKind::WouldBlock | ErrorKind::TimedOut) => break, + Err(e) => { + tracing::warn!(error = ?e, "UDP recv_from error; continuing"); + break; + } + } + } + + // === Step 2: drain poll_output, interleaving outbound writes. === + // `next_timeout` is set in either the `Timeout` or `Err` exit arms of + // the drain loop below before any read on the path forward — no + // initial value needed. Both exit arms assign before breaking, so the + // borrow checker is satisfied; clippy flagged the previous `= None` + // initializer as a write-then-overwrite. + let mut next_timeout: Option; + // Track whether we owe a Writer write this cycle; re-drain if so. + // str0m's "mutate → drain to Timeout" invariant: after Writer::write, + // poll_output must be drained to Timeout before any other mutation. + let mut needs_redrain = false; + loop { + match session.rtc.poll_output() { + Ok(Output::Timeout(t)) => { + next_timeout = Some(t); + if needs_redrain { + // We did an outbound write in the previous iteration; + // str0m needs to be drained again. Loop continues, + // but only handle Transmit/Event briefly before next Timeout. + needs_redrain = false; + continue; + } + break; // engine is fully drained + } + Ok(Output::Transmit(t)) => { + // `Transmit.contents` is a `DatagramSend` newtype that + // `Deref`s to `[u8]` — passing `&t.contents` deref-coerces + // to `&[u8]` for `send_to`. + if let Err(e) = session.socket.send_to(&t.contents, t.destination) { + if !matches!(e.kind(), ErrorKind::WouldBlock) { + tracing::warn!(error = ?e, "UDP send_to error; dropping"); + } + } + } + Ok(Output::Event(event)) => { + handle_event(session, event, now); + // Loop continues — mutations from inside the drain loop + // are fine (str0m docs, "single-mutation invariant"): + // events are observations, not external mutations. + } + Err(e) => { + tracing::warn!(error = ?e, "str0m poll_output error; continuing"); + next_timeout = Some(now + OUTBOUND_TICK); + break; + } + } + } + + // === Step 3: outbound encode tick (the echo path). === + // If wallclock has crossed a 20 ms boundary since the last outbound + // frame, pull a PcmFrame from the source, encode to Opus, and write + // via Writer::write. This IS the slice-1 echo: inbound decode → pipe + // → outbound encode. + if now.duration_since(session.last_outbound_at) >= OUTBOUND_TICK { + if let Some(mid) = session.audio_mid { + if let Some(frame) = session.pipe.next_pcm_frame() { + if let Some(opus_payload) = session.encoder.encode(&frame) { + // Writer::write signature (str0m 0.21, verified): + // write(pt: Pt, wallclock: Instant, rtp_time: MediaTime, + // data: impl Into>) -> Result<(), RtcError> + // - pt: payload type for Opus. `writer.payload_params()` + // returns `impl Iterator`; the + // first one's `.pt()` is our Opus PT (str0m negotiates + // this in the SDP answer). + // - wallclock: when the sample was produced — local `now`. + // - rtp_time: RTP timestamp in the 48 kHz audio clock for + // Opus. Increment by 960 per 20 ms (48000 * 0.020). + // `MediaTime` has no `add(Duration)` method — use + // `mt + MediaTime::from(duration)`. + // + // `rtc.writer(mid)` returns `Option>` — `None` + // if direction isn't sending (we'd be in a recvonly state). + if let Some(writer) = session.rtc.writer(mid) { + // Pull the Opus payload type out of the iterator + // BEFORE `writer.write(...)`, which consumes `writer` + // by value. Doing both inline trips E0505 because + // `payload_params()` borrows `writer` while + // `write(self, ...)` moves it — so separate the + // borrow from the move with a tight scope. + let pt = writer.payload_params().next().map(|p| p.pt()); + if let Some(pt) = pt { + let rtp_time = session.next_media_time; + if writer + .write(pt, now, rtp_time, opus_payload.as_slice()) + .is_ok() + { + // Advance media time for next 20 ms frame. + // `MediaTime + MediaTime::from(Duration)` — + // no `add()` method on `MediaTime` in str0m 0.21. + session.next_media_time += + str0m::media::MediaTime::from(Duration::from_millis(20)); + needs_redrain = true; + } + } + } + } + } + session.last_outbound_at = now; + } + } + + // If the outbound write happened, we owe str0m one more drain before + // returning — Writer::write is a mutation per str0m's invariant. + if needs_redrain { + loop { + match session.rtc.poll_output() { + Ok(Output::Timeout(t)) => { + next_timeout = Some(t); + break; + } + Ok(Output::Transmit(t)) => { + let _ = session.socket.send_to(&t.contents, t.destination); + } + Ok(Output::Event(e)) => handle_event(session, e, now), + Err(_) => break, + } + } + } + + // === Step 4: idle timeout (spec §4.5). === + if now.duration_since(session.last_rx) > IDLE_TIMEOUT { + tracing::info!( + channel_id = %session.channel.id, + "idle timeout (60 s no RX); closing session" + ); + session.channel.state = rutster_call_model::ChannelState::Closed; + return None; + } + + session.next_timeout = next_timeout; + next_timeout.map(|t| t.saturating_duration_since(now)) +} + +/// Dispatch a str0m `Event` to the audio pipe or to state bookkeeping. +fn handle_event(session: &mut RtcSession, event: str0m::Event, _now: Instant) { + use str0m::Event; + match event { + Event::MediaData(media) => { + // Inbound decoded audio frame from the peer (Frame API, spec §3.2). + // str0m has already done RTP depacketization; `MediaData.data` + // is the encoded Opus payload (type: `Arc<[u8]>` — passing + // `&media.data` deref-coerces to `&[u8]` for `OpusDecoder::decode`). + if let Some(pcm) = session.decoder.decode(&media.data) { + session.pipe.on_pcm_frame(pcm); + } + // Decode failed → drop + observe (per §3.8). Don't kill the peer. + } + Event::IceConnectionStateChange(state) => { + tracing::info!( + channel_id = %session.channel.id, + ?state, + "ICE state change" + ); + if state == str0m::IceConnectionState::Connected { + session.channel.state = rutster_call_model::ChannelState::Connected; + } + } + Event::EgressBitrateEstimate(_) => { /* BWE — irrelevant in slice 1 */ } + _ => { /* str0m emits several other event variants we don't need in slice 1. */ } + } +} diff --git a/crates/rutster-media/src/rtc_session.rs b/crates/rutster-media/src/rtc_session.rs new file mode 100644 index 0000000..6fbdc06 --- /dev/null +++ b/crates/rutster-media/src/rtc_session.rs @@ -0,0 +1,300 @@ +//! # `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>` in the +/// binary's `DashMap` (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>` (not `Arc>`) +/// +/// 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(crate) 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, + /// Last deadline from `Rtc::poll_output` — the next time the loop + /// should wake the rtc with `Input::Timeout`. + pub(crate) next_timeout: Option, + /// 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` 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::new_internal() + } + + 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: 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 { + 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`; 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 { + 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); + } +}