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

o_sfu_core/engine/media_transport/rtc/
bootstrap.rs
