o_sfu_core/engine/room/media_graph/
topology.rs

1//! Room publications and subscriptions share transport-placement authority.
2
3use std::{
4    collections::{BTreeMap, BTreeSet},
5    sync::Arc,
6};
7
8use o_sfu_router::{
9    MediaKind, ProducerId, Router, RouterError,
10    rtp::{MediaCapabilities, MediaStream as RouterRtpParameters},
11};
12use tracing::{error, warn};
13
14use super::{
15    CommittedConsumerSetup, ConsumerId, ConsumerRouteView, ConsumerSetupTarget,
16    DeclaredConsumerSetup, PendingConsumerRouteView, PendingConsumerSetup, PublishedSource,
17    ReceiverRouteActivity, SubscriptionKey, ValidatedPublish,
18    producer::{PublicationCommitError, allocate_source_descriptor},
19    route_graph::{CurrentPublication, PendingUpgrade, RemovedRoutes, RouteGraph},
20    source_index::PublishedSources,
21};
22use crate::engine::{
23    ConnectionId, MediaWorkerId, RoomInstanceId, UserId,
24    media_transport::{
25        SessionUploadEncoding, SourceActivityRevision, SourceActivityUpdate,
26        TransportConsumerRoute, TransportMediaId, TransportRelayRouteEffect, TransportSessionKey,
27        TransportSourceActivityEffect, TransportSourceKey, TransportTeardown,
28    },
29    room::{
30        RoomMediaCounts, RoomRuntimeContext, RouterPlacement,
31        effects::transport::RoomTransportPlan, outbound::OutboundSender,
32    },
33    source_model::{
34        ActiveSpeakerSourceRole, ConsumerSourceSelection, PolicyPauseReason,
35        PublishedSourceDescriptor, PublishedSourceId, SourceSubscriptionIntent, UserStreamId,
36    },
37};
38
39#[cfg(test)]
40#[path = "TESTS/topology_support.rs"]
41mod topology_support;
42
43/// Keeps source records and route realization beside [`Router`] so transport
44/// effects resolve from the same committed placement.
45///
46/// Transport-facing mutations return resolved work for execution after the room
47/// state lock is released.
48#[derive(Debug)]
49pub struct RoomTopology {
50    instance: RoomInstanceId,
51    sources: PublishedSources,
52    route_graph: RouteGraph,
53    router: Router,
54    next_producer_id: u64,
55    // Revision of committed video allocation inputs across transport awaits.
56    video_allocation_revision: u64,
57}
58
59/// The new placement is authoritative before displaced-session cleanup is returned.
60///
61/// The key snapshots the committed connection and worker-resolved transport identity
62/// across the room-state lock boundary. Placement lifetime remains controlled by
63/// [`RoomTopology::commit_session_placement`], [`RoomTopology::remove_session`]
64/// or [`RoomTopology::retire_committed_placement`]. Bundling the key with resolved
65/// cleanup lets membership release `room.state` without another placement lookup.
66#[derive(Debug)]
67pub struct SessionPlacementCommit {
68    pub transport_session_key: TransportSessionKey,
69    /// Empty for a first placement.
70    pub replacement_transport_plan: RoomTransportPlan,
71}
72
73#[derive(Debug)]
74pub(in crate::engine::room) struct ConsumerActivityCommit {
75    pub(super) update: Option<ReceiverRouteActivity>,
76    pub(super) relay_effects: Vec<TransportRelayRouteEffect>,
77    pub(super) evictions: usize,
78}
79
80#[derive(Debug)]
81pub enum SessionPlacementRejection {
82    MissingPreviousSession { previous_connection: ConnectionId },
83    Router(RouterError),
84}
85
86impl RoomTopology {
87    pub fn new(
88        runtime_context: &RoomRuntimeContext,
89        router_rtp_capabilities: MediaCapabilities,
90        pending_target_limit: usize,
91    ) -> Self {
92        let router = match runtime_context.initial_router_placements() {
93            Some(placements) => {
94                Router::with_placements(placements.clone(), router_rtp_capabilities)
95            }
96            None => Router::new(runtime_context.primary_router(), router_rtp_capabilities),
97        };
98        Self {
99            instance: runtime_context.instance(),
100            sources: PublishedSources::default(),
101            route_graph: RouteGraph::new(pending_target_limit),
102            router,
103            next_producer_id: 1,
104            video_allocation_revision: 0,
105        }
106    }
107
108    pub(in crate::engine::room) const fn video_allocation_revision(&self) -> u64 {
109        self.video_allocation_revision
110    }
111
112    fn invalidate_video_allocation(&mut self) {
113        self.video_allocation_revision = self.video_allocation_revision.wrapping_add(1);
114    }
115
116    pub(in crate::engine::room) fn router(&self) -> &Router {
117        &self.router
118    }
119
120    /// Returns `None` unless the exact user connection remains committed.
121    #[must_use]
122    pub fn committed_transport_user_key(
123        &self,
124        user_id: impl Into<Arc<UserId>>,
125        connection_id: ConnectionId,
126    ) -> Option<TransportSessionKey> {
127        let user_id = user_id.into();
128        let worker = self
129            .router
130            .committed_media_worker_id(user_id.as_ref(), connection_id)?;
131        Some(self.transport_session_key(user_id, connection_id, worker))
132    }
133
134    /// Requires the exact user connection to remain committed.
135    ///
136    /// Use [`Self::committed_transport_user_key`] for stale callbacks or teardown
137    /// races where the placement may already be retired.
138    ///
139    /// # Panics
140    ///
141    /// Panics when no committed router placement exists.
142    #[must_use]
143    #[expect(
144        clippy::unreachable,
145        reason = "current room operations require committed connection placement and must not synthesize a transport worker"
146    )]
147    pub fn transport_user_key(
148        &self,
149        user_id: impl Into<Arc<UserId>>,
150        connection_id: ConnectionId,
151    ) -> TransportSessionKey {
152        let user_id = user_id.into();
153        let Some(worker) = self
154            .router
155            .committed_media_worker_id(user_id.as_ref(), connection_id)
156        else {
157            unreachable!("transport session key lookup requires committed connection placement");
158        };
159        self.transport_session_key(user_id, connection_id, worker)
160    }
161
162    fn transport_session_key(
163        &self,
164        user_id: Arc<UserId>,
165        connection_id: ConnectionId,
166        media_worker_id: MediaWorkerId,
167    ) -> TransportSessionKey {
168        TransportSessionKey::new(self.instance, media_worker_id, connection_id, user_id)
169    }
170
171    /// Returns `None` when `connection_id` is not the user's committed placement.
172    pub fn retire_committed_placement(
173        &mut self,
174        user_id: &UserId,
175        connection_id: ConnectionId,
176    ) -> Option<TransportSessionKey> {
177        let media_worker = self
178            .router
179            .retire_committed_placement(user_id, connection_id)?;
180        Some(self.transport_session_key(user_id.clone().into(), connection_id, media_worker))
181    }
182
183    /// Counts each logical subscription once while pending or committed.
184    #[must_use]
185    pub fn media_counts(&self) -> RoomMediaCounts {
186        RoomMediaCounts {
187            publications: self.sources.publication_count(),
188            subscriptions: self.route_graph.subscription_count(),
189        }
190    }
191
192    /// Counts committed consumers, including inactive selections.
193    pub(in crate::engine::room) fn consumer_count(&self) -> usize {
194        self.route_graph.count()
195    }
196
197    #[cfg(any(test, feature = "testing-transport"))]
198    pub(in crate::engine::room) fn first_published_transport_media_id(
199        &self,
200    ) -> Option<TransportMediaId> {
201        self.sources.first_transport_media_id()
202    }
203
204    #[cfg(any(test, feature = "testing-transport"))]
205    pub(in crate::engine::room) fn producer_transport_media_id(
206        &self,
207        user_id: &UserId,
208        connection_id: ConnectionId,
209        stream_id: &UserStreamId,
210    ) -> Option<TransportMediaId> {
211        self.sources
212            .transport_media_id(user_id, connection_id, stream_id)
213    }
214
215    #[must_use]
216    pub(in crate::engine::room) fn source_descriptor(
217        &self,
218        source_id: PublishedSourceId,
219    ) -> Option<&PublishedSourceDescriptor> {
220        self.sources
221            .source(source_id)
222            .map(|source| &source.descriptor)
223    }
224
225    #[must_use]
226    pub(in crate::engine::room) fn source_for_transport_media(
227        &self,
228        transport_media_id: TransportMediaId,
229    ) -> Option<&PublishedSource> {
230        self.sources.source_for_transport(transport_media_id)
231    }
232
233    #[must_use]
234    pub(in crate::engine::room) fn source_id_for_owner_stream(
235        &self,
236        owner_user_id: &UserId,
237        stream_id: &UserStreamId,
238    ) -> Option<PublishedSourceId> {
239        self.sources.id_for_owner_stream(owner_user_id, stream_id)
240    }
241
242    /// Returns the source ID only when owner, connection and stream remain current.
243    #[must_use]
244    pub(in crate::engine::room) fn published_source_id(
245        &self,
246        owner: &UserId,
247        connection: ConnectionId,
248        stream_id: &UserStreamId,
249    ) -> Option<PublishedSourceId> {
250        let id = self.sources.id_for_owner_stream(owner, stream_id)?;
251        let source = self.sources.source(id)?;
252        (source.transport.session_key().connection_id() == connection).then_some(id)
253    }
254
255    /// Iterates committed sources in source-ID order.
256    pub(in crate::engine::room) fn published_sources(
257        &self,
258    ) -> impl Iterator<Item = &PublishedSource> {
259        self.sources.iter()
260    }
261
262    /// Counts distinct users with active publications for each logical stream.
263    pub(in crate::engine::room) fn active_stream_user_counts(&self) -> BTreeMap<UserStreamId, u64> {
264        let mut users_by_stream: BTreeMap<UserStreamId, BTreeSet<UserId>> = BTreeMap::new();
265        for source in self.sources.iter().filter(|source| source.active) {
266            users_by_stream
267                .entry(source.descriptor.stream_id().clone())
268                .or_default()
269                .insert(source.descriptor.owner().user_id().clone());
270        }
271        users_by_stream
272            .into_iter()
273            .map(|(stream_id, users)| (stream_id, u64::try_from(users.len()).unwrap_or(u64::MAX)))
274            .collect()
275    }
276
277    /// Returns a detector owner with an active promotable source in the same group.
278    #[must_use]
279    pub(in crate::engine::room) fn active_speaker_detector_owner(
280        &self,
281        transport_media_id: TransportMediaId,
282    ) -> Option<UserId> {
283        let source = self.source_for_transport_media(transport_media_id)?;
284        let detector_policy = source.descriptor.policy().active_speaker()?;
285        if detector_policy.role() != ActiveSpeakerSourceRole::Detector {
286            return None;
287        }
288        let owner = source.descriptor.owner().user_id();
289        self.sources
290            .owner_has_promotable_source_in_group(owner, detector_policy.group())
291            .then(|| owner.clone())
292    }
293
294    pub(in crate::engine::room) fn committed_consumer_routes(
295        &self,
296    ) -> impl Iterator<Item = ConsumerRouteView<'_>> {
297        self.route_graph
298            .attached()
299            .filter_map(|(key, current)| self.consumer_route(key, current))
300    }
301
302    pub(in crate::engine::room) fn committed_consumer_routes_for_user(
303        &self,
304        user_id: &UserId,
305    ) -> impl Iterator<Item = ConsumerRouteView<'_>> {
306        self.route_graph
307            .attached_for_receiver(user_id)
308            .filter_map(|(key, current)| self.consumer_route(key, current))
309    }
310
311    pub(super) fn detach_declined_consumers(
312        &mut self,
313        session: &TransportSessionKey,
314        declined: &[TransportMediaId],
315    ) -> (Vec<TransportRelayRouteEffect>, Vec<TransportTeardown>, bool) {
316        let RemovedRoutes {
317            routes,
318            consumers,
319            relays,
320        } = self
321            .route_graph
322            .detach_declined_consumers(session, declined);
323        let detached = !routes.is_empty();
324        if detached {
325            self.invalidate_video_allocation();
326        }
327        for consumer in consumers {
328            if let Some(error) = self.router.remove_consumer(consumer).err() {
329                error!(?consumer, ?error, "failed to remove declined room consumer");
330            }
331        }
332        let teardown = Self::media_teardowns([], routes).collect();
333        (relays, teardown, detached)
334    }
335
336    pub(in crate::engine::room) fn pending_consumer_routes_for_user(
337        &self,
338        user_id: &UserId,
339    ) -> impl Iterator<Item = PendingConsumerRouteView<'_>> {
340        self.route_graph
341            .attached_for_receiver(user_id)
342            .filter(|(_, current)| current.is_pending())
343            .filter_map(|(_, current)| {
344                let source = self.sources.source(current.source_id)?;
345                Some(PendingConsumerRouteView {
346                    source,
347                    selection: current.selection,
348                })
349            })
350    }
351
352    /// Returns `None` when the attached source differs from `source_id`.
353    #[must_use]
354    pub(in crate::engine::room) fn consumer_source_selection(
355        &self,
356        key: &SubscriptionKey,
357        source_id: PublishedSourceId,
358    ) -> Option<ConsumerSourceSelection> {
359        self.route_graph.selection(key, source_id)
360    }
361
362    /// Returns merged receiver intent or the default for a missing subscription.
363    #[must_use]
364    pub(in crate::engine::room) fn subscription_intent(
365        &self,
366        key: &SubscriptionKey,
367    ) -> SourceSubscriptionIntent {
368        self.route_graph.intent(key)
369    }
370
371    pub(in crate::engine::room) fn committed_consumer_user_ids_for_source(
372        &self,
373        source_id: PublishedSourceId,
374    ) -> BTreeSet<UserId> {
375        self.route_graph
376            .attached_for_source(source_id)
377            .filter(|(_, current)| current.committed().is_some())
378            .map(|(key, _)| key.receiver.clone())
379            .collect()
380    }
381
382    pub(in crate::engine::room) fn committed_consumer_user_ids_for_owner_sources(
383        &self,
384        user_id: &UserId,
385    ) -> BTreeSet<UserId> {
386        let route_graph = &self.route_graph;
387        self.sources
388            .ids_for_owner(user_id)
389            .flat_map(|source_id| route_graph.attached_for_source(source_id))
390            .filter(|(_, current)| current.committed().is_some())
391            .map(|(key, _)| key.receiver.clone())
392            .collect()
393    }
394
395    #[must_use]
396    pub(in crate::engine::room) fn committed_consumer_route_for_key(
397        &self,
398        key: &SubscriptionKey,
399    ) -> Option<ConsumerRouteView<'_>> {
400        let (key, current) = self.route_graph.current(key)?;
401        self.consumer_route(key, current)
402    }
403
404    #[must_use]
405    pub(in crate::engine::room) fn published_source(
406        &self,
407        source_id: PublishedSourceId,
408    ) -> Option<&PublishedSource> {
409        self.sources.source(source_id)
410    }
411
412    /// Attaches `source_id` to each eligible receiver lacking a route realization.
413    pub(super) fn missing_consumer_targets_for_source<'a>(
414        &mut self,
415        source_id: PublishedSourceId,
416        receivers: impl IntoIterator<Item = (&'a UserId, ConnectionId)>,
417    ) -> Vec<ConsumerSetupTarget> {
418        receivers
419            .into_iter()
420            .filter_map(|(user, connection)| self.consumer_target(user, connection, source_id))
421            .collect()
422    }
423
424    fn consumer_target(
425        &mut self,
426        user_id: &UserId,
427        connection_id: ConnectionId,
428        source_id: PublishedSourceId,
429    ) -> Option<ConsumerSetupTarget> {
430        let consumer_session = self.committed_transport_user_key(user_id.clone(), connection_id)?;
431        let (sources, route_graph) = (&self.sources, &mut self.route_graph);
432        Self::attach_consumer_target(route_graph, &consumer_session, sources.source(source_id)?)
433    }
434
435    /// Skips self-consumption and attaches only an absent realization.
436    fn attach_consumer_target(
437        route_graph: &mut RouteGraph,
438        consumer_session: &TransportSessionKey,
439        source: &PublishedSource,
440    ) -> Option<ConsumerSetupTarget> {
441        if source.descriptor.owner().user_id() == consumer_session.user_id() {
442            return None;
443        }
444        let target = ConsumerSetupTarget::new(consumer_session.clone(), source);
445        route_graph
446            .attach_for_setup(target.subscription_key(), target.source_id)
447            .then_some(target)
448    }
449
450    /// Attaches each matching source with no realization to the exact committed receiver.
451    ///
452    /// Returns an empty list when `connection_id` is stale.
453    pub(super) fn missing_consumer_targets(
454        &mut self,
455        user_id: &UserId,
456        connection_id: ConnectionId,
457        include_source: impl Fn(&PublishedSource) -> bool,
458    ) -> Vec<ConsumerSetupTarget> {
459        let Some(consumer_session) =
460            self.committed_transport_user_key(user_id.clone(), connection_id)
461        else {
462            return Vec::new();
463        };
464        let (sources, route_graph) = (&self.sources, &mut self.route_graph);
465        sources
466            .iter()
467            .filter(|source| include_source(source))
468            .filter_map(|source| {
469                Self::attach_consumer_target(route_graph, &consumer_session, source)
470            })
471            .collect()
472    }
473
474    /// Updates selection while subscription, source and exact route still match.
475    ///
476    /// Returns `false` when async transport work refers to a displaced route.
477    pub fn update_consumer_source_selection(
478        &mut self,
479        key: &SubscriptionKey,
480        source_id: PublishedSourceId,
481        route: &TransportConsumerRoute,
482        update: impl FnOnce(&mut ConsumerSourceSelection),
483    ) -> bool {
484        let updated = self
485            .route_graph
486            .update_selection(key, source_id, route, update);
487        if updated {
488            self.invalidate_video_allocation();
489        }
490        updated
491    }
492
493    /// Commits a hold only for the active source, subscription and exact route.
494    pub(in crate::engine::room) fn update_consumer_upgrade(
495        &mut self,
496        key: &SubscriptionKey,
497        source_id: PublishedSourceId,
498        route: &TransportConsumerRoute,
499        pending_upgrade: Option<PendingUpgrade>,
500    ) -> bool {
501        if !self
502            .sources
503            .source(source_id)
504            .is_some_and(|source| source.active)
505        {
506            return false;
507        }
508        self.route_graph
509            .update_upgrade(key, source_id, route, pending_upgrade)
510    }
511
512    fn detach_user_sources(
513        &mut self,
514        user_id: &UserId,
515    ) -> (Vec<TransportSourceKey>, RemovedRoutes) {
516        let mut sources = Vec::new();
517        let mut removed = RemovedRoutes::default();
518        let source_ids = self.sources.ids_for_owner(user_id).collect::<Vec<_>>();
519        for source_id in source_ids {
520            if let Some((source, routes)) = self.remove_source(source_id) {
521                sources.push(source.transport);
522                removed.extend(routes);
523            }
524        }
525        (sources, removed)
526    }
527
528    fn remove_source(
529        &mut self,
530        source_id: PublishedSourceId,
531    ) -> Option<(PublishedSource, RemovedRoutes)> {
532        let source = self.sources.remove(source_id)?;
533        let routes = self.route_graph.detach_source(source_id);
534        self.invalidate_video_allocation();
535        Some((source, routes))
536    }
537
538    fn consumer_route<'a>(
539        &'a self,
540        key: &'a SubscriptionKey,
541        current: &'a CurrentPublication,
542    ) -> Option<ConsumerRouteView<'a>> {
543        let committed = current.committed()?;
544        let source = self.sources.source(current.source_id)?;
545        Some(ConsumerRouteView {
546            key,
547            route: &committed.route,
548            mid: &committed.mid,
549            pending_upgrade: committed.pending_upgrade.as_ref(),
550            source,
551            selection: current.selection,
552        })
553    }
554
555    /// Commits a negotiated source only after its router producer succeeds.
556    ///
557    /// # Errors
558    ///
559    /// Returns [`PublicationCommitError::Source`] when descriptor allocation or
560    /// validation fails. Returns [`PublicationCommitError::Router`] when the
561    /// router rejects the producer dependency.
562    pub(in crate::engine::room) fn commit_publication(
563        &mut self,
564        publish: ValidatedPublish,
565        rtp: RouterRtpParameters,
566        encodings: &[SessionUploadEncoding],
567        media: TransportMediaId,
568    ) -> Result<PublishedSourceId, PublicationCommitError> {
569        let descriptor = allocate_source_descriptor(&mut self.sources, &publish, &rtp, encodings)?;
570        let source_id = descriptor.source_id();
571        let producer_id = ProducerId::allocate(&mut self.next_producer_id);
572        let routed = self
573            .router
574            .add_producer(publish.session_key.user_id(), producer_id)?;
575        self.sources.insert(PublishedSource {
576            descriptor,
577            transport: TransportSourceKey::new(publish.session_key, media),
578            rtp,
579            routed,
580            active: true,
581            activity_revision: SourceActivityRevision::default(),
582        });
583        Ok(source_id)
584    }
585
586    fn media_teardowns(
587        sources: impl IntoIterator<Item = TransportSourceKey>,
588        routes: impl IntoIterator<Item = TransportConsumerRoute>,
589    ) -> impl Iterator<Item = TransportTeardown> {
590        sources
591            .into_iter()
592            .map(|source| TransportTeardown::RemoveMedia {
593                session_key: source.session_key().clone(),
594                transport_media_id: source.transport_media_id(),
595            })
596            .chain(
597                routes
598                    .into_iter()
599                    .map(|route| TransportTeardown::RemoveMedia {
600                        session_key: route.consumer_session_key().clone(),
601                        transport_media_id: route.consumer_transport_media_id(),
602                    }),
603            )
604    }
605
606    /// Advances the revision only when the exact source connection changes activity.
607    ///
608    /// Returns `None` when the source is missing, the connection is stale or the
609    /// requested activity already matches.
610    pub fn set_published_source_activity(
611        &mut self,
612        source_id: PublishedSourceId,
613        connection_id: ConnectionId,
614        active: bool,
615    ) -> Option<SourceActivityRevision> {
616        let source = self.sources.source_mut(source_id)?;
617        if source.transport.session_key().connection_id() != connection_id
618            || source.active == active
619        {
620            return None;
621        }
622        source.active = active;
623        source.activity_revision = source.activity_revision.next();
624        let revision = source.activity_revision;
625        if !active {
626            self.route_graph.clear_source_upgrades(source_id);
627        }
628        self.invalidate_video_allocation();
629        Some(revision)
630    }
631
632    pub(super) fn source_activity_effects(
633        &self,
634        source: &TransportSourceKey,
635        update: SourceActivityUpdate,
636    ) -> Vec<TransportSourceActivityEffect> {
637        self.route_graph
638            .source_activity_target_workers(source)
639            .map(|target_media_worker_id| TransportSourceActivityEffect {
640                source: source.clone(),
641                target_media_worker_id,
642                update,
643            })
644            .collect()
645    }
646
647    /// Commits the new placement before returning cleanup for `previous_connection`.
648    ///
649    /// Replacement joins must pass the currently committed `previous_connection`.
650    ///
651    /// # Errors
652    ///
653    /// Returns [`SessionPlacementRejection::MissingPreviousSession`] when the
654    /// expected replacement target is no longer committed. Returns
655    /// [`SessionPlacementRejection::Router`] when the router rejects the new
656    /// connection or placement.
657    pub fn commit_session_placement(
658        &mut self,
659        user_id: &UserId,
660        connection_id: ConnectionId,
661        previous_connection: Option<ConnectionId>,
662        home_placement: RouterPlacement,
663    ) -> Result<SessionPlacementCommit, SessionPlacementRejection> {
664        // Preserve the displaced key before the router replaces its user mapping.
665        let previous_session_key = if let Some(previous_connection) = previous_connection {
666            let Some(key) = self.committed_transport_user_key(user_id.clone(), previous_connection)
667            else {
668                return Err(SessionPlacementRejection::MissingPreviousSession {
669                    previous_connection,
670                });
671            };
672            Some(key)
673        } else {
674            None
675        };
676        let media_worker = self
677            .router
678            .commit_session_placement(user_id, connection_id, home_placement)
679            .map_err(SessionPlacementRejection::Router)?;
680        self.route_graph.publisher_joined(user_id);
681        let transport_session_key =
682            self.transport_session_key(user_id.clone().into(), connection_id, media_worker);
683        let replacement_transport_plan = previous_session_key.as_ref().map_or_else(
684            RoomTransportPlan::default,
685            |replaced_session_key| {
686                let close_session = TransportTeardown::CloseSession {
687                    session_key: replaced_session_key.clone(),
688                };
689                // Session close removes source media while subscriber routes need
690                // explicit teardown.
691                let (_, removed_sources) = self.detach_user_sources(user_id);
692                let receiver_relays = self.route_graph.reset_receiver_for_replacement(user_id);
693                let RemovedRoutes {
694                    routes, mut relays, ..
695                } = removed_sources;
696                relays.extend(receiver_relays);
697                let teardown = Self::media_teardowns([], routes).chain([close_session]);
698                RoomTransportPlan::from_relays_and_teardown(relays, teardown)
699            },
700        );
701        if previous_session_key.is_some() {
702            self.invalidate_video_allocation();
703        }
704        Ok(SessionPlacementCommit {
705            transport_session_key,
706            replacement_transport_plan,
707        })
708    }
709
710    /// Returns the cleanup plan even if router removal fails.
711    pub fn remove_session(&mut self, user_id: &UserId) -> (RoomTransportPlan, usize) {
712        let (sources, mut removed) = self.detach_user_sources(user_id);
713        removed.extend(self.route_graph.remove_receiver(user_id));
714        let evictions = self.route_graph.publisher_left(user_id);
715        self.invalidate_video_allocation();
716        let teardown = Self::media_teardowns(sources, removed.routes);
717        if let Some(error) = self.router.remove_session(user_id).err() {
718            error!(?user_id, ?error, "failed to remove user from room router");
719        }
720        (
721            RoomTransportPlan::from_relays_and_teardown(removed.relays, teardown),
722            evictions,
723        )
724    }
725
726    /// Selects MID from the transport declaration, negotiated RTP MID then the
727    /// consumer identity.
728    ///
729    /// # Errors
730    ///
731    /// Returns the declared route and relay release effects when the reservation
732    /// is stale or the router rejects the consumer dependency.
733    pub(super) fn commit_consumer_setup(
734        &mut self,
735        setup: DeclaredConsumerSetup,
736        selection: ConsumerSourceSelection,
737    ) -> Result<CommittedConsumerSetup, (TransportConsumerRoute, Vec<TransportRelayRouteEffect>)>
738    {
739        let DeclaredConsumerSetup {
740            pending:
741                PendingConsumerSetup {
742                    target,
743                    consumer,
744                    reservation,
745                    sender,
746                    rtp,
747                    relays: _,
748                },
749            route,
750            mid,
751        } = setup;
752        let active = selection.delivery_active();
753        let declared_active = reservation.declared_activity().is_active();
754        let committed_mid = mid.unwrap_or_else(|| {
755            rtp.mid()
756                .map_or_else(|| consumer.to_string(), ToOwned::to_owned)
757        });
758        let (route_graph, room_router) = (&mut self.route_graph, &mut self.router);
759        // Remove pending state first so router rejection cannot strand a reservation.
760        let result =
761            route_graph.commit(reservation, route.clone(), committed_mid, selection, || {
762                match room_router.add_consumer(target.session.user_id(), consumer, target.routed) {
763                    Ok(routed_consumer) => Some(routed_consumer),
764                    Err(error) => {
765                        warn!(
766                            consumer_user_id = ?target.session.user_id(),
767                            source_id = ?target.source_id,
768                            ?error,
769                            "router rejected consumer creation"
770                        );
771                        None
772                    }
773                }
774            });
775        if let Err(relays) = result {
776            return Err((route, relays));
777        }
778        self.invalidate_video_allocation();
779        Ok(CommittedConsumerSetup {
780            target,
781            route,
782            sender,
783            transport_activity_update: (active != declared_active).then_some(active),
784        })
785    }
786
787    /// Returns `None` unless the route realization remains absent.
788    ///
789    /// A cross-worker reservation also claims relay ownership.
790    pub(super) fn reserve_consumer_setup(
791        &mut self,
792        target: ConsumerSetupTarget,
793        consumer: ConsumerId,
794        selection: ConsumerSourceSelection,
795        sender: OutboundSender,
796        rtp: RouterRtpParameters,
797    ) -> Option<PendingConsumerSetup> {
798        let key = target.subscription_key();
799        // Relay activity follows receiver intent rather than policy-gated delivery.
800        // A temporary policy pause must remain resumable without rebuilding the
801        // shared cross-worker source path.
802        let relay_active = selection.active();
803        let reservation =
804            self.route_graph
805                .reserve_consumer_setup(key, target.source_id, selection)?;
806        let source_worker = target.source.session_key().media_worker_id();
807        let target_worker = target.session.media_worker_id();
808        let relays = if source_worker == target_worker {
809            Vec::new()
810        } else {
811            self.route_graph
812                .reserve_relay(&reservation, &target, target_worker, relay_active)
813        };
814        Some(PendingConsumerSetup {
815            target,
816            consumer,
817            reservation,
818            sender,
819            rtp,
820            relays,
821        })
822    }
823
824    /// A stale reservation releases no relay ownership.
825    pub(super) fn release_consumer_setup(
826        &mut self,
827        setup: PendingConsumerSetup,
828    ) -> Vec<TransportRelayRouteEffect> {
829        self.route_graph.release_consumer_setup(setup.reservation)
830    }
831
832    /// Merges sparse intent before projecting activity to an attached route.
833    ///
834    /// Empty updates have no effect. Intent survives absent publications and
835    /// unready receivers. Returns no activity work for layout-only updates,
836    /// missing sources, changed attachments or displaced consumer connections.
837    pub(in crate::engine::room) fn apply_subscription_intent(
838        &mut self,
839        key: &SubscriptionKey,
840        connection_id: ConnectionId,
841        intent: SourceSubscriptionIntent,
842        receiver_deafened: bool,
843        publisher_present: bool,
844    ) -> ConsumerActivityCommit {
845        let mut commit = ConsumerActivityCommit {
846            update: None,
847            relay_effects: Vec::new(),
848            evictions: 0,
849        };
850        if intent.is_empty() {
851            return commit;
852        }
853        commit.evictions = self
854            .route_graph
855            .merge_intent(key, intent, publisher_present);
856        self.invalidate_video_allocation();
857        let Some(active) = intent.active() else {
858            return commit;
859        };
860        let Some(source_id) = self.source_id_for_owner_stream(&key.publisher, &key.stream) else {
861            return commit;
862        };
863        // Deafening pauses audio delivery without replacing explicit receiver intent.
864        let policy_pause_reason = (active
865            && receiver_deafened
866            && self
867                .source_descriptor(source_id)
868                .is_some_and(|source| source.media_kind() == MediaKind::Audio))
869        .then_some(PolicyPauseReason::ReceiverDeafened);
870        let Some(relay_effects) = self.route_graph.set_activity(
871            key,
872            source_id,
873            connection_id,
874            active,
875            policy_pause_reason,
876        ) else {
877            return commit;
878        };
879        let update = self
880            .committed_consumer_route_for_key(key)
881            .filter(|route| route.route.consumer_session_key().connection_id() == connection_id)
882            .map(|route| {
883                ReceiverRouteActivity::new(route.target(), route.selection.delivery_active())
884            });
885        commit.update = update;
886        commit.relay_effects = relay_effects;
887        commit
888    }
889}
o_sfu_core/engine/room/media_graph/topology.rs

o_sfu_core/engine/room/media_graph/
topology.rs
