slice-5: MediaAddressConfig — advertised addr + port range (review B1)

Re-pins the rtc_session.rs seam hash (loud by design). loop_driver.rs
unchanged. Default config reproduces pre-slice-5 behavior exactly.

Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_012QndwfhjyTiZcUYp87dwW8
Signed-off-by: Aaron D. Lee <himself@adlee.work>

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;
 

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<IpAddr>,
+    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<UdpSocket> {
+    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<Mid>,
@@ -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, RtcSessionError> {
-        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<Self, RtcSessionError> {
+        // 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<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 {
@@ -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());
+    }
 }

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<String>) -> Result<SocketAddr, String> {
     }
 }
 
+/// 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<String>,
+    advertised_ip: Option<String>,
+    port_range: Option<String>,
+) -> Result<MediaAddressConfig, String> {
+    let mut cfg = MediaAddressConfig::default();
+    if let Some(s) = bind_ip {
+        cfg.bind_ip = s
+            .parse::<IpAddr>()
+            .map_err(|e| format!("RUTSTER_MEDIA_BIND_IP: {e}"))?;
+    }
+    if let Some(s) = advertised_ip {
+        cfg.advertised_ip = Some(
+            s.parse::<IpAddr>()
+                .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)));
+    }
 }

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 /

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<dyn EventSink>`). 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<dyn EventSink>` 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<Self, std::io::Error> {
         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<MediaCmd>,
     default_tap_url: url::Url,
+    opts: MediaThreadOpts,
     tokio_handle: tokio::runtime::Handle,
     media_cmd_tx: mpsc::Sender<MediaCmd>,
 ) {
@@ -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()

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

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