o_sfu_core/
sfu.rs

1//! [`SfuCore::admit_user`] returns one [`MediaSession`] per admitted room connection.
2//!
3//! The session sequences offer/answer, publication, subscription, recording and
4//! cleanup without exposing room or transport internals.
5//!
6//! ```text
7//! SfuCore::admit_user -> MediaSession
8//!
9//! establish             -> offer -> answer
10//! publish               -> [offer -> answer]?
11//! deactivate_publication -> cancel pending or suppress committed source
12//! subscribe             -> persist intent and reconcile eligible routes
13//! close                 -> remove only the current connection
14//! ```
15//!
16//! `&mut MediaSession` serializes negotiation. [`MediaSession::publish`] queues
17//! the first intent for each stream while an offer awaits its answer. A successful
18//! [`MediaSession::answer`] may return the follow-up offer.
19use std::{collections::BTreeMap, mem::replace, sync::Arc};
20
21pub use crate::engine::media_transport::{
22    SessionOffer as NegotiationOffer, SessionUploadEncoding as UploadEncoding,
23    SessionUploadSlot as UploadSlot,
24};
25use crate::{
26    ConnectionId,
27    engine::{
28        AvailableFeatures, JsonPayload, PeerSnapshot, RecordingOptions, RecordingState, UserId,
29        UserInfo,
30        media_transport::{
31            MediaTransport, TransportAdapterError, TransportSessionHealth, TransportSessionKey,
32        },
33        room::{
34            BroadcastPayloadError, DeactivateIntentOutcome, JoinUserRequest, PublishIntentOutcome,
35            Room, RoomManager, RoomManagerJoinError, RoomUserOperation,
36        },
37        source_model::{
38            SourceDeactivateIntent, SourcePublishIntent, SourceSubscriptionIntent, UserStreamId,
39        },
40    },
41};
42
43/// media-session negotiation phase
44///
45/// queued publishes live beside the offer state so a user action that arrives
46/// while a browser answer is pending cannot interleave with the in-flight SDP
47/// exchange
48#[derive(Debug, Default)]
49enum SessionPhase {
50    /// no initial offer has been created yet
51    #[default]
52    BeforeInitialOffer,
53    /// no offer is awaiting an answer
54    Stable,
55    /// one offer has been sent and must be answered before the next offer
56    WaitingForAnswer(InFlightOffer),
57}
58
59/// in-flight offer state plus mutations deferred until its answer is accepted
60#[derive(Debug)]
61struct InFlightOffer {
62    purpose: SessionOfferPurpose,
63    queued_publishes: BTreeMap<UserStreamId, SourcePublishIntent>,
64    follow_up_renegotiation: bool,
65}
66
67impl SessionPhase {
68    fn can_stage_publish(&self) -> bool {
69        !matches!(self, Self::WaitingForAnswer(_))
70    }
71
72    fn has_queued_publish(&self, stream_id: &UserStreamId) -> bool {
73        matches!(
74            self,
75            Self::WaitingForAnswer(pending)
76                if pending.queued_publishes.contains_key(stream_id)
77        )
78    }
79
80    fn queue_publish(&mut self, intent: SourcePublishIntent) {
81        if let Self::WaitingForAnswer(pending) = self {
82            let stream_id = intent.stream_id().clone();
83            pending.queued_publishes.insert(stream_id, intent);
84        }
85    }
86
87    fn remove_queued_publish(&mut self, stream_id: &UserStreamId) -> bool {
88        let Self::WaitingForAnswer(pending) = self else {
89            return false;
90        };
91        pending.queued_publishes.remove(stream_id).is_some()
92    }
93
94    fn clear_queued_publishes(&mut self) {
95        if let Self::WaitingForAnswer(pending) = self {
96            pending.queued_publishes.clear();
97        }
98    }
99
100    fn request_renegotiation(&mut self) -> bool {
101        match self {
102            Self::BeforeInitialOffer => false,
103            Self::Stable => true,
104            Self::WaitingForAnswer(pending) => {
105                pending.follow_up_renegotiation = true;
106                false
107            }
108        }
109    }
110
111    fn mark_follow_up_renegotiation(&mut self) {
112        if let Self::WaitingForAnswer(pending) = self {
113            pending.follow_up_renegotiation = true;
114        }
115    }
116
117    fn wait_for_answer(&mut self, purpose: SessionOfferPurpose) {
118        *self = Self::WaitingForAnswer(InFlightOffer {
119            purpose,
120            queued_publishes: BTreeMap::new(),
121            follow_up_renegotiation: false,
122        });
123    }
124
125    #[expect(
126        clippy::unreachable,
127        reason = "answer validates the phase before awaiting with exclusive session access"
128    )]
129    fn complete_answer(&mut self) -> InFlightOffer {
130        match replace(self, Self::Stable) {
131            Self::WaitingForAnswer(pending) => pending,
132            _ => unreachable!("answer completion requires an in-flight offer"),
133        }
134    }
135}
136
137/// reason an offer is waiting for an answer
138#[derive(Debug)]
139enum SessionOfferPurpose {
140    EstablishSession,
141    RefreshSession,
142}
143
144/// error returned by [`MediaSession`] operations
145#[derive(Debug, Clone, Copy, PartialEq, Eq, thiserror::Error)]
146pub enum SessionError {
147    /// [`MediaSession::answer`] was called without an in-flight offer
148    #[error("no pending media request")]
149    NoPendingRequest,
150    /// lower core operation failed or rejected the request
151    #[error(transparent)]
152    Core(#[from] SfuCoreError),
153}
154
155impl SessionError {
156    /// whether the runtime should report the error as a client or protocol fault
157    ///
158    /// transport failures that indicate malformed client input are client
159    /// errors, while infrastructure failures are internal errors
160    #[must_use]
161    pub const fn is_client_error(self) -> bool {
162        match self {
163            Self::NoPendingRequest => true,
164            Self::Core(error) => error.is_client_error(),
165        }
166    }
167}
168
169#[derive(Debug, Clone, Copy, PartialEq, Eq, thiserror::Error)]
170pub enum SfuCoreError {
171    /// media transport command failed
172    #[error("transport operation failed")]
173    Transport(#[source] TransportAdapterError),
174    /// accepted answer did not yield client capabilities needed by room state
175    #[error("capability projection failed")]
176    CapabilityProjection(#[source] TransportAdapterError),
177    /// initial answer was valid transport input but stale for room state
178    #[error("session negotiation rejected")]
179    SessionNegotiationRejected,
180    /// refresh answer was valid transport input but stale for room state
181    #[error("session refresh rejected")]
182    SessionRefreshRejected,
183    /// subscription intent targeted a stale or invalid room connection
184    #[error("subscription update rejected")]
185    SubscriptionUpdateRejected,
186}
187
188impl SfuCoreError {
189    /// whether this error should close the client as a protocol fault
190    #[must_use]
191    pub const fn is_client_error(self) -> bool {
192        matches!(
193            self,
194            Self::Transport(TransportAdapterError::InvalidInput)
195                | Self::CapabilityProjection(_)
196                | Self::SessionNegotiationRejected
197                | Self::SessionRefreshRejected
198                | Self::SubscriptionUpdateRejected
199        )
200    }
201}
202
203/// A cloneable core handle that admits users into room-bound [`MediaSession`]s.
204#[derive(Debug, Clone)]
205pub struct SfuCore {
206    media_transport: MediaTransport,
207    rooms: Arc<RoomManager>,
208}
209
210/// One admitted user connection in one room.
211///
212/// Room mutations revalidate the connection before committing room state.
213/// [`close`](Self::close) cannot remove a replacement connection and drains
214/// connection-scoped staged media when this session is current.
215///
216/// Futures returned by [`establish`](Self::establish), [`answer`](Self::answer),
217/// [`publish`](Self::publish),
218/// [`deactivate_publication`](Self::deactivate_publication),
219/// [`renegotiate`](Self::renegotiate), [`subscribe`](Self::subscribe),
220/// [`update_info`](Self::update_info) and [`close`](Self::close) are not
221/// cancellation safe. Once polled, await them to completion. Negotiation can
222/// otherwise leave room and transport state out of step with the local phase.
223/// Other mutations can leave committed room state without its transport or
224/// output effects.
225///
226/// Call [`close`](Self::close) before dropping the session. `Drop` performs no
227/// room or transport cleanup.
228#[derive(Debug)]
229pub struct MediaSession {
230    core: SfuCore,
231    room: Arc<Room>,
232    transport_user_key: TransportSessionKey,
233    phase: SessionPhase,
234    closed: bool,
235}
236
237impl SfuCore {
238    #[must_use]
239    pub fn new(media_transport: MediaTransport, rooms: Arc<RoomManager>) -> Self {
240        Self {
241            media_transport,
242            rooms,
243        }
244    }
245
246    /// Admits one room user and returns its connection-owning media session.
247    ///
248    /// An existing `request.user_id` replaces its current connection.
249    /// Replacement room and transport effects finish before return.
250    /// This future is not cancellation safe. Once polled, await it to completion.
251    /// Dropping it after membership commits can leave room or replacement
252    /// transport effects unfinished.
253    ///
254    /// # Errors
255    ///
256    /// Returns [`RoomManagerJoinError::MissingRoom`] when `room_id` is not
257    /// current. Returns [`RoomManagerJoinError::RoomFull`] when a new user
258    /// exceeds room capacity. Returns [`RoomManagerJoinError::RouterState`] when
259    /// router placement cannot commit.
260    ///
261    /// # Panics
262    ///
263    /// Panics when existing relay state refers to an uncommitted source
264    /// placement.
265    pub async fn admit_user(
266        &self,
267        room_id: &str,
268        request: JoinUserRequest,
269    ) -> Result<MediaSession, RoomManagerJoinError> {
270        let admission = self
271            .rooms
272            .join_user(room_id, request, &self.media_transport)
273            .await?;
274        Ok(MediaSession {
275            core: self.clone(),
276            room: admission.room,
277            transport_user_key: admission.transport_session_key,
278            phase: SessionPhase::default(),
279            closed: false,
280        })
281    }
282}
283
284impl MediaSession {
285    /// creates the first browser offer for this connection
286    ///
287    /// returns `Ok(None)` after the initial offer has already been requested
288    /// this lets reconnect or duplicate-start paths retry safely without
289    /// creating a second initial offer
290    ///
291    /// # Errors
292    ///
293    /// returns [`SessionError::Core`] when the transport cannot create the
294    /// initial offer
295    pub async fn establish(&mut self) -> Result<Option<NegotiationOffer>, SessionError> {
296        if !matches!(self.phase, SessionPhase::BeforeInitialOffer) {
297            return Ok(None);
298        }
299        let offer = self
300            .core
301            .media_transport
302            .create_initial_session_offer(self.room.uuid(), &self.transport_user_key)
303            .await
304            .map_err(SfuCoreError::Transport)?;
305        self.phase
306            .wait_for_answer(SessionOfferPurpose::EstablishSession);
307        Ok(Some(offer))
308    }
309
310    /// accepts the answer for the pending offer and commits any ready room work
311    ///
312    /// a rejection before the worker consumes its pending offer leaves the
313    /// application round in place so a direct caller may retry it
314    /// failures after the worker consumes that offer do not promise retry even
315    /// when the RTC backend rejects the answer
316    /// when queued publish intent needs another SDP round the returned offer
317    /// must be sent to the client before the next answer
318    ///
319    /// # Errors
320    ///
321    /// returns [`SessionError::NoPendingRequest`] when no offer is pending
322    /// returns [`SessionError::Core`] when answer application fails, capability
323    /// projection fails or room state rejects the accepted answer as stale
324    pub async fn answer(&mut self, sdp: &str) -> Result<Option<NegotiationOffer>, SessionError> {
325        if !matches!(self.phase, SessionPhase::WaitingForAnswer(_)) {
326            return Err(SessionError::NoPendingRequest);
327        }
328        let applied_answer = self
329            .core
330            .media_transport
331            .apply_session_answer(&self.transport_user_key, sdp)
332            .await
333            .map_err(SfuCoreError::Transport)?;
334        let InFlightOffer {
335            purpose,
336            queued_publishes,
337            follow_up_renegotiation,
338        } = self.phase.complete_answer();
339        match purpose {
340            SessionOfferPurpose::EstablishSession => {
341                let client_capabilities = applied_answer.client_capabilities().cloned().ok_or(
342                    SfuCoreError::CapabilityProjection(TransportAdapterError::InvalidInput),
343                )?;
344                self.room_operation()
345                    .apply_session_negotiated(
346                        client_capabilities,
347                        applied_answer.declined_consumers(),
348                    )
349                    .await
350                    .ok_or(SfuCoreError::SessionNegotiationRejected)?;
351            }
352            SessionOfferPurpose::RefreshSession => {
353                self.room_operation()
354                    .apply_session_refreshed(applied_answer.declined_consumers())
355                    .await
356                    .ok_or(SfuCoreError::SessionRefreshRejected)?;
357            }
358        }
359        self.room_operation()
360            .commit_staged_publishes(&applied_answer)
361            .await;
362        let staged = self.stage_queued_publishes(queued_publishes).await?;
363        if staged || follow_up_renegotiation {
364            return self.renegotiate().await;
365        }
366        Ok(None)
367    }
368
369    /// applies publish intent for one user stream
370    ///
371    /// returns no offer when the intent is already queued, already active or
372    /// must wait for an in-flight answer
373    /// returns an offer when the browser must answer a new offer before the
374    /// publication can commit
375    ///
376    /// # Errors
377    ///
378    /// returns [`SessionError::Core`] when the media backend cannot stage a
379    /// publish that needs negotiation
380    pub async fn publish(
381        &mut self,
382        intent: SourcePublishIntent,
383    ) -> Result<Option<NegotiationOffer>, SessionError> {
384        if self.phase.has_queued_publish(intent.stream_id()) {
385            return Ok(None);
386        }
387        match self
388            .start_publish(&intent, self.phase.can_stage_publish())
389            .await?
390        {
391            PublishIntentOutcome::Noop | PublishIntentOutcome::Activated => Ok(None),
392            PublishIntentOutcome::Queue => {
393                self.phase.queue_publish(intent);
394                Ok(None)
395            }
396            PublishIntentOutcome::Staged => self.renegotiate().await,
397        }
398    }
399
400    /// deactivates one publication without changing negotiated media
401    ///
402    /// a queued first publication is cancelled
403    /// a staged first publication is rolled back and its pending answer creates
404    /// the cleanup offer
405    /// a committed publication keeps its source identity, routes and negotiated
406    /// MID until session teardown
407    pub async fn deactivate_publication(&mut self, intent: SourceDeactivateIntent) {
408        if self.phase.remove_queued_publish(intent.stream_id()) {
409            return;
410        }
411        match self.room_operation().deactivate_publication(&intent).await {
412            DeactivateIntentOutcome::RolledBack => {
413                self.phase.mark_follow_up_renegotiation();
414            }
415            DeactivateIntentOutcome::Deactivated | DeactivateIntentOutcome::Noop => {}
416        }
417    }
418
419    /// closes this media session and removes its room connection if still current
420    ///
421    /// the call is idempotent
422    /// it returns `true` only when the room manager removed the current
423    /// connection
424    /// current-session cleanup drains connection-scoped staged media through
425    /// room state
426    /// stale sessions do not remove a replacement connection for the same user
427    pub async fn close(&mut self) -> bool {
428        if self.closed {
429            return false;
430        }
431        self.phase.clear_queued_publishes();
432        let did_close = self
433            .core
434            .rooms
435            .close_session(
436                self.room_id(),
437                self.user_id(),
438                self.connection_id(),
439                &self.core.media_transport,
440            )
441            .await;
442        self.closed = true;
443        did_close
444    }
445
446    /// creates a refresh offer when the stable session needs renegotiation
447    ///
448    /// returns `Ok(None)` before the initial offer, while an answer is pending
449    /// or when the transport reports that the requested refresh is unsupported
450    /// a call made while an answer is pending records that another offer should
451    /// be created after the answer commits
452    ///
453    /// # Errors
454    ///
455    /// returns [`SessionError::Core`] when the transport rejects
456    /// renegotiation
457    pub async fn renegotiate(&mut self) -> Result<Option<NegotiationOffer>, SessionError> {
458        if !self.phase.request_renegotiation() {
459            return Ok(None);
460        }
461        let offer = match self
462            .core
463            .media_transport
464            .create_session_renegotiation_offer(&self.transport_user_key)
465            .await
466        {
467            Ok(offer) => offer,
468            Err(TransportAdapterError::UnsupportedFeature) => return Ok(None),
469            Err(error) => return Err(SfuCoreError::Transport(error).into()),
470        };
471        self.phase
472            .wait_for_answer(SessionOfferPurpose::RefreshSession);
473        Ok(Some(offer))
474    }
475
476    /// applies receiver intent for sources published by another user
477    ///
478    /// subscription intent is remembered even when no producer is currently
479    /// routable
480    /// once negotiation makes the receiver consumable, room effects create the
481    /// missing consumer routes
482    ///
483    /// # Errors
484    ///
485    /// returns [`SessionError::Core`] when room state rejects this connection
486    /// as stale
487    pub async fn subscribe(
488        &self,
489        target_user_id: &UserId,
490        intents: &BTreeMap<UserStreamId, SourceSubscriptionIntent>,
491    ) -> Result<(), SessionError> {
492        self.room_operation()
493            .apply_receiver_intent(target_user_id, intents)
494            .await
495            .ok_or(SfuCoreError::SubscriptionUpdateRejected)?;
496        Ok(())
497    }
498
499    /// returns `None` when the transport has no endpoint for this session key
500    #[must_use]
501    pub fn endpoint_health(&self) -> Option<TransportSessionHealth> {
502        self.core
503            .media_transport
504            .session_transport_health(&self.transport_user_key)
505    }
506
507    #[must_use]
508    pub fn user_id(&self) -> &UserId {
509        self.transport_user_key.user_id()
510    }
511
512    #[must_use]
513    pub const fn connection_id(&self) -> ConnectionId {
514        self.transport_user_key.connection_id()
515    }
516
517    #[must_use]
518    pub fn room_id(&self) -> &str {
519        self.room.uuid()
520    }
521
522    pub async fn is_current_connection(&self) -> bool {
523        self.room
524            .has_connection(self.user_id(), self.connection_id())
525            .await
526    }
527
528    #[must_use]
529    pub fn available_features(&self) -> AvailableFeatures {
530        self.room.available_features()
531    }
532
533    pub async fn recording_state(&self) -> RecordingState {
534        self.room.recording_state().await
535    }
536
537    /// Returns snapshots for current room users other than this session's user.
538    pub async fn peer_snapshots(&self) -> Vec<PeerSnapshot> {
539        self.room.user_snapshots_except(self.user_id()).await
540    }
541
542    fn room_operation(&self) -> RoomUserOperation<'_> {
543        self.room.user_operation(
544            self.user_id(),
545            self.connection_id(),
546            &self.core.media_transport,
547        )
548    }
549
550    async fn start_publish(
551        &self,
552        intent: &SourcePublishIntent,
553        can_stage: bool,
554    ) -> Result<PublishIntentOutcome, SfuCoreError> {
555        self.room_operation()
556            .start_publish(intent, can_stage)
557            .await
558            .map_err(SfuCoreError::Transport)
559    }
560
561    /// Updates user information for this room connection.
562    ///
563    /// Missing or stale connections are ignored. `is_camera_on` and
564    /// `is_screen_sharing_on` are derived from publication state and discarded
565    /// from `info`.
566    pub async fn update_info(&self, info: UserInfo) {
567        self.room
568            .update_user_info(
569                self.user_id(),
570                self.connection_id(),
571                &self.core.media_transport,
572                info,
573            )
574            .await;
575    }
576
577    /// Broadcasts `message` to every other current room user.
578    ///
579    /// Missing or stale sender connections return `Ok(())` without delivery.
580    ///
581    /// # Errors
582    ///
583    /// Returns [`BroadcastPayloadError::TooLarge`] when the serialized payload
584    /// exceeds the room broadcast limit. Returns
585    /// [`BroadcastPayloadError::JsonSerialization`] when JSON serialization
586    /// fails.
587    pub async fn broadcast(&self, message: JsonPayload) -> Result<(), BroadcastPayloadError> {
588        self.room
589            .broadcast(self.user_id(), self.connection_id(), message)
590            .await
591    }
592
593    /// Rejects recording start because no persistent recording backend is enabled.
594    ///
595    /// `options` has no effect. The request records a rejection and returns
596    /// `false`.
597    #[must_use]
598    #[expect(
599        clippy::unused_async,
600        reason = "keeps the public MediaSession recording facade async while disabled recording is synchronous"
601    )]
602    pub async fn start_recording(&self, options: RecordingOptions) -> bool {
603        self.room
604            .apply_recording_start(self.user_id(), self.connection_id(), options)
605    }
606
607    /// Rejects recording stop because no persistent recording backend is enabled.
608    ///
609    /// The request records a rejection and returns `false`.
610    #[must_use]
611    #[expect(
612        clippy::unused_async,
613        reason = "keeps the public MediaSession recording facade async while disabled recording is synchronous"
614    )]
615    pub async fn stop_recording(&self) -> bool {
616        self.room
617            .apply_recording_stop(self.user_id(), self.connection_id())
618    }
619
620    async fn stage_queued_publishes(
621        &self,
622        queued: BTreeMap<UserStreamId, SourcePublishIntent>,
623    ) -> Result<bool, SessionError> {
624        let mut staged = false;
625        for intent in queued.into_values() {
626            if self.start_publish(&intent, true).await? == PublishIntentOutcome::Staged {
627                staged = true;
628            }
629        }
630        Ok(staged)
631    }
632}
o_sfu_core/sfu.rs

o_sfu_core/
sfu.rs
