o_sfu_router/model/topology/
mod.rs

1//! Room-local routed topology and media placement lifetimes.
2//!
3//! [`Router`] coordinates user sessions across one or more routers. Each user
4//! holds a **Home Session** on their assigned placement router where their published
5//! producers live.
6//!
7//! # Cross-Router Routing & Foreign Shadows
8//!
9//! When User 2 (placed on Router 2) subscribes to User 1's producer (published on Router 1):
10//! 1. Producer P1 lives on Router 1.
11//! 2. Router 1 uses a **Foreign Session Shadow** for User 2 (using User 2's connection ID),
12//!    creating it only when absent.
13//! 3. Consumer C2 is created inside this shadow session on Router 1, attached directly to Producer P1.
14//! 4. The router records the C2-to-P1 dependency and `o-sfu-core` forwards to User 2's media worker.
15//!
16//! ```text
17//!   RouterId 1 (Publisher Placement)                RouterId 2 (Subscriber Placement)
18//! +------------------------------------+          +------------------------------------+
19//! | Home Session (User 1):             |          | Home Session (User 2):             |
20//! |   Producer P1 dependent IDs:       |          |   Connection Conn_2                |
21//! |     |                              |          |     ^                              |
22//! |     +--> C1 (User 3 home session)  |          |     |                              |
23//! |     |                              |          |     |                              |
24//! |     +--> C2 dependency             |          |     |                              |
25//! |          (Conn_2 foreign shadow)   | -------->|-----+ (Core forwards to worker)    |
26//! +------------------------------------+          +------------------------------------+
27//!
28//! Lifecycle & Cascade Teardowns:
29//!   - Producer P1 Removed (`Router::remove_producer`):
30//!     -> Cascades down to remove local C1 and foreign C2 on Router 1.
31//!     -> The shadow session for Conn_2 becomes empty and is automatically pruned.
32//!   - User 2 Disconnects / Replaces Connection:
33//!     -> Removes User 2's Home Session on Router 2.
34//!     -> Removes C2 from P1's dependents and prunes the foreign shadow on Router 1.
35//!     -> User 1's Producer P1 and local Consumer C1 remain active and unaffected.
36//! ```
37
38use std::{
39    collections::{BTreeMap, BTreeSet},
40    iter, mem,
41};
42
43use o_sfu_model::UserId;
44
45use crate::model::{
46    ConnectionId, ConsumerId, MediaCapabilities, MediaWorkerId, ProducerId, RouterError, RouterId,
47};
48
49#[cfg(test)]
50#[path = "../TESTS/topology_support.rs"]
51pub(crate) mod test_support;
52
53/// producer identity plus its source router and connection
54#[derive(Debug, Clone, Copy, PartialEq, Eq, PartialOrd, Ord)]
55pub struct RoutedProducerId(RouterId, ConnectionId, ProducerId);
56
57impl RoutedProducerId {
58    #[cfg(any(test, feature = "test-support"))]
59    #[must_use]
60    pub const fn for_test(
61        router: RouterId,
62        connection: ConnectionId,
63        producer: ProducerId,
64    ) -> Self {
65        Self(router, connection, producer)
66    }
67
68    #[must_use]
69    pub const fn router_id(self) -> RouterId {
70        self.0
71    }
72
73    #[must_use]
74    pub const fn connection_id(self) -> ConnectionId {
75        self.1
76    }
77
78    #[must_use]
79    pub const fn producer_id(self) -> ProducerId {
80        self.2
81    }
82}
83
84/// consumer identity plus its source router and receiver connection
85#[derive(Debug, Clone, Copy, PartialEq, Eq, PartialOrd, Ord)]
86pub struct RoutedConsumerId(RouterId, ConnectionId, ConsumerId);
87
88impl RoutedConsumerId {
89    #[cfg(any(test, feature = "test-support"))]
90    #[must_use]
91    pub const fn for_test(
92        router: RouterId,
93        connection: ConnectionId,
94        consumer: ConsumerId,
95    ) -> Self {
96        Self(router, connection, consumer)
97    }
98
99    #[must_use]
100    pub const fn router_id(self) -> RouterId {
101        self.0
102    }
103
104    #[must_use]
105    pub const fn connection_id(self) -> ConnectionId {
106        self.1
107    }
108
109    #[must_use]
110    pub const fn consumer_id(self) -> ConsumerId {
111        self.2
112    }
113}
114
115#[derive(Debug, Clone, Copy, PartialEq, Eq)]
116pub struct RouterPlacement {
117    pub router: RouterId,
118    pub media_worker: MediaWorkerId,
119}
120
121#[derive(Debug, Clone, PartialEq, Eq)]
122pub struct RouterPlacements {
123    primary: RouterPlacement,
124    spillover: Vec<RouterPlacement>,
125}
126
127#[derive(Debug, Clone, Copy, PartialEq, Eq)]
128pub enum RouterPlacementsError {
129    Empty,
130}
131
132impl RouterPlacements {
133    #[must_use]
134    pub fn new(primary: RouterPlacement, spillover: Vec<RouterPlacement>) -> Self {
135        let mut placements = Self {
136            primary,
137            spillover: Vec::new(),
138        };
139        for placement in spillover {
140            if placement.router != primary.router {
141                placements.upsert(placement);
142            }
143        }
144        placements
145    }
146
147    /// # Errors
148    ///
149    /// returns [`RouterPlacementsError::Empty`] when `placements` is empty
150    pub fn try_from_vec(placements: Vec<RouterPlacement>) -> Result<Self, RouterPlacementsError> {
151        let mut placements = placements.into_iter();
152        let Some(primary) = placements.next() else {
153            return Err(RouterPlacementsError::Empty);
154        };
155        Ok(Self::new(primary, placements.collect()))
156    }
157
158    #[must_use]
159    pub const fn primary(&self) -> RouterPlacement {
160        self.primary
161    }
162
163    fn upsert(&mut self, placement: RouterPlacement) {
164        if self.primary.router == placement.router {
165            self.primary = placement;
166            return;
167        }
168        if let Some(existing) = self
169            .spillover
170            .iter_mut()
171            .find(|existing| existing.router == placement.router)
172        {
173            *existing = placement;
174        } else {
175            self.spillover.push(placement);
176        }
177    }
178
179    fn iter(&self) -> impl Iterator<Item = RouterPlacement> + '_ {
180        iter::once(self.primary).chain(self.spillover.iter().copied())
181    }
182}
183
184#[derive(Debug, Clone, PartialEq, Eq)]
185pub struct PlacementSnapshot {
186    primary: RouterId,
187    placements: Option<Vec<RouterPlacement>>,
188}
189
190impl PlacementSnapshot {
191    #[must_use]
192    pub const fn primary(&self) -> RouterId {
193        self.primary
194    }
195
196    #[cfg(any(test, feature = "test-support"))]
197    #[must_use]
198    pub fn next_router(&self) -> RouterId {
199        let router = self
200            .placements
201            .as_deref()
202            .unwrap_or_default()
203            .iter()
204            .map(|placement| placement.router.0)
205            .max()
206            .map_or(self.primary.0, |router| router.saturating_add(1));
207        RouterId(router)
208    }
209
210    #[must_use]
211    pub fn assigned_placements(&self) -> &[RouterPlacement] {
212        self.placements.as_deref().unwrap_or_default()
213    }
214}
215
216/// pure room router for placement and routed media lifetimes
217#[derive(Debug)]
218pub struct Router {
219    primary: RouterId,
220    placements: Option<RouterPlacements>,
221    capabilities: MediaCapabilities,
222    routers: BTreeMap<RouterId, LocalRouter>,
223    sessions: BTreeMap<UserId, CommittedSession>,
224    users: BTreeMap<ConnectionId, UserId>,
225}
226
227#[derive(Debug, Clone, Copy)]
228struct CommittedSession {
229    connection: ConnectionId,
230    placement: RouterPlacement,
231}
232
233#[derive(Debug, Default)]
234struct LocalRouter {
235    sessions: BTreeMap<ConnectionId, LocalSession>,
236}
237
238#[derive(Debug, Default)]
239struct LocalSession {
240    producers: BTreeMap<ProducerId, BTreeSet<RoutedConsumerId>>,
241    consumers: BTreeMap<ConsumerId, RoutedProducerId>,
242}
243
244impl LocalSession {
245    fn is_empty(&self) -> bool {
246        self.producers.is_empty() && self.consumers.is_empty()
247    }
248}
249
250impl Router {
251    #[must_use]
252    pub fn new(primary: RouterId, capabilities: MediaCapabilities) -> Self {
253        let mut routers = BTreeMap::new();
254        routers.insert(primary, LocalRouter::default());
255        Self {
256            primary,
257            placements: None,
258            capabilities,
259            routers,
260            sessions: BTreeMap::new(),
261            users: BTreeMap::new(),
262        }
263    }
264
265    #[must_use]
266    pub fn with_placements(placements: RouterPlacements, capabilities: MediaCapabilities) -> Self {
267        let mut router = Self::new(placements.primary().router, capabilities);
268        router.placements = Some(placements);
269        router
270    }
271
272    #[must_use]
273    pub const fn rtp_capabilities(&self) -> &MediaCapabilities {
274        &self.capabilities
275    }
276
277    #[must_use]
278    pub fn committed_media_worker_id(
279        &self,
280        user: &UserId,
281        connection: ConnectionId,
282    ) -> Option<MediaWorkerId> {
283        let session = self.sessions.get(user)?;
284        (session.connection == connection).then_some(session.placement.media_worker)
285    }
286
287    /// commit one user connection to a router placement
288    ///
289    /// # Errors
290    ///
291    /// returns an error when the connection is occupied or when the placement
292    /// contradicts an existing router-to-worker assignment
293    pub fn commit_session_placement(
294        &mut self,
295        user: &UserId,
296        connection: ConnectionId,
297        placement: RouterPlacement,
298    ) -> Result<MediaWorkerId, RouterError> {
299        if self.users.contains_key(&connection) {
300            return Err(RouterError::DuplicateConnection(connection));
301        }
302        self.validate_placement(placement)?;
303        if self.sessions.contains_key(user) {
304            self.remove_session(user)?;
305        }
306        self.attach_placement(placement);
307        let local = self
308            .routers
309            .get_mut(&placement.router)
310            .ok_or(RouterError::MissingRouter(placement.router))?;
311        local.sessions.insert(connection, LocalSession::default());
312        self.sessions.insert(
313            user.clone(),
314            CommittedSession {
315                connection,
316                placement,
317            },
318        );
319        self.users.insert(connection, user.clone());
320        Ok(placement.media_worker)
321    }
322
323    pub fn retire_committed_placement(
324        &mut self,
325        user: &UserId,
326        connection: ConnectionId,
327    ) -> Option<MediaWorkerId> {
328        let worker = self.committed_media_worker_id(user, connection)?;
329        self.remove_session(user).ok()?;
330        Some(worker)
331    }
332
333    /// # Panics
334    ///
335    /// panics when `connection` has no committed placement
336    #[expect(
337        clippy::unreachable,
338        reason = "route planning requires a committed connection placement"
339    )]
340    #[must_use]
341    pub fn media_worker_id_for_connection(&self, connection: ConnectionId) -> MediaWorkerId {
342        let Some(user) = self.users.get(&connection) else {
343            unreachable!("media worker lookup requires committed connection placement");
344        };
345        let Some(session) = self.sessions.get(user) else {
346            unreachable!("connection owner must have a committed session");
347        };
348        session.placement.media_worker
349    }
350
351    #[must_use]
352    pub fn primary_worker(&self) -> Option<MediaWorkerId> {
353        self.placements
354            .as_ref()
355            .map(|placements| placements.primary().media_worker)
356    }
357
358    pub fn assigned_media_workers(&self) -> impl Iterator<Item = MediaWorkerId> + '_ {
359        self.placements
360            .iter()
361            .flat_map(RouterPlacements::iter)
362            .map(|placement| placement.media_worker)
363    }
364
365    #[must_use]
366    pub fn placement_snapshot(&self) -> PlacementSnapshot {
367        let placements = self
368            .placements
369            .as_ref()
370            .map(|placements| placements.iter().collect());
371        PlacementSnapshot {
372            primary: self.primary,
373            placements,
374        }
375    }
376
377    #[cfg(any(test, feature = "test-support"))]
378    #[must_use]
379    pub fn router_count(&self) -> usize {
380        self.routers.len()
381    }
382
383    #[cfg(any(test, feature = "test-support"))]
384    #[must_use]
385    pub fn consumer_dependency_count(&self) -> usize {
386        self.routers
387            .values()
388            .flat_map(|router| router.sessions.values())
389            .flat_map(|session| session.producers.values())
390            .map(BTreeSet::len)
391            .sum()
392    }
393
394    /// create a producer on the user's home router
395    ///
396    /// # Errors
397    ///
398    /// returns a missing session or duplicate producer error
399    pub fn add_producer(
400        &mut self,
401        user: &UserId,
402        producer: ProducerId,
403    ) -> Result<RoutedProducerId, RouterError> {
404        let session = *self.require_session(user)?;
405        let routed = RoutedProducerId(session.placement.router, session.connection, producer);
406        let local = self
407            .routers
408            .get_mut(&routed.router_id())
409            .ok_or(RouterError::MissingRouter(routed.router_id()))?
410            .sessions
411            .get_mut(&routed.connection_id())
412            .ok_or_else(|| RouterError::MissingSession(user.clone()))?;
413        if local.producers.contains_key(&producer) {
414            return Err(RouterError::DuplicateProducer(producer));
415        }
416        local.producers.insert(producer, BTreeSet::new());
417        Ok(routed)
418    }
419
420    /// create a consumer on its producer's source router
421    ///
422    /// # Errors
423    ///
424    /// returns a missing session, router, producer or duplicate consumer error
425    pub fn add_consumer(
426        &mut self,
427        user: &UserId,
428        consumer: ConsumerId,
429        producer: RoutedProducerId,
430    ) -> Result<RoutedConsumerId, RouterError> {
431        let receiver = *self.require_session(user)?;
432        let local = self
433            .routers
434            .get_mut(&producer.router_id())
435            .ok_or(RouterError::MissingRouter(producer.router_id()))?;
436        if local
437            .sessions
438            .get(&producer.connection_id())
439            .and_then(|session| session.producers.get(&producer.producer_id()))
440            .is_none()
441        {
442            return Err(RouterError::MissingProducer(producer));
443        }
444        if local
445            .sessions
446            .get(&receiver.connection)
447            .is_some_and(|session| session.consumers.contains_key(&consumer))
448        {
449            return Err(RouterError::DuplicateConsumer(consumer));
450        }
451
452        let routed = RoutedConsumerId(producer.router_id(), receiver.connection, consumer);
453        local
454            .sessions
455            .entry(receiver.connection)
456            .or_default()
457            .consumers
458            .insert(consumer, producer);
459        let producer_session = local
460            .sessions
461            .get_mut(&producer.connection_id())
462            .ok_or(RouterError::MissingProducer(producer))?;
463        let consumers = producer_session
464            .producers
465            .get_mut(&producer.producer_id())
466            .ok_or(RouterError::MissingProducer(producer))?;
467        consumers.insert(routed);
468        Ok(routed)
469    }
470
471    /// remove one consumer and its empty foreign receiver session
472    ///
473    /// # Errors
474    ///
475    /// returns a missing router or consumer error
476    pub fn remove_consumer(&mut self, consumer: RoutedConsumerId) -> Result<(), RouterError> {
477        let local = self
478            .routers
479            .get_mut(&consumer.router_id())
480            .ok_or(RouterError::MissingRouter(consumer.router_id()))?;
481        let producer = local
482            .sessions
483            .get_mut(&consumer.connection_id())
484            .and_then(|session| session.consumers.remove(&consumer.consumer_id()))
485            .ok_or(RouterError::MissingConsumer(consumer))?;
486        if let Some(dependents) = local
487            .sessions
488            .get_mut(&producer.connection_id())
489            .and_then(|session| session.producers.get_mut(&producer.producer_id()))
490        {
491            dependents.remove(&consumer);
492        }
493        self.prune_foreign_session(consumer.router_id(), consumer.connection_id());
494        Ok(())
495    }
496
497    /// remove one producer and every dependent consumer
498    ///
499    /// # Errors
500    ///
501    /// returns a missing router or producer error
502    pub fn remove_producer(&mut self, producer: RoutedProducerId) -> Result<(), RouterError> {
503        let local = self
504            .routers
505            .get_mut(&producer.router_id())
506            .ok_or(RouterError::MissingRouter(producer.router_id()))?;
507        let consumers = local
508            .sessions
509            .get_mut(&producer.connection_id())
510            .and_then(|session| session.producers.remove(&producer.producer_id()))
511            .ok_or(RouterError::MissingProducer(producer))?;
512        self.remove_dependents(producer.router_id(), consumers);
513        Ok(())
514    }
515
516    /// remove a user's home session, foreign sessions and routed media
517    ///
518    /// # Errors
519    ///
520    /// returns a missing session or router error
521    pub fn remove_session(&mut self, user: &UserId) -> Result<(), RouterError> {
522        let session = *self.require_session(user)?;
523        let mut producers = mem::take(
524            &mut self
525                .routers
526                .get_mut(&session.placement.router)
527                .ok_or(RouterError::MissingRouter(session.placement.router))?
528                .sessions
529                .get_mut(&session.connection)
530                .ok_or_else(|| RouterError::MissingSession(user.clone()))?
531                .producers,
532        );
533        for consumers in producers.values_mut() {
534            self.remove_dependents(session.placement.router, mem::take(consumers));
535        }
536        for local in self.routers.values_mut() {
537            let Some(removed) = local.sessions.remove(&session.connection) else {
538                continue;
539            };
540            for (consumer, producer) in &removed.consumers {
541                let routed = RoutedConsumerId(producer.router_id(), session.connection, *consumer);
542                if let Some(dependents) = local
543                    .sessions
544                    .get_mut(&producer.connection_id())
545                    .and_then(|source| source.producers.get_mut(&producer.producer_id()))
546                {
547                    dependents.remove(&routed);
548                }
549            }
550        }
551        self.sessions.remove(user);
552        self.users.remove(&session.connection);
553        Ok(())
554    }
555
556    fn remove_dependents(&mut self, router: RouterId, consumers: BTreeSet<RoutedConsumerId>) {
557        for consumer in consumers {
558            if let Some(session) = self
559                .routers
560                .get_mut(&router)
561                .and_then(|local| local.sessions.get_mut(&consumer.connection_id()))
562            {
563                session.consumers.remove(&consumer.consumer_id());
564            }
565            self.prune_foreign_session(router, consumer.connection_id());
566        }
567    }
568
569    fn attach_placement(&mut self, placement: RouterPlacement) {
570        match &mut self.placements {
571            Some(placements) => placements.upsert(placement),
572            None => self.placements = Some(RouterPlacements::new(placement, Vec::new())),
573        }
574        self.routers.entry(placement.router).or_default();
575    }
576
577    fn validate_placement(&self, placement: RouterPlacement) -> Result<(), RouterError> {
578        let Some(placements) = &self.placements else {
579            return if placement.router == self.primary {
580                Ok(())
581            } else {
582                Err(RouterError::PrimaryRouterMismatch {
583                    expected: self.primary,
584                    actual: placement.router,
585                })
586            };
587        };
588        if let Some(existing) = placements
589            .iter()
590            .find(|existing| existing.router == placement.router)
591            && existing.media_worker != placement.media_worker
592        {
593            return Err(RouterError::MediaWorkerMismatch {
594                router: placement.router,
595                expected: existing.media_worker,
596                actual: placement.media_worker,
597            });
598        }
599        Ok(())
600    }
601
602    fn require_session(&self, user: &UserId) -> Result<&CommittedSession, RouterError> {
603        self.sessions
604            .get(user)
605            .ok_or_else(|| RouterError::MissingSession(user.clone()))
606    }
607
608    fn prune_foreign_session(&mut self, router: RouterId, connection: ConnectionId) {
609        let home = self
610            .users
611            .get(&connection)
612            .and_then(|user| self.sessions.get(user))
613            .map(|session| session.placement.router);
614        if home == Some(router) {
615            return;
616        }
617        let Some(local) = self.routers.get_mut(&router) else {
618            return;
619        };
620        if local
621            .sessions
622            .get(&connection)
623            .is_some_and(LocalSession::is_empty)
624        {
625            local.sessions.remove(&connection);
626        }
627    }
628}
o_sfu_router/model/topology/mod.rs

o_sfu_router/model/topology/
mod.rs
