o_sfu_core/engine/media_transport/rtc/
bootstrap.rs

1//! cold-path rtc socket and session bootstrap
2//!
3//! this module creates worker sockets during construction and session-local
4//! str0m state during negotiation
5//!
6//! bootstrap stops at transport setup
7//! room policy, media registration, SDP
8//! staging and packet routing stay in the worker modules responsible for those
9//! contracts
10
11use std::{
12    io::ErrorKind,
13    net::{IpAddr, Ipv4Addr, Ipv6Addr, SocketAddr, UdpSocket as StdUdpSocket},
14    sync::Arc,
15    time::{Duration, Instant},
16};
17
18use o_sfu_rfc::webrtc;
19use str0m::{Candidate, bwe::Bitrate as Str0mBitrate};
20use tracing::{info, warn};
21
22use super::{
23    RtpProfile,
24    bitrate::MediaBitrateCounter,
25    local_send_rewrite::{ConsumerStreamStore, RTX_CACHE_MAX_PACKETS},
26    packet_loop::{RtcUdpSocket, UdpIngress},
27    slots::SessionStore,
28    state::{RtcSessionState, SessionSdpNegotiationState, SharedRtcSocket},
29};
30use crate::{
31    Bitrate, RtcPortRange, RtcUdpIoBackend,
32    engine::media_transport::{TransportAdapterError, TransportSessionKey},
33};
34#[cfg(any(test, feature = "internal-benchmarks", fuzzing))]
35use crate::{CodecPreferences, MediaCodecFlags};
36
37/// bind the shared worker UDP socket and return the advertised candidate tuple
38///
39/// the bind address uses the unspecified address for the configured public IP
40/// family so one socket can receive traffic for every session assigned to the
41/// worker
42/// the advertised candidate keeps the configured public IP with the
43/// bound port because that is what browsers must see in SDP
44///
45/// tries configured ports in order and skips only [`ErrorKind::AddrInUse`]
46/// any other bind, nonblocking or backend initialization error aborts startup
47///
48/// # errors
49///
50/// returns [`TransportAdapterError::TransportUnavailable`] when every port is
51/// occupied or another socket operation fails
52pub(super) fn bind_shared_rtc_socket(
53    announced_ip: IpAddr,
54    rtc_port_range: RtcPortRange,
55    rtc_udp_io_backend: RtcUdpIoBackend,
56) -> Result<SharedRtcSocket, TransportAdapterError> {
57    let bind_ip = bind_ip_for_announced_ip(announced_ip);
58    for port in rtc_port_range.ports() {
59        let bind_addr = SocketAddr::new(bind_ip, port);
60        let socket = match StdUdpSocket::bind(bind_addr) {
61            Ok(socket) => socket,
62            Err(error) if error.kind() == ErrorKind::AddrInUse => continue,
63            Err(error) => {
64                warn!(%bind_addr, ?error, "failed to bind shared rtc UDP socket");
65                return Err(TransportAdapterError::TransportUnavailable);
66            }
67        };
68        socket.set_nonblocking(true).map_err(|error| {
69            warn!(%bind_addr, ?error, "failed to configure shared rtc UDP socket");
70            TransportAdapterError::TransportUnavailable
71        })?;
72        let socket = RtcUdpSocket::from_std(socket, rtc_udp_io_backend).map_err(|error| {
73            warn!(
74                %bind_addr,
75                backend = rtc_udp_io_backend.wire_name(),
76                ?error,
77                "failed to initialize shared rtc UDP socket"
78            );
79            TransportAdapterError::TransportUnavailable
80        })?;
81        let candidate_addr = SocketAddr::new(announced_ip, port);
82        let ingress = UdpIngress::new(socket.clone(), bind_addr, candidate_addr);
83        info!(
84            %bind_addr,
85            %candidate_addr,
86            "booted shared rtc UDP socket"
87        );
88        return Ok(SharedRtcSocket {
89            socket,
90            ingress,
91            candidate_addr,
92        });
93    }
94    warn!(
95        %bind_ip,
96        port_min = rtc_port_range.min(),
97        port_max = rtc_port_range.max(),
98        "rtc UDP port range is unavailable"
99    );
100    Err(TransportAdapterError::TransportUnavailable)
101}
102
103/// choose the local bind address that matches the advertised IP family
104///
105/// workers bind to all local interfaces for that family while keeping SDP
106/// candidates anchored to the configured announced IP
107fn bind_ip_for_announced_ip(announced_ip: IpAddr) -> IpAddr {
108    match announced_ip {
109        IpAddr::V4(_) => IpAddr::V4(Ipv4Addr::UNSPECIFIED),
110        IpAddr::V6(_) => IpAddr::V6(Ipv6Addr::UNSPECIFIED),
111    }
112}
113
114/// ensure one worker-local [`RtcSessionState`] exists for a session
115///
116/// this is idempotent because negotiation may ask for readiness more than once
117/// while a session is still alive
118/// `Ok(true)` means a fresh [`str0m::Rtc`] was created and inserted
119/// `Ok(false)` means the existing session state still matches that key
120///
121/// new sessions start in ICE-lite mode, with RTP mode enabled, bandwidth
122/// estimation capped by `max_bitrate_out` and exactly one local host candidate
123/// attached to the shared worker socket's advertised address
124///
125/// # errors
126///
127/// returns `TransportUnavailable` if the local candidate cannot be represented
128/// by str0m or cannot be attached to the newly created rtc state
129#[cfg(any(test, feature = "internal-benchmarks", fuzzing))]
130pub(super) fn ensure_session_rtc_state(
131    users: &mut SessionStore,
132    session_key: &TransportSessionKey,
133    candidate_addr: SocketAddr,
134    max_bitrate_out: Bitrate,
135) -> Result<bool, TransportAdapterError> {
136    let profile = RtpProfile::compile(MediaCodecFlags::default(), CodecPreferences::default())?;
137    ensure_session_rtc_state_with_stats_interval(
138        users,
139        Arc::from("test-room"),
140        session_key,
141        candidate_addr,
142        max_bitrate_out,
143        &profile,
144        None,
145    )
146}
147
148pub(super) fn ensure_session_rtc_state_with_stats_interval(
149    users: &mut SessionStore,
150    room_id: Arc<str>,
151    session_key: &TransportSessionKey,
152    candidate_addr: SocketAddr,
153    max_bitrate_out: Bitrate,
154    profile: &RtpProfile,
155    stats_interval: Option<Duration>,
156) -> Result<bool, TransportAdapterError> {
157    if users.contains_key(session_key) {
158        return Ok(false);
159    }
160    let started_at = Instant::now();
161    let mut config = profile
162        .session_config()
163        .set_send_buffer_video(RTX_CACHE_MAX_PACKETS);
164    if let Some(stats_interval) = stats_interval {
165        config = config.set_stats_interval(Some(stats_interval));
166    }
167    let mut rtc = config
168        .enable_bwe(Some(Str0mBitrate::bps(max_bitrate_out.as_bps())))
169        .set_ice_lite(true)
170        .build(started_at);
171    let candidate = Candidate::host(candidate_addr, webrtc::IceTransport::Udp.as_str())
172        .map_err(|_error| TransportAdapterError::TransportUnavailable)?;
173    if rtc.add_local_candidate(candidate).is_none() {
174        return Err(TransportAdapterError::TransportUnavailable);
175    }
176    let local_ice_ufrag = rtc.direct_api().local_ice_credentials().ufrag;
177    users.insert(
178        session_key.clone(),
179        RtcSessionState {
180            room_id,
181            rtc,
182            started_at,
183            rtcp_ingress_budget: super::state::RtcpIngressBudget::new(started_at),
184            defer_rtx_expiry: false,
185            pending_rtp_input: None,
186            nack_totals: super::state::RtcNackTotals::default(),
187            egress_bitrate: Arc::new(MediaBitrateCounter::new(started_at)),
188            local_ice_ufrag,
189            #[cfg(test)]
190            max_bitrate_in: None,
191            #[cfg(test)]
192            max_bitrate_out: Some(max_bitrate_out),
193            receiver_bwe_target: None,
194            #[cfg(test)]
195            receiver_bwe_str0m_update_count: 0,
196            dtls_started: false,
197            packet_loop_dirty: false,
198            sdp_negotiation: SessionSdpNegotiationState::default(),
199            consumer_streams: ConsumerStreamStore::default(),
200            #[cfg(test)]
201            last_local_write: None,
202        },
203    );
204    Ok(true)
205}
o_sfu_core/engine/media_transport/rtc/bootstrap.rs

o_sfu_core/engine/media_transport/rtc/
bootstrap.rs
