diff --git a/.github/workflows/ci.yml b/.github/workflows/ci.yml index 774c338..0a817b1 100644 --- a/.github/workflows/ci.yml +++ b/.github/workflows/ci.yml @@ -41,17 +41,18 @@ jobs: steps: - uses: actions/checkout@v4 - # Seam gate (slice-4 §7 #3 / PORT_PLAN phasing): + # Seam gate (slice-4 §7 #3; re-pinned slice-5): # - # `loop_driver.rs` and `rtc_session.rs` are the hot-path media seam - # preserved from slice-3. They MUST remain byte-identical across - # slice-4 so the reflex wrapper can be added without touching the - # core poll loop. If a legitimate change is needed, update the pinned - # blob hashes below in the same PR — loud and reviewable by design. + # `loop_driver.rs` stays byte-identical to slice-3 — the hot-path + # poll loop is untouched. `rtc_session.rs` was re-pinned in slice-5 + # for MediaAddressConfig (2026-07-04 scalability review, B1: + # bind/advertised address split + port range). If a legitimate + # change is needed, update the pinned blob hashes below in the same + # PR — loud and reviewable by design. - name: Seam gate — loop_driver + rtc_session byte-identical to slice-3 run: | EXPECTED_LOOP_DRIVER='744bf314edf7f4925c8bb3bd0f5176dbc88f8113' - EXPECTED_RTC_SESSION='a4c9f2ae64e56c08e1990956391514929535b526' + EXPECTED_RTC_SESSION='76bbcf8c85be0041abda1bcc3390b299fcef3d6f' GOT_LOOP_DRIVER=$(git hash-object crates/rutster-media/src/loop_driver.rs) GOT_RTC_SESSION=$(git hash-object crates/rutster-media/src/rtc_session.rs) if [ "$GOT_LOOP_DRIVER" != "$EXPECTED_LOOP_DRIVER" ]; then diff --git a/crates/rutster-media/src/lib.rs b/crates/rutster-media/src/lib.rs index 4798783..51373b2 100644 --- a/crates/rutster-media/src/lib.rs +++ b/crates/rutster-media/src/lib.rs @@ -44,7 +44,7 @@ pub use reflex::{ AdvisoryEvent, LocalVadReflex, Reflex, ReflexMetrics, ReflexMetricsSnapshot, VAD_DEBOUNCE_FRAMES, VAD_RMS_THRESHOLD, }; -pub use rtc_session::{RtcSession, RtcSessionError}; +pub use rtc_session::{MediaAddressConfig, RtcSession, RtcSessionError}; use thiserror::Error; diff --git a/crates/rutster-media/src/rtc_session.rs b/crates/rutster-media/src/rtc_session.rs index a4c9f2a..76bbcf8 100644 --- a/crates/rutster-media/src/rtc_session.rs +++ b/crates/rutster-media/src/rtc_session.rs @@ -17,7 +17,7 @@ //! (which owns any carrier SIP/SDP, outside the trust boundary). rutster's //! WebRTC-ICE-coupled SDP lives entirely in str0m. -use std::net::SocketAddr; +use std::net::{IpAddr, SocketAddr, UdpSocket}; use std::time::{Duration, Instant}; use rutster_call_model::{Channel, ChannelId, ChannelState}; @@ -57,6 +57,55 @@ pub enum RtcSessionError { use str0m::change::SdpOffer; use str0m::media::Mid; +/// Where the media socket binds vs. what address peers are told to reach. +/// +/// # Why bind and advertised are SEPARATE concepts (slice-5, review B1) +/// +/// Behind cloud NAT / a 1:1 elastic IP, the address a peer must send RTP +/// to is NOT the local bind address — and RTP cannot ride the HTTP load +/// balancer, so every instance must advertise its own reachable address. +/// str0m also rejects the unspecified address (`0.0.0.0`) as a candidate, +/// which forces the split at the type level: you may bind wildcard, but +/// you must then say what to advertise. +/// +/// `port_range` bounds the UDP ports (inclusive) so a fleet's security +/// groups can be written; `None` keeps the OS-ephemeral behavior. +#[derive(Debug, Clone)] +pub struct MediaAddressConfig { + pub bind_ip: IpAddr, + pub advertised_ip: Option, + pub port_range: Option<(u16, u16)>, +} + +impl Default for MediaAddressConfig { + /// Loopback + ephemeral — byte-for-byte the pre-slice-5 behavior, so + /// every existing test and the dev loop run unchanged. + fn default() -> Self { + Self { + bind_ip: IpAddr::from([127, 0, 0, 1]), + advertised_ip: None, + port_range: None, + } + } +} + +/// Bind the first free UDP port in `[lo, hi]` (inclusive). +/// +/// Sequential scan — O(range) worst case, cold path only (session +/// construction). The OS is the allocator: a failed bind means "taken," +/// so no shared allocator state is needed across sessions or threads. +fn bind_in_range(ip: IpAddr, lo: u16, hi: u16) -> std::io::Result { + let mut last_err = None; + for port in lo..=hi { + match UdpSocket::bind(SocketAddr::new(ip, port)) { + Ok(s) => return Ok(s), + Err(e) => last_err = Some(e), + } + } + Err(last_err + .unwrap_or_else(|| std::io::Error::new(std::io::ErrorKind::AddrInUse, "empty port range"))) +} + /// The per-peer media owner (spec §3.1, §4.5). /// /// # Ownership / sharing @@ -86,6 +135,11 @@ pub struct RtcSession { pub(crate) socket: std::net::UdpSocket, /// Local socket address — cached because `local_addr()` is a syscall. pub(crate) local_addr: SocketAddr, + /// The address written into our ICE host candidate — advertised IP + /// (if configured) + the actually-bound port. `local_addr` stays the + /// real socket address (loop_driver's `Receive::new` needs it); + /// `advertised_addr` is what the SDP answer tells the peer. + pub(crate) advertised_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, @@ -114,7 +168,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() + Self::new_with_config(&MediaAddressConfig::default()) + } + + /// Construct with explicit addressing (slice-5, review B1). The + /// binary threads `MediaThreadOpts.media_cfg` here; `new()` keeps the + /// loopback default for tests and the dev loop. + pub fn new_with_config(cfg: &MediaAddressConfig) -> Result { + // Fail-fast: a wildcard bind with nothing to advertise would die + // later inside accept_offer (str0m rejects 0.0.0.0 candidates). + // Cold path → a real error, not an expect(). + if cfg.bind_ip.is_unspecified() && cfg.advertised_ip.is_none() { + return Err(RtcSessionError::Socket(std::io::Error::new( + std::io::ErrorKind::InvalidInput, + "RUTSTER_MEDIA_BIND_IP is unspecified (0.0.0.0/::) — set RUTSTER_MEDIA_ADVERTISED_IP", + ))); + } + let socket = match cfg.port_range { + Some((lo, hi)) => bind_in_range(cfg.bind_ip, lo, hi)?, + None => UdpSocket::bind(SocketAddr::new(cfg.bind_ip, 0))?, + }; + socket.set_nonblocking(true)?; + let local_addr = socket.local_addr()?; + let advertised_addr = + SocketAddr::new(cfg.advertised_ip.unwrap_or(cfg.bind_ip), local_addr.port()); + + 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, + advertised_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, @@ -163,41 +258,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 { @@ -233,11 +293,10 @@ impl RtcSession { // 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. + let candidate = str0m::Candidate::host(self.advertised_addr, "udp") + .expect("host candidate from validated advertised address"); + // ^-- expect stays acceptable: new_with_config rejected the only + // input (unspecified IP) that str0m refuses, so this is const-ish. if self.rtc.add_local_candidate(candidate).is_none() { tracing::warn!(channel_id = %self.channel.id, "str0m rejected local candidate"); } @@ -352,4 +411,60 @@ a=candidate:1 1 UDP 2113667327 192.168.1.42 50000 typ host\r let _ = session.accept_offer(BROWSER_SDP_OFFER).expect("answer"); assert_eq!(session.channel_state(), ChannelState::Connecting); } + + #[test] + fn default_config_binds_loopback_ephemeral() { + // Byte-for-byte behavioral compatibility: new() == new_with_config(default). + let s = RtcSession::new_with_config(&MediaAddressConfig::default()).expect("session"); + assert!(s.local_addr.ip().is_loopback()); + } + + #[test] + fn port_range_is_respected_and_exhaustion_errors() { + let cfg = MediaAddressConfig { + bind_ip: "127.0.0.1".parse().unwrap(), + advertised_ip: None, + port_range: Some((41500, 41501)), // room for exactly two sessions + }; + let a = RtcSession::new_with_config(&cfg).expect("first"); + let b = RtcSession::new_with_config(&cfg).expect("second"); + assert!((41500..=41501).contains(&a.local_addr.port())); + assert!((41500..=41501).contains(&b.local_addr.port())); + // Range full → cold-path Socket error, not a panic. + assert!(matches!( + RtcSession::new_with_config(&cfg), + Err(RtcSessionError::Socket(_)) + )); + } + + #[test] + fn sdp_answer_advertises_the_advertised_ip_not_the_bind_ip() { + let cfg = MediaAddressConfig { + bind_ip: "127.0.0.1".parse().unwrap(), + advertised_ip: Some("203.0.113.7".parse().unwrap()), // TEST-NET-3 + port_range: None, + }; + let mut s = RtcSession::new_with_config(&cfg).expect("session"); + let answer = s.accept_offer(BROWSER_SDP_OFFER).expect("answer"); + assert!( + answer.contains("203.0.113.7"), + "candidate must carry advertised IP" + ); + assert!( + !answer.contains("127.0.0.1 "), + "bind IP must not leak into the candidate" + ); + } + + #[test] + fn unspecified_bind_without_advertised_ip_is_rejected() { + // str0m's Candidate::host rejects 0.0.0.0 — fail at construction + // with a config error, not at offer-accept with an expect(). + let cfg = MediaAddressConfig { + bind_ip: "0.0.0.0".parse().unwrap(), + advertised_ip: None, + port_range: None, + }; + assert!(RtcSession::new_with_config(&cfg).is_err()); + } } diff --git a/crates/rutster/src/config.rs b/crates/rutster/src/config.rs index ed84062..844b5db 100644 --- a/crates/rutster/src/config.rs +++ b/crates/rutster/src/config.rs @@ -7,7 +7,9 @@ //! //! Closes 2026-07-04 scalability review "HTTP bind hardcoded" (minor). -use std::net::SocketAddr; +use std::net::{IpAddr, SocketAddr}; + +use rutster_media::MediaAddressConfig; /// Resolve the HTTP bind address from `RUTSTER_HTTP_BIND`. /// @@ -23,6 +25,49 @@ pub fn http_bind(raw: Option) -> Result { } } +/// Parse `RUTSTER_MEDIA_PORT_RANGE` — `"lo-hi"`, inclusive, lo ≤ hi. +pub fn parse_port_range(raw: &str) -> Result<(u16, u16), String> { + let (lo, hi) = raw + .split_once('-') + .ok_or_else(|| format!("RUTSTER_MEDIA_PORT_RANGE {raw:?}: expected \"lo-hi\""))?; + let lo: u16 = lo + .trim() + .parse() + .map_err(|e| format!("RUTSTER_MEDIA_PORT_RANGE lo: {e}"))?; + let hi: u16 = hi + .trim() + .parse() + .map_err(|e| format!("RUTSTER_MEDIA_PORT_RANGE hi: {e}"))?; + if lo > hi { + return Err(format!("RUTSTER_MEDIA_PORT_RANGE {raw:?}: lo > hi")); + } + Ok((lo, hi)) +} + +/// Assemble `MediaAddressConfig` from the three `RUTSTER_MEDIA_*` vars. +pub fn media_address_config( + bind_ip: Option, + advertised_ip: Option, + port_range: Option, +) -> Result { + let mut cfg = MediaAddressConfig::default(); + if let Some(s) = bind_ip { + cfg.bind_ip = s + .parse::() + .map_err(|e| format!("RUTSTER_MEDIA_BIND_IP: {e}"))?; + } + if let Some(s) = advertised_ip { + cfg.advertised_ip = Some( + s.parse::() + .map_err(|e| format!("RUTSTER_MEDIA_ADVERTISED_IP: {e}"))?, + ); + } + if let Some(s) = port_range { + cfg.port_range = Some(parse_port_range(&s)?); + } + Ok(cfg) +} + #[cfg(test)] mod tests { use super::*; @@ -48,4 +93,37 @@ mod tests { let err = http_bind(Some("not-an-addr".into())).unwrap_err(); assert!(err.contains("RUTSTER_HTTP_BIND")); } + + #[test] + fn port_range_parses_inclusive_pair() { + assert_eq!(parse_port_range("10000-20000").unwrap(), (10000, 20000)); + } + + #[test] + fn port_range_rejects_inverted_and_garbage() { + assert!(parse_port_range("20000-10000").is_err()); + assert!(parse_port_range("10000").is_err()); + assert!(parse_port_range("a-b").is_err()); + } + + #[test] + fn media_address_config_defaults_to_loopback_ephemeral() { + let cfg = media_address_config(None, None, None).unwrap(); + assert!(cfg.bind_ip.is_loopback()); + assert!(cfg.advertised_ip.is_none()); + assert!(cfg.port_range.is_none()); + } + + #[test] + fn media_address_config_parses_all_three() { + let cfg = media_address_config( + Some("0.0.0.0".into()), + Some("203.0.113.7".into()), + Some("10000-10100".into()), + ) + .unwrap(); + assert!(cfg.bind_ip.is_unspecified()); + assert_eq!(cfg.advertised_ip.unwrap().to_string(), "203.0.113.7"); + assert_eq!(cfg.port_range, Some((10000, 10100))); + } } diff --git a/crates/rutster/src/main.rs b/crates/rutster/src/main.rs index bd3c09f..8a4caa1 100644 --- a/crates/rutster/src/main.rs +++ b/crates/rutster/src/main.rs @@ -37,8 +37,16 @@ async fn main() { // from the spawned MediaThread below. let (placeholder_tx, _placeholder_rx) = mpsc::channel(1); let app_state = AppState::new(placeholder_tx, default_tap_url); + + let media_cfg = rutster::config::media_address_config( + std::env::var("RUTSTER_MEDIA_BIND_IP").ok(), + std::env::var("RUTSTER_MEDIA_ADVERTISED_IP").ok(), + std::env::var("RUTSTER_MEDIA_PORT_RANGE").ok(), + ) + .expect("RUTSTER_MEDIA_* env config invalid"); + let opts = rutster::media_thread::MediaThreadOpts { media_cfg }; let (app_state, media_thread) = app_state - .spawn_media_thread(tokio::runtime::Handle::current()) + .spawn_media_thread_with(opts, tokio::runtime::Handle::current()) .expect("media thread spawn"); // RUTSTER_HTTP_BIND: per-instance bind is the first thing any LB / diff --git a/crates/rutster/src/media_thread.rs b/crates/rutster/src/media_thread.rs index 26c7406..a50adcd 100644 --- a/crates/rutster/src/media_thread.rs +++ b/crates/rutster/src/media_thread.rs @@ -38,6 +38,26 @@ const META_TICK: Duration = Duration::from_millis(10); /// Capacity for the command channel from axum to the media thread. const CMD_CHANNEL_CAPACITY: usize = 64; +/// Knobs the binary resolves from env and hands to the media thread. +/// Manual Default (not derive) because later slices add non-derivable +/// fields (Task 7's `Arc`). Defaults reproduce the +/// pre-slice-5 single-node dev behavior exactly. +pub struct MediaThreadOpts { + pub media_cfg: rutster_media::MediaAddressConfig, +} + +// clippy sees a single-field struct today and suggests `#[derive(Default)]`; +// we keep the manual impl on purpose (see doc comment above) so Task 7's +// `Arc` field doesn't force a churn-y derive → manual flip. +#[allow(clippy::derivable_impls)] +impl Default for MediaThreadOpts { + fn default() -> Self { + Self { + media_cfg: rutster_media::MediaAddressConfig::default(), + } + } +} + /// Commands axum sends to the media thread (cold-path only — NEVER on /// the 20ms tick). The thread owns RtcSessions exclusively; this is the /// ONLY entry point for axum-side mutation. @@ -95,6 +115,7 @@ impl MediaThread { /// crash" policy. pub fn spawn( default_tap_url: url::Url, + opts: MediaThreadOpts, tokio_handle: tokio::runtime::Handle, ) -> Result { let (cmd_tx, cmd_rx) = mpsc::channel(CMD_CHANNEL_CAPACITY); @@ -106,7 +127,13 @@ impl MediaThread { let join = std::thread::Builder::new() .name("rutster-media".into()) .spawn(move || { - run_media_thread(cmd_rx, default_tap_url, tokio_handle, cmd_tx_for_thread); + run_media_thread( + cmd_rx, + default_tap_url, + opts, + tokio_handle, + cmd_tx_for_thread, + ); })?; Ok(Self { cmd_tx, @@ -158,6 +185,7 @@ struct ThreadSession { fn run_media_thread( mut cmd_rx: mpsc::Receiver, default_tap_url: url::Url, + opts: MediaThreadOpts, tokio_handle: tokio::runtime::Handle, media_cmd_tx: mpsc::Sender, ) { @@ -168,29 +196,31 @@ fn run_media_thread( // === Step 1: drain ALL pending commands (cold path) BEFORE ticking. === while let Ok(cmd) = cmd_rx.try_recv() { match cmd { - MediaCmd::Register { tap_url, reply } => match RtcSession::new() { - Ok(session) => { - let id = session.channel_id(); - let app_state = crate::session_map::AppState::new( - media_cmd_tx.clone(), - default_tap_url.clone(), - ); - sessions.insert( - id, - ThreadSession { - rtc: session, - tap_url, - app_state, - tap_conn: None, - }, - ); - let _ = reply.send(Ok(id)); - debug!(channel_id = %id, "session registered"); + MediaCmd::Register { tap_url, reply } => { + match RtcSession::new_with_config(&opts.media_cfg) { + Ok(session) => { + let id = session.channel_id(); + let app_state = crate::session_map::AppState::new( + media_cmd_tx.clone(), + default_tap_url.clone(), + ); + sessions.insert( + id, + ThreadSession { + rtc: session, + tap_url, + app_state, + tap_conn: None, + }, + ); + let _ = reply.send(Ok(id)); + debug!(channel_id = %id, "session registered"); + } + Err(e) => { + let _ = reply.send(Err(format!("RtcSession::new: {e}"))); + } } - Err(e) => { - let _ = reply.send(Err(format!("RtcSession::new: {e}"))); - } - }, + } MediaCmd::AcceptOffer { id, sdp, reply } => { let result = match sessions.get_mut(&id) { Some(s) => s.rtc.accept_offer(&sdp).map_err(|e| format!("{e}")), @@ -313,7 +343,8 @@ mod tests { let rt = tokio::runtime::Runtime::new().unwrap(); let handle = rt.handle().clone(); let url = url::Url::parse("ws://127.0.0.1:8081/echo").unwrap(); - let thread = MediaThread::spawn(url, handle).expect("media thread spawn in test"); + let thread = MediaThread::spawn(url, MediaThreadOpts::default(), handle) + .expect("media thread spawn in test"); let (reply, rx) = oneshot::channel(); thread .cmd_tx() diff --git a/crates/rutster/src/session_map.rs b/crates/rutster/src/session_map.rs index 05eb55d..d9f3d6e 100644 --- a/crates/rutster/src/session_map.rs +++ b/crates/rutster/src/session_map.rs @@ -85,11 +85,24 @@ impl AppState { /// Returns the state with the real command-channel sender wired in, /// plus the `MediaThread` handle. Routes on the returned `AppState` /// will send `MediaCmd` to the running media thread. + /// Spawn with defaults — the test-facing convenience; production goes + /// through `spawn_media_thread_with` so env config reaches the thread. pub fn spawn_media_thread( - mut self, + self, tokio_handle: tokio::runtime::Handle, ) -> Result<(Self, MediaThread), std::io::Error> { - let thread = MediaThread::spawn(self.default_tap_url.clone(), tokio_handle)?; + self.spawn_media_thread_with( + crate::media_thread::MediaThreadOpts::default(), + tokio_handle, + ) + } + + pub fn spawn_media_thread_with( + mut self, + opts: crate::media_thread::MediaThreadOpts, + tokio_handle: tokio::runtime::Handle, + ) -> Result<(Self, MediaThread), std::io::Error> { + let thread = MediaThread::spawn(self.default_tap_url.clone(), opts, tokio_handle)?; self.cmd_tx = thread.cmd_tx(); Ok((self, thread)) } diff --git a/crates/rutster/tests/api_integration.rs b/crates/rutster/tests/api_integration.rs index 70081b8..dba04dd 100644 --- a/crates/rutster/tests/api_integration.rs +++ b/crates/rutster/tests/api_integration.rs @@ -7,7 +7,7 @@ use axum::body::Body; use axum::http::{Request, StatusCode}; -use rutster::media_thread::MediaThread; +use rutster::media_thread::{MediaThread, MediaThreadOpts}; use rutster::session_map::AppState; use tower::ServiceExt; // enables `oneshot` on the Router for sync tests @@ -23,8 +23,12 @@ fn app_state_with_thread() -> (AppState, MediaThread) { let default = default_tap_url(); let (placeholder_tx, _placeholder_rx) = tokio::sync::mpsc::channel(1); let mut state = AppState::new(placeholder_tx, default.clone()); - let media_thread = - MediaThread::spawn(default, tokio::runtime::Handle::current()).expect("media thread spawn"); + let media_thread = MediaThread::spawn( + default, + MediaThreadOpts::default(), + tokio::runtime::Handle::current(), + ) + .expect("media thread spawn"); state.cmd_tx = media_thread.cmd_tx(); (state, media_thread) }