o_sfu_core/engine/room/
membership.rs

1//! Membership commit and effect boundary.
2//!
3//! [`JoinCommit`], [`ConnectionCloseCommit`] and
4//! [`DisconnectCommit`](crate::engine::room::state::DisconnectCommit) capture
5//! authoritative membership changes and the work consumed by [`RoomEffects`].
6//! Effects execute after the room-state write guard is released.
7
8#[cfg(any(test, feature = "testing-transport"))]
9use o_sfu_router::rtp::MediaCapabilities;
10use o_sfu_telemetry::schema::event as telemetry_event;
11use tracing::{info, warn};
12
13use super::{
14    BroadcastPayloadError, Room, RoomJoinError, UserOutboundSender,
15    effects::batch::{RoomEffectContext, RoomEffects},
16    media_graph::CommittedTransportReceipt,
17    placement::JoinAdmissionTurn,
18    state::ConnectionCloseCommit,
19};
20use crate::engine::{
21    ConnectionId, MediaWorkerId, UserId, UserInfo, UserPermissions,
22    media_transport::MediaTransport, room::state::JoinCommit,
23};
24
25/// Room-state marker that collapses every authenticated [`UserPermissions`] value.
26#[derive(Debug, Clone, Copy, Default, PartialEq, Eq)]
27pub struct RoomUserPermissions;
28
29impl From<UserPermissions> for RoomUserPermissions {
30    fn from(_value: UserPermissions) -> Self {
31        Self
32    }
33}
34
35#[derive(Debug, Clone, Copy, PartialEq, Eq)]
36pub enum UserCloseReason {
37    Replaced,
38    RemovedByRuntime,
39}
40
41/// Existing users replace their current connection without consuming another admission slot.
42pub struct JoinUserRequest {
43    pub user_id: UserId,
44    /// Ignored by room admission.
45    pub label: Option<String>,
46    /// Collapsed to [`RoomUserPermissions`] during admission.
47    pub permissions: UserPermissions,
48    pub sender: UserOutboundSender,
49}
50
51impl Room {
52    /// Commits the admission turn with the room router.
53    ///
54    /// # Errors
55    ///
56    /// Returns [`RoomJoinError::RoomFull`] when a new user exceeds capacity.
57    /// Returns [`RoomJoinError::RouterState`] when placement cannot commit.
58    pub(super) async fn commit_admission(
59        &self,
60        admission: JoinAdmissionTurn<'_, impl FnOnce() -> o_sfu_router::RouterId>,
61        context: RoomEffectContext<'_>,
62    ) -> Result<JoinCommit, RoomJoinError> {
63        let joined_fanout = context.user_joined_fanout();
64        admission.commit(self, joined_fanout).await
65    }
66
67    /// Executes context-enabled [`RoomEffects`] before returning the committed receipt
68    ///
69    /// # Panics
70    ///
71    /// Panics when existing relay state refers to an uncommitted source placement.
72    pub(super) async fn finalize_admission(
73        &self,
74        commit: JoinCommit,
75        context: RoomEffectContext<'_>,
76    ) -> CommittedTransportReceipt {
77        let receipt = commit.receipt.clone();
78        RoomEffects::from_join(commit).execute(self, context).await;
79        let session = &receipt.transport_session_key;
80        info!(
81            event = telemetry_event::USER_JOINED,
82            room_id = self.uuid(),
83            user_id = %session.user_id().path_segment(),
84            connection_id = receipt.connection_id.as_u64(),
85            media_worker_id = session.media_worker_id().as_usize(),
86            "user joined room"
87        );
88        receipt
89    }
90
91    /// Returns `true` only when `connection_id` removed the current room user.
92    ///
93    /// # Panics
94    ///
95    /// Panics when detached relay state refers to an uncommitted source placement.
96    pub(crate) async fn remove_user(
97        &self,
98        user_id: &UserId,
99        connection_id: ConnectionId,
100        media_transport: &MediaTransport,
101    ) -> bool {
102        self.remove_user_with_teardown(
103            user_id,
104            connection_id,
105            RoomEffectContext::runtime(media_transport),
106        )
107        .await
108    }
109
110    /// Returns `true` only when `connection_id` was current. A stale committed
111    /// placement may still be retired before returning `false`.
112    ///
113    /// # Panics
114    ///
115    /// Panics when detached relay state refers to an uncommitted source placement.
116    pub async fn remove_user_with_teardown(
117        &self,
118        user_id: &UserId,
119        connection_id: ConnectionId,
120        context: RoomEffectContext<'_>,
121    ) -> bool {
122        let commit = {
123            let mut state = self.state.write().await;
124            state.close_connection(user_id, connection_id)
125        };
126        let removed_current_user = matches!(&commit, Some(ConnectionCloseCommit::Current { .. }));
127        let closed = commit.as_ref().and_then(|commit| match commit {
128            ConnectionCloseCommit::Current {
129                user_id,
130                connection_id,
131                session_teardown,
132                ..
133            } => Some((
134                user_id.clone(),
135                *connection_id,
136                session_teardown
137                    .as_ref()
138                    .map(|teardown| teardown.session_key().media_worker_id()),
139            )),
140            ConnectionCloseCommit::StalePlacement { .. } => None,
141        });
142        if let Some(commit) = commit {
143            RoomEffects::from_connection_close(commit)
144                .execute(self, context)
145                .await;
146        }
147        if let Some((user_id, connection_id, media_worker_id)) = closed {
148            info!(
149                event = telemetry_event::USER_CLOSED,
150                room_id = self.uuid(),
151                user_id = %user_id.path_segment(),
152                connection_id = connection_id.as_u64(),
153                media_worker_id = media_worker_id.map(MediaWorkerId::as_usize),
154                "user closed"
155            );
156        }
157        removed_current_user
158    }
159
160    /// Captures recipients in the state snapshot that validates `connection_id`
161    /// as current for `sender_id`. Missing or stale senders are ignored.
162    ///
163    /// # Errors
164    ///
165    /// Returns [`BroadcastPayloadError`] when the payload exceeds the room
166    /// broadcast byte limit or cannot be measured as serialized JSON.
167    pub(crate) async fn broadcast(
168        &self,
169        sender_id: &UserId,
170        connection_id: ConnectionId,
171        message: serde_json::Value,
172    ) -> Result<(), BroadcastPayloadError> {
173        let fanout = {
174            let state = self.state.read().await;
175            state.broadcast_fanout(sender_id, connection_id, message)
176        }?;
177        if let Some(fanout) = fanout {
178            fanout.emit();
179        }
180        Ok(())
181    }
182
183    pub(crate) async fn has_connection(
184        &self,
185        user_id: &UserId,
186        connection_id: ConnectionId,
187    ) -> bool {
188        self.state.read().await.user_connection_id(user_id) == Some(connection_id)
189    }
190
191    /// Ignores missing or stale connections. Publication transitions remain
192    /// authoritative for camera and screen-sharing presence.
193    pub(crate) async fn update_user_info(
194        &self,
195        user_id: &UserId,
196        connection_id: ConnectionId,
197        media_transport: &MediaTransport,
198        mut info: UserInfo,
199    ) {
200        info.is_camera_on = None;
201        info.is_screen_sharing_on = None;
202        let commit = {
203            let mut state = self.state.write().await;
204            state.apply_presence_update(user_id, connection_id, &info)
205        };
206        if let Some(commit) = commit {
207            RoomEffects::from_presence(commit)
208                .execute(self, RoomEffectContext::runtime(media_transport))
209                .await;
210        } else {
211            warn!(
212                ?user_id,
213                connection_id = ?connection_id,
214                ?info,
215                "user info update was rejected by room state"
216            );
217        }
218    }
219
220    /// Missing users are ignored.
221    ///
222    /// # Panics
223    ///
224    /// Panics if a current room user has no committed router placement or detached
225    /// relay state refers to an uncommitted source placement.
226    pub(crate) async fn disconnect_users(
227        &self,
228        user_ids: &[UserId],
229        media_transport: &MediaTransport,
230    ) {
231        self.disconnect_users_with_teardown(user_ids, RoomEffectContext::runtime(media_transport))
232            .await;
233    }
234
235    /// Removes current sessions in one state commit and ignores missing users.
236    ///
237    /// # Panics
238    ///
239    /// Panics if a current room user has no committed router placement or detached
240    /// relay state refers to an uncommitted source placement.
241    pub async fn disconnect_users_with_teardown(
242        &self,
243        user_ids: &[UserId],
244        context: RoomEffectContext<'_>,
245    ) {
246        let commit = {
247            let mut state = self.state.write().await;
248            state.apply_disconnect_users(user_ids)
249        };
250        let sessions = commit
251            .session_teardowns
252            .iter()
253            .map(|teardown| teardown.session_key().clone())
254            .collect::<Vec<_>>();
255        RoomEffects::from_disconnect(commit)
256            .execute(self, context)
257            .await;
258        for session in sessions {
259            info!(
260                event = telemetry_event::USER_DISCONNECTED,
261                room_id = self.uuid(),
262                user_id = %session.user_id().path_segment(),
263                connection_id = session.connection_id().as_u64(),
264                media_worker_id = session.media_worker_id().as_usize(),
265                "user disconnected"
266            );
267        }
268    }
269
270    #[cfg(any(test, feature = "testing-transport"))]
271    /// Returns `None` for a missing or stale connection. The first accepted
272    /// capabilities commit realizes receiver routes waiting on negotiation.
273    pub async fn apply_session_negotiated(
274        &self,
275        user_id: &UserId,
276        connection_id: ConnectionId,
277        capabilities: MediaCapabilities,
278        media_port: &MediaTransport,
279    ) -> Option<()> {
280        self.user_operation(user_id, connection_id, media_port)
281            .apply_session_negotiated(capabilities, &[])
282            .await
283    }
284
285    #[cfg(test)]
286    pub(super) async fn user_count(&self) -> usize {
287        self.state.read().await.user_count()
288    }
289
290    pub(super) async fn is_empty(&self) -> bool {
291        self.state.read().await.is_empty()
292    }
293}
o_sfu_core/engine/room/membership.rs

o_sfu_core/engine/room/
membership.rs
