o_sfu_core/engine/room/state/membership/
mod.rs

1use std::{
2    collections::{BTreeMap, BTreeSet},
3    mem,
4    sync::Arc,
5};
6
7use o_sfu_router::rtp::MediaCapabilities;
8use tracing::{debug, error, warn};
9
10use super::{
11    super::{
12        BroadcastPayload, BroadcastPayloadError, RoomEventMessage, RoomJoinError, RouterPlacement,
13        UserCloseReason,
14        effects::transport::RoomTransportPlan,
15        media_graph::{SessionPlacementCommit, SessionPlacementRejection},
16        outbound::{MessageFanout, OutboundSender, VersionedRemoteTrackSnapshot, fanout_all},
17    },
18    UserJoinedFanout,
19    shared::{ActiveUser, RoomState},
20};
21#[cfg(test)]
22use crate::engine::MediaWorkerId;
23use crate::engine::{
24    ConnectionId, UserId, UserInfo,
25    media_transport::{TransportSessionKey, TransportTeardown},
26};
27
28#[cfg(test)]
29#[expect(non_snake_case, reason = "test modules map to local TESTS directories")]
30mod TESTS;
31
32#[derive(Debug, Default)]
33pub struct LifecycleEffects {
34    pub close_requests: Vec<UserCloseRequest>,
35    pub fanouts: Vec<MessageFanout>,
36    pub track_snapshots: Vec<(OutboundSender, VersionedRemoteTrackSnapshot)>,
37}
38
39impl LifecycleEffects {
40    fn push_fanout(&mut self, fanout: Option<MessageFanout>) {
41        if let Some(fanout) = fanout {
42            self.fanouts.push(fanout);
43        }
44    }
45
46    fn push_close_request(&mut self, request: Option<UserCloseRequest>) {
47        if let Some(request) = request {
48            self.close_requests.push(request);
49        }
50    }
51}
52
53#[derive(Debug)]
54pub struct UserCloseRequest {
55    pub sender: OutboundSender,
56    pub reason: UserCloseReason,
57}
58
59type RuntimeUserRemoval = (ActiveUser, RoomTransportPlan, usize);
60
61#[derive(Debug)]
62pub struct PresenceCommit {
63    pub fanout: MessageFanout,
64}
65
66#[derive(Debug)]
67pub struct JoinCommit {
68    pub effects: LifecycleEffects,
69    pub transport_session_key: TransportSessionKey,
70    pub transport_plan: RoomTransportPlan,
71}
72
73#[expect(
74    clippy::large_enum_variant,
75    reason = "connection close is cold and boxing the room transport plan would add allocation without simplifying ownership"
76)]
77#[derive(Debug)]
78pub enum ConnectionCloseCommit {
79    Current {
80        session_teardown: Option<TransportTeardown>,
81        effects: LifecycleEffects,
82        transport_plan: RoomTransportPlan,
83        evictions: usize,
84    },
85    StalePlacement {
86        session_teardown: TransportTeardown,
87    },
88}
89
90#[derive(Debug)]
91pub struct DisconnectCommit {
92    pub session_teardowns: Vec<TransportTeardown>,
93    pub effects: LifecycleEffects,
94    pub transport_plan: RoomTransportPlan,
95    pub evictions: usize,
96}
97
98impl RoomState {
99    pub fn fanout_all(&self, message: RoomEventMessage) -> MessageFanout {
100        fanout_all(self.users.values().map(|user| user.sender.clone()), message)
101    }
102
103    pub fn fanout_all_except(
104        &self,
105        message: RoomEventMessage,
106        excluded_user_id: &UserId,
107    ) -> MessageFanout {
108        fanout_all(
109            self.users
110                .iter()
111                .filter(|(user_id, _session)| excluded_user_id != *user_id)
112                .map(|(_user_id, user)| user.sender.clone()),
113            message,
114        )
115    }
116
117    fn apply_join_routing(
118        &mut self,
119        user_id: &UserId,
120        connection_id: ConnectionId,
121        previous_connection: Option<ConnectionId>,
122        home_placement: RouterPlacement,
123    ) -> Result<SessionPlacementCommit, RoomJoinError> {
124        self.topology
125            .commit_session_placement(user_id, connection_id, previous_connection, home_placement)
126            .map_err(|rejection| {
127                match rejection {
128                    SessionPlacementRejection::MissingPreviousSession {
129                        previous_connection,
130                    } => {
131                        error!(
132                            ?user_id,
133                            connection_id = ?previous_connection,
134                            "missing committed routing session for replacement join"
135                        );
136                    }
137                    SessionPlacementRejection::Router(error) => {
138                        error!(
139                            ?user_id,
140                            ?error,
141                            "failed to mirror user join into room router"
142                        );
143                    }
144                }
145                RoomJoinError::RouterState
146            })
147    }
148
149    #[cfg(test)]
150    fn fallback_join_placement(&self) -> RouterPlacement {
151        RouterPlacement {
152            router: self.topology.router().placement_snapshot().primary(),
153            media_worker: MediaWorkerId::from_raw(0),
154        }
155    }
156
157    fn install_joined_session(
158        &mut self,
159        user_id: &UserId,
160        sender: OutboundSender,
161        connection_id: ConnectionId,
162    ) -> Option<OutboundSender> {
163        if let Some(user) = self.users.get_mut(user_id) {
164            let old_sender = mem::replace(&mut user.sender, sender);
165            user.reset_presentation();
166            user.parsed_client_rtp_capabilities = None;
167            user.connection_id = connection_id;
168            user.video_soft_pause_deadline = None;
169            return Some(old_sender);
170        }
171        self.users.insert(
172            user_id.clone(),
173            ActiveUser {
174                user_id: Arc::new(user_id.clone()),
175                info: UserInfo::default(),
176                server_featured: None,
177                parsed_client_rtp_capabilities: None,
178                connection_id,
179                video_soft_pause_deadline: None,
180                sender,
181            },
182        );
183        None
184    }
185
186    #[cfg(test)]
187    pub fn apply_join(
188        &mut self,
189        user_id: &UserId,
190        sender: OutboundSender,
191    ) -> Result<JoinCommit, RoomJoinError> {
192        self.apply_join_on_placement(
193            user_id,
194            sender,
195            UserJoinedFanout::Suppress,
196            self.fallback_join_placement(),
197        )
198    }
199
200    pub fn apply_join_on_placement(
201        &mut self,
202        user_id: &UserId,
203        sender: OutboundSender,
204        joined_fanout: UserJoinedFanout,
205        home_placement: RouterPlacement,
206    ) -> Result<JoinCommit, RoomJoinError> {
207        let previous_connection = self.users.get(user_id).map(|user| user.connection_id);
208        let is_new = previous_connection.is_none();
209        if is_new && self.users.len() >= self.admission_policy.max_sessions {
210            return Err(RoomJoinError::RoomFull);
211        }
212        let connection_id = ConnectionId::allocate(&mut self.next_connection_id);
213        let mut source_recipients = if previous_connection.is_some() {
214            self.topology
215                .committed_consumer_user_ids_for_owner_sources(user_id)
216        } else {
217            BTreeSet::new()
218        };
219        source_recipients.remove(user_id);
220        let placement =
221            self.apply_join_routing(user_id, connection_id, previous_connection, home_placement)?;
222        let transport_session_key = placement.transport_session_key;
223        let mut transport_plan = placement.replacement_transport_plan;
224        if let Some(previous_connection) = previous_connection {
225            transport_plan.extend_teardown(
226                self.staged_publishes
227                    .take_teardowns_for_connection(user_id, previous_connection),
228            );
229        }
230
231        let previous_sender = self.install_joined_session(user_id, sender, connection_id);
232        let had_previous_sender = previous_sender.is_some();
233
234        let mut effects = LifecycleEffects::default();
235        effects.push_close_request(previous_sender.map(|sender| UserCloseRequest {
236            sender,
237            reason: UserCloseReason::Replaced,
238        }));
239        effects
240            .track_snapshots
241            .extend(self.remote_track_snapshots_for_users(source_recipients, true));
242        effects.push_fanout(had_previous_sender.then(|| {
243            self.fanout_all_except(
244                RoomEventMessage::UserDeparted {
245                    user_id: user_id.clone(),
246                },
247                user_id,
248            )
249        }));
250        effects.push_fanout(if joined_fanout == UserJoinedFanout::Emit {
251            self.user_info_snapshot(user_id)
252                .map(|(joined_user_id, info)| {
253                    self.fanout_all_except(
254                        RoomEventMessage::UserJoined {
255                            user_id: joined_user_id,
256                            info,
257                        },
258                        user_id,
259                    )
260                })
261        } else {
262            None
263        });
264        Ok(JoinCommit {
265            effects,
266            transport_session_key,
267            transport_plan,
268        })
269    }
270
271    fn remove_runtime_user(&mut self, user_id: &UserId) -> Option<RuntimeUserRemoval> {
272        let user = self.users.remove(user_id)?;
273        let (mut transport_plan, evictions) = self.topology.remove_session(user_id);
274        transport_plan.extend_teardown(
275            self.staged_publishes
276                .take_teardowns_for_connection(user_id, user.connection_id),
277        );
278        Some((user, transport_plan, evictions))
279    }
280
281    pub fn close_connection(
282        &mut self,
283        user_id: &UserId,
284        connection_id: ConnectionId,
285    ) -> Option<ConnectionCloseCommit> {
286        if self
287            .users
288            .get(user_id)
289            .is_none_or(|user| user.connection_id != connection_id)
290        {
291            let session_key = self
292                .topology
293                .retire_committed_placement(user_id, connection_id)?;
294            return Some(ConnectionCloseCommit::StalePlacement {
295                session_teardown: TransportTeardown::CloseSession { session_key },
296            });
297        }
298        let session_teardown = self
299            .committed_transport_user_key(user_id, connection_id)
300            .map(|session_key| TransportTeardown::CloseSession { session_key });
301        let mut source_recipients = self
302            .topology
303            .committed_consumer_user_ids_for_owner_sources(user_id);
304        source_recipients.remove(user_id);
305        let (user, transport_plan, evictions) = self.remove_runtime_user(user_id)?;
306        Some(ConnectionCloseCommit::Current {
307            session_teardown,
308            effects: LifecycleEffects {
309                close_requests: vec![UserCloseRequest {
310                    sender: user.sender,
311                    reason: UserCloseReason::RemovedByRuntime,
312                }],
313                fanouts: vec![self.fanout_all(RoomEventMessage::UserDeparted {
314                    user_id: user_id.clone(),
315                })],
316                track_snapshots: self.remote_track_snapshots_for_users(source_recipients, true),
317            },
318            transport_plan,
319            evictions,
320        })
321    }
322
323    pub fn apply_presence_update(
324        &mut self,
325        user_id: &UserId,
326        connection_id: ConnectionId,
327        info: &UserInfo,
328    ) -> Option<PresenceCommit> {
329        let Some(current_user) = self.users.get_mut(user_id) else {
330            warn!(
331                ?user_id,
332                connection_id = ?connection_id,
333                ?info,
334                "discarding user presence update because the user is missing"
335            );
336            return None;
337        };
338        if current_user.connection_id != connection_id {
339            warn!(
340                ?user_id,
341                connection_id = ?connection_id,
342                current_connection_id = ?current_user.connection_id,
343                ?info,
344                "discarding user presence update because the connection is stale"
345            );
346            return None;
347        }
348        current_user.apply_info_update(info);
349        let snapshot = BTreeMap::from([self.user_info_snapshot(user_id)?]);
350        debug!(
351            ?user_id,
352            connection_id = ?connection_id,
353            ?info,
354            snapshot_len = snapshot.len(),
355            "applied user presence update and staged user info fanout"
356        );
357        Some(PresenceCommit {
358            fanout: self.fanout_all(RoomEventMessage::UserInfoChanged(snapshot)),
359        })
360    }
361
362    pub fn set_user_negotiated(
363        &mut self,
364        user_id: &UserId,
365        connection_id: ConnectionId,
366        capabilities: MediaCapabilities,
367    ) -> Option<bool> {
368        let user = self.user_mut_for_connection(user_id, connection_id)?;
369        let became_ready = user.parsed_client_rtp_capabilities.is_none();
370        user.parsed_client_rtp_capabilities = Some(capabilities);
371        Some(became_ready)
372    }
373
374    pub fn apply_disconnect_users(&mut self, user_ids: &[UserId]) -> DisconnectCommit {
375        let mut source_recipients = BTreeSet::new();
376        for user_id in user_ids {
377            source_recipients.extend(
378                self.topology
379                    .committed_consumer_user_ids_for_owner_sources(user_id),
380            );
381        }
382        for user_id in user_ids {
383            source_recipients.remove(user_id);
384        }
385        let mut close_requests = Vec::new();
386        let mut session_teardowns = Vec::new();
387        let mut fanouts = Vec::new();
388        let mut transport_plan = RoomTransportPlan::default();
389        let mut evictions = 0;
390        for user_id in user_ids {
391            let Some(connection_id) = self.users.get(user_id).map(|user| user.connection_id) else {
392                continue;
393            };
394            let session_teardown = TransportTeardown::CloseSession {
395                session_key: self.transport_user_key(user_id, connection_id),
396            };
397            let Some((user, user_transport_plan, user_evictions)) =
398                self.remove_runtime_user(user_id)
399            else {
400                continue;
401            };
402            transport_plan.extend(user_transport_plan);
403            evictions += user_evictions;
404            session_teardowns.push(session_teardown);
405            close_requests.push(UserCloseRequest {
406                sender: user.sender,
407                reason: UserCloseReason::RemovedByRuntime,
408            });
409            fanouts.push(self.fanout_all(RoomEventMessage::UserDeparted {
410                user_id: user_id.clone(),
411            }));
412        }
413        DisconnectCommit {
414            session_teardowns,
415            effects: LifecycleEffects {
416                close_requests,
417                fanouts,
418                track_snapshots: self.remote_track_snapshots_for_users(source_recipients, true),
419            },
420            transport_plan,
421            evictions,
422        }
423    }
424
425    /// Plans a broadcast to every current user except `user_id`.
426    ///
427    /// Returns `Ok(None)` when `connection_id` is missing or stale for
428    /// `user_id`.
429    ///
430    /// # Errors
431    ///
432    /// Returns [`BroadcastPayloadError::TooLarge`] when the serialized payload
433    /// exceeds the room broadcast limit. Returns
434    /// [`BroadcastPayloadError::JsonSerialization`] when JSON serialization
435    /// fails.
436    pub fn broadcast_fanout(
437        &self,
438        user_id: &UserId,
439        connection_id: ConnectionId,
440        message: serde_json::Value,
441    ) -> Result<Option<MessageFanout>, BroadcastPayloadError> {
442        if self.user_for_connection(user_id, connection_id).is_none() {
443            return Ok(None);
444        }
445        let message = BroadcastPayload::try_new(message)?;
446        Ok(Some(self.fanout_all_except(
447            RoomEventMessage::Broadcast {
448                sender_id: user_id.clone(),
449                message,
450            },
451            user_id,
452        )))
453    }
454}
o_sfu_core/engine/room/state/membership/mod.rs

o_sfu_core/engine/room/state/membership/
mod.rs
