o_sfu/runtime/
mod.rs

1//! Wires process services and drains them during shutdown.
2//!
3//! Request handlers receive [`RuntimeState`] without boot or teardown control.
4
5use std::{collections::BTreeSet, future::Future, io, process, sync::Arc, time::Duration};
6
7use anyhow::Result as AnyResult;
8use futures_util::{StreamExt, stream::SelectAll};
9use thiserror::Error;
10#[cfg(unix)]
11use tokio::signal::unix::{SignalKind, signal};
12use tokio::{
13    net::TcpListener,
14    runtime::{Builder, Runtime as TokioRuntime},
15    signal::ctrl_c,
16    sync::Semaphore,
17    task::JoinHandle,
18    time::{MissedTickBehavior, interval, sleep},
19};
20use tokio_util::{
21    sync::CancellationToken,
22    task::{AbortOnDropHandle, TaskTracker},
23};
24use tracing::{info, warn};
25
26use crate::{config::Config, core::prelude::SfuCore};
27
28pub(crate) mod auth;
29pub(crate) mod diagnostics;
30pub(crate) mod http_server;
31pub(crate) mod options;
32pub(crate) mod request_origin;
33#[cfg(test)]
34#[path = "TESTS/support.rs"]
35pub(super) mod test_support;
36pub(crate) mod websocket_server;
37
38use http_server::{serve_http, serve_http_on};
39pub(crate) use o_sfu_core::{
40    prelude::{ConnectionId, SessionBitrateLimits},
41    server::{metrics, packet_sinks, room, transport as media_transport},
42};
43pub(crate) use o_sfu_telemetry::{self as telemetry, prometheus};
44use options::{RuntimeConfig, effective_feature_flags};
45use room::{RoomAdmissionPolicy, RoomManager, RoomRuntimePolicy};
46use telemetry::{init_tracing, schema::event as telemetry_event};
47
48pub(crate) use self::{
49    media_transport::{MediaTransport, MediaTransportConfig, MediaTransportDeps},
50    metrics::RuntimeMetrics,
51    packet_sinks::RoomPacketSinkRegistry,
52};
53
54/// Failure to serve or fully drain a [`Runtime`].
55#[derive(Debug, Error)]
56pub enum ServeError {
57    /// The listener or process shutdown signal failed.
58    #[error(transparent)]
59    Io(#[from] io::Error),
60    /// The deadline elapsed before runtime drainage finished.
61    #[error(
62        "runtime shutdown exceeded its deadline with {remaining_sessions} sessions or HTTP mutations remaining"
63    )]
64    ShutdownIncomplete {
65        /// Tracked WebSocket sessions and HTTP mutations that had not returned.
66        remaining_sessions: usize,
67    },
68}
69
70/// Process services and lifecycle configuration.
71#[derive(Debug)]
72pub struct Runtime {
73    config: RuntimeConfig,
74    room_manager: Arc<RoomManager>,
75    metrics: Arc<RuntimeMetrics>,
76    media_transport: MediaTransport,
77}
78
79/// Cloneable request dependencies without process lifecycle control.
80#[derive(Debug, Clone)]
81pub(super) struct RuntimeState {
82    config: RuntimeConfig,
83    room_manager: Arc<RoomManager>,
84    media_transport: MediaTransport,
85    sfu_core: SfuCore,
86    metrics: Arc<RuntimeMetrics>,
87    pre_auth_websocket_admission: websocket_server::PreAuthWebSocketAdmission,
88    session_shutdown: CancellationToken,
89    session_tasks: TaskTracker,
90    http_mutation_admission: Arc<Semaphore>,
91}
92
93#[derive(Default)]
94pub(super) struct RuntimeServices {
95    metrics: Arc<RuntimeMetrics>,
96    packet_sink_registry: Arc<RoomPacketSinkRegistry>,
97}
98
99impl Runtime {
100    /// Builds the room manager and media workers from loaded configuration.
101    ///
102    /// # Errors
103    ///
104    /// Returns [`anyhow::Error`] when the authentication key is malformed or
105    /// shorter than the HS256 minimum, the diagnostics token is invalid, proxy
106    /// mode lacks trusted proxies, listener limits are outside their supported
107    /// ranges or the media transport cannot be built.
108    pub fn new(config: &Config) -> AnyResult<Self> {
109        Self::from_services(config, RuntimeServices::default())
110    }
111
112    #[cfg(feature = "testing-transport")]
113    #[doc(hidden)]
114    #[must_use]
115    pub fn media_transport_for_test(&self) -> MediaTransport {
116        self.media_transport.clone()
117    }
118
119    fn from_services(config: &Config, services: RuntimeServices) -> AnyResult<Self> {
120        config.http.validate()?;
121        config.diagnostics.validate()?;
122        let runtime_config = RuntimeConfig::from_config(config)?;
123        let media_transport = build_media_transport(config, &services)?;
124        let room_runtime_policy = build_room_runtime_policy(config, &media_transport);
125        info!(
126            event = telemetry_event::RUNTIME_BOOT,
127            rtc_udp_io_backend = config.transport.rtc_udp_io_backend.wire_name(),
128            "runtime configuration loaded"
129        );
130        let room_manager = build_room_manager(
131            room_runtime_policy,
132            &services,
133            config.user.room_reservation_ttl,
134            config.user.departure_grace,
135        );
136        Ok(Self {
137            config: runtime_config,
138            room_manager,
139            metrics: services.metrics,
140            media_transport,
141        })
142    }
143
144    /// Serves a caller-provided listener until `shutdown` resolves.
145    ///
146    /// Tokenless operator access follows the listener's actual local address.
147    ///
148    /// # Errors
149    ///
150    /// Returns [`ServeError::Io`] for serving failures or
151    /// [`ServeError::ShutdownIncomplete`] when the drainage deadline expires.
152    pub async fn serve_listener(
153        self,
154        listener: TcpListener,
155        shutdown: impl Future<Output = ()>,
156    ) -> Result<(), ServeError> {
157        self.serve(
158            |state, token| serve_http_on(listener, state, token),
159            async move {
160                shutdown.await;
161                Ok(())
162            },
163        )
164        .await
165    }
166
167    async fn serve<F, HttpServer, Shutdown>(
168        self,
169        http_server: F,
170        shutdown: Shutdown,
171    ) -> Result<(), ServeError>
172    where
173        F: FnOnce(RuntimeState, CancellationToken) -> HttpServer,
174        HttpServer: Future<Output = io::Result<()>>,
175        Shutdown: Future<Output = io::Result<()>>,
176    {
177        let timeout = self.config.http.shutdown_timeout.as_duration();
178        let tasks = RuntimeTasks::spawn(Arc::clone(&self.room_manager), self.media_transport);
179        let state = RuntimeState::from_parts(
180            self.config,
181            self.room_manager,
182            self.metrics,
183            tasks.media_transport.clone(),
184            tasks.session_shutdown.clone(),
185            tasks.session_tasks.clone(),
186        );
187        let listener_shutdown = tasks.shutdown_token.child_token();
188        let server = http_server(state, listener_shutdown.clone());
189        tokio::pin!(server);
190        tokio::pin!(shutdown);
191        let (server_done, mut failure) = tokio::select! {
192            result = &mut server => (true, result.err()),
193            result = &mut shutdown => {
194                listener_shutdown.cancel();
195                (false, result.err())
196            }
197        };
198        let deadline = sleep(timeout);
199        tokio::pin!(deadline);
200        if let Some(error) = &failure {
201            warn!(?error, "runtime serving stopped with an error");
202        }
203        let session_tasks = tasks.session_tasks.clone();
204        let teardown = async move {
205            if !server_done && let Err(error) = server.await {
206                warn!(?error, "HTTP server stopped with an error during shutdown");
207                failure.get_or_insert(error);
208            }
209            tasks.shutdown().await;
210            failure.map_or(Ok(()), |error| Err(ServeError::Io(error)))
211        };
212        tokio::select! {
213            biased;
214            () = &mut deadline => Err(ServeError::ShutdownIncomplete {
215                remaining_sessions: session_tasks.len(),
216            }),
217            result = teardown => result,
218        }
219    }
220}
221
222/// Cancels process work when the server future is dropped.
223struct RuntimeTasks {
224    shutdown_token: CancellationToken,
225    session_shutdown: CancellationToken,
226    session_tasks: TaskTracker,
227    source_packet_policy_sync: AbortOnDropHandle<()>,
228    rooms_reservations_reaper: AbortOnDropHandle<()>,
229    media_transport: MediaTransport,
230}
231
232impl RuntimeTasks {
233    fn spawn(room_manager: Arc<RoomManager>, media_transport: MediaTransport) -> Self {
234        let shutdown_token = CancellationToken::new();
235        let source_packet_policy_sync = spawn_source_packet_policy_update_task(
236            Arc::clone(&room_manager),
237            media_transport.clone(),
238            shutdown_token.child_token(),
239        );
240        let rooms_reservations_reaper =
241            spawn_room_reservation_expiration_reaper(room_manager, shutdown_token.child_token());
242        Self {
243            session_shutdown: shutdown_token.child_token(),
244            shutdown_token,
245            session_tasks: TaskTracker::new(),
246            source_packet_policy_sync: AbortOnDropHandle::new(source_packet_policy_sync),
247            rooms_reservations_reaper: AbortOnDropHandle::new(rooms_reservations_reaper),
248            media_transport,
249        }
250    }
251
252    async fn shutdown(mut self) {
253        self.session_tasks.close();
254        self.session_shutdown.cancel();
255        self.session_tasks.wait().await;
256        self.shutdown_token.cancel();
257        if let Err(error) = (&mut self.source_packet_policy_sync).await
258            && !error.is_cancelled()
259        {
260            warn!(
261                ?error,
262                task = "source packet policy update",
263                "runtime background task stopped unexpectedly"
264            );
265        }
266        if let Err(error) = (&mut self.rooms_reservations_reaper).await
267            && !error.is_cancelled()
268        {
269            warn!(
270                ?error,
271                task = "rooms reservation reaper",
272                "runtime background task stopped unexpectedly"
273            );
274        }
275        self.media_transport.shutdown().await;
276    }
277}
278
279impl Drop for RuntimeTasks {
280    fn drop(&mut self) {
281        self.shutdown_token.cancel();
282        self.media_transport.cancel();
283    }
284}
285
286impl RuntimeState {
287    fn from_parts(
288        config: RuntimeConfig,
289        rooms: Arc<RoomManager>,
290        metrics: Arc<RuntimeMetrics>,
291        media_transport: MediaTransport,
292        session_shutdown: CancellationToken,
293        session_tasks: TaskTracker,
294    ) -> Self {
295        let sfu_core = SfuCore::new(media_transport.clone(), Arc::clone(&rooms));
296        let pre_auth_websocket_admission = websocket_server::PreAuthWebSocketAdmission::new(
297            config.auth.max_pre_auth_websocket_sessions,
298            config.auth.max_pre_auth_websocket_sessions_per_origin,
299        );
300        let http_mutation_admission = Arc::new(Semaphore::new(config.http.max_http_connections));
301        Self {
302            config,
303            room_manager: rooms,
304            media_transport,
305            sfu_core,
306            metrics,
307            pre_auth_websocket_admission,
308            session_shutdown,
309            session_tasks,
310            http_mutation_admission,
311        }
312    }
313}
314
315async fn shutdown_signal() -> io::Result<()> {
316    #[cfg(unix)]
317    {
318        let mut terminate = signal(SignalKind::terminate())?;
319        tokio::select! {
320            result = ctrl_c() => result,
321            _ = terminate.recv() => Ok(()),
322        }
323    }
324    #[cfg(not(unix))]
325    ctrl_c().await
326}
327
328/// Recomputes room packet policy after transport observations change.
329fn spawn_source_packet_policy_update_task(
330    rooms: Arc<RoomManager>,
331    media_transport: MediaTransport,
332    shutdown_token: CancellationToken,
333) -> JoinHandle<()> {
334    info!("booted source packet policy update task");
335    let updates = media_transport.source_policy_subscription();
336    tokio::spawn(async move {
337        let mut pending = BTreeSet::new();
338        let mut running = BTreeSet::new();
339        let mut batches = SelectAll::new();
340        loop {
341            tokio::select! {
342                biased;
343                () = shutdown_token.cancelled() => break,
344                completed = batches.next(), if !batches.is_empty() => {
345                    if let Some(room_id) = completed {
346                        running.remove(&room_id);
347                    }
348                },
349                dirty_rooms = updates.wait_for_update() => pending.extend(dirty_rooms),
350            }
351            let ready = pending
352                .difference(&running)
353                .copied()
354                .collect::<BTreeSet<_>>();
355            if ready.is_empty() {
356                continue;
357            }
358            pending.retain(|room| !ready.contains(room));
359            running.extend(ready.iter().copied());
360            batches.push(rooms.source_policy_turns(&ready, &media_transport).await);
361        }
362        // A policy turn can have accepted transport effects. Finish those turns
363        // before shutdown releases their room state.
364        batches.for_each(|_| async {}).await;
365    })
366}
367
368fn spawn_room_reservation_expiration_reaper(
369    rooms: Arc<RoomManager>,
370    shutdown_token: CancellationToken,
371) -> JoinHandle<()> {
372    info!("booted room reservation expiration reaper");
373    tokio::spawn(async move {
374        let mut interval = interval(Duration::from_secs(10));
375        interval.set_missed_tick_behavior(MissedTickBehavior::Skip);
376        loop {
377            tokio::select! {
378                biased;
379                () = shutdown_token.cancelled() => return,
380                _ = interval.tick() => {},
381            };
382            rooms.check_expired_room_reservations().await;
383        }
384    })
385}
386
387fn build_media_transport(config: &Config, services: &RuntimeServices) -> AnyResult<MediaTransport> {
388    Ok(MediaTransport::build(
389        MediaTransportConfig {
390            worker_count: config.transport.rtc_media_worker_count,
391            announced_ip: config.transport.announced_ip,
392            bitrate_limits: SessionBitrateLimits::new(
393                config.transport.max_bitrate_in,
394                config.transport.max_bitrate_out,
395            ),
396            video_bitrate_limits: config.transport.video_bitrate_limits,
397            rtc_port_range: config.transport.rtc_port_range,
398            rtc_udp_io_backend: config.transport.rtc_udp_io_backend,
399            rtc_tcp_config: config.transport.rtc_tcp_config,
400            codec_flags: config.codecs.flags,
401            codec_preferences: config.codecs.preferences,
402            media_quality_interval: config.telemetry.media_quality_interval,
403        },
404        MediaTransportDeps {
405            packet_sink_registry: Arc::clone(&services.packet_sink_registry),
406            metrics: Arc::clone(&services.metrics),
407        },
408    )?)
409}
410
411fn build_room_runtime_policy(
412    config: &Config,
413    media_transport: &MediaTransport,
414) -> RoomRuntimePolicy {
415    RoomRuntimePolicy::new(
416        RoomAdmissionPolicy::new(config.user.room_size),
417        effective_feature_flags(config.features),
418        media_transport.router_rtp_capabilities(),
419    )
420    .with_room_worker_policy(config.transport.room_worker_policy)
421    .with_media_limits(config.transport.room_media_limits)
422    .with_video_adaptation_tuning(config.transport.video_adaptation_tuning)
423}
424
425fn build_room_manager(
426    runtime_policy: RoomRuntimePolicy,
427    services: &RuntimeServices,
428    reservation_ttl: Duration,
429    departure_grace: Duration,
430) -> Arc<RoomManager> {
431    Arc::new(RoomManager::new(
432        runtime_policy,
433        Arc::clone(&services.metrics),
434        reservation_ttl,
435        departure_grace,
436    ))
437}
438
439fn build_tokio_runtime(config: &Config) -> io::Result<TokioRuntime> {
440    Builder::new_multi_thread()
441        .worker_threads(config.runtime_thread_count)
442        .enable_all()
443        .build()
444}
445
446/// # Errors
447///
448/// Returns [`anyhow::Error`] when configuration, tracing, Tokio startup, serving
449/// or shutdown fails.
450pub fn run() -> AnyResult<()> {
451    let config = Config::from_env()?;
452    let _telemetry = init_tracing(&config.telemetry, process::id())?;
453    let runtime = Runtime::new(&config)?;
454    // Telemetry must outlive Tokio teardown so dropped tasks can finish their spans.
455    Ok(build_tokio_runtime(&config)?.block_on(runtime.serve(serve_http, shutdown_signal()))?)
456}
457
458#[cfg(test)]
459#[path = "TESTS/runtime.rs"]
460mod tests;
o_sfu/runtime/mod.rs

o_sfu/runtime/
mod.rs
