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    #[must_use]
359    pub fn placement_snapshot(&self) -> PlacementSnapshot {
360        let placements = self
361            .placements
362            .as_ref()
363            .map(|placements| placements.iter().collect());
364        PlacementSnapshot {
365            primary: self.primary,
366            placements,
367        }
368    }
369
370    #[cfg(any(test, feature = "test-support"))]
371    #[must_use]
372    pub fn router_count(&self) -> usize {
373        self.routers.len()
374    }
375
376    #[cfg(any(test, feature = "test-support"))]
377    #[must_use]
378    pub fn consumer_dependency_count(&self) -> usize {
379        self.routers
380            .values()
381            .flat_map(|router| router.sessions.values())
382            .flat_map(|session| session.producers.values())
383            .map(BTreeSet::len)
384            .sum()
385    }
386
387    /// create a producer on the user's home router
388    ///
389    /// # Errors
390    ///
391    /// returns a missing session or duplicate producer error
392    pub fn add_producer(
393        &mut self,
394        user: &UserId,
395        producer: ProducerId,
396    ) -> Result<RoutedProducerId, RouterError> {
397        let session = *self.require_session(user)?;
398        let routed = RoutedProducerId(session.placement.router, session.connection, producer);
399        let local = self
400            .routers
401            .get_mut(&routed.router_id())
402            .ok_or(RouterError::MissingRouter(routed.router_id()))?
403            .sessions
404            .get_mut(&routed.connection_id())
405            .ok_or_else(|| RouterError::MissingSession(user.clone()))?;
406        if local.producers.contains_key(&producer) {
407            return Err(RouterError::DuplicateProducer(producer));
408        }
409        local.producers.insert(producer, BTreeSet::new());
410        Ok(routed)
411    }
412
413    /// create a consumer on its producer's source router
414    ///
415    /// # Errors
416    ///
417    /// returns a missing session, router, producer or duplicate consumer error
418    pub fn add_consumer(
419        &mut self,
420        user: &UserId,
421        consumer: ConsumerId,
422        producer: RoutedProducerId,
423    ) -> Result<RoutedConsumerId, RouterError> {
424        let receiver = *self.require_session(user)?;
425        let local = self
426            .routers
427            .get_mut(&producer.router_id())
428            .ok_or(RouterError::MissingRouter(producer.router_id()))?;
429        if local
430            .sessions
431            .get(&producer.connection_id())
432            .and_then(|session| session.producers.get(&producer.producer_id()))
433            .is_none()
434        {
435            return Err(RouterError::MissingProducer(producer));
436        }
437        if local
438            .sessions
439            .get(&receiver.connection)
440            .is_some_and(|session| session.consumers.contains_key(&consumer))
441        {
442            return Err(RouterError::DuplicateConsumer(consumer));
443        }
444
445        let routed = RoutedConsumerId(producer.router_id(), receiver.connection, consumer);
446        local
447            .sessions
448            .entry(receiver.connection)
449            .or_default()
450            .consumers
451            .insert(consumer, producer);
452        let producer_session = local
453            .sessions
454            .get_mut(&producer.connection_id())
455            .ok_or(RouterError::MissingProducer(producer))?;
456        let consumers = producer_session
457            .producers
458            .get_mut(&producer.producer_id())
459            .ok_or(RouterError::MissingProducer(producer))?;
460        consumers.insert(routed);
461        Ok(routed)
462    }
463
464    /// remove one consumer and its empty foreign receiver session
465    ///
466    /// # Errors
467    ///
468    /// returns a missing router or consumer error
469    pub fn remove_consumer(&mut self, consumer: RoutedConsumerId) -> Result<(), RouterError> {
470        let local = self
471            .routers
472            .get_mut(&consumer.router_id())
473            .ok_or(RouterError::MissingRouter(consumer.router_id()))?;
474        let producer = local
475            .sessions
476            .get_mut(&consumer.connection_id())
477            .and_then(|session| session.consumers.remove(&consumer.consumer_id()))
478            .ok_or(RouterError::MissingConsumer(consumer))?;
479        if let Some(dependents) = local
480            .sessions
481            .get_mut(&producer.connection_id())
482            .and_then(|session| session.producers.get_mut(&producer.producer_id()))
483        {
484            dependents.remove(&consumer);
485        }
486        self.prune_foreign_session(consumer.router_id(), consumer.connection_id());
487        Ok(())
488    }
489
490    /// remove one producer and every dependent consumer
491    ///
492    /// # Errors
493    ///
494    /// returns a missing router or producer error
495    pub fn remove_producer(&mut self, producer: RoutedProducerId) -> Result<(), RouterError> {
496        let local = self
497            .routers
498            .get_mut(&producer.router_id())
499            .ok_or(RouterError::MissingRouter(producer.router_id()))?;
500        let consumers = local
501            .sessions
502            .get_mut(&producer.connection_id())
503            .and_then(|session| session.producers.remove(&producer.producer_id()))
504            .ok_or(RouterError::MissingProducer(producer))?;
505        self.remove_dependents(producer.router_id(), consumers);
506        Ok(())
507    }
508
509    /// remove a user's home session, foreign sessions and routed media
510    ///
511    /// # Errors
512    ///
513    /// returns a missing session or router error
514    pub fn remove_session(&mut self, user: &UserId) -> Result<(), RouterError> {
515        let session = *self.require_session(user)?;
516        let mut producers = mem::take(
517            &mut self
518                .routers
519                .get_mut(&session.placement.router)
520                .ok_or(RouterError::MissingRouter(session.placement.router))?
521                .sessions
522                .get_mut(&session.connection)
523                .ok_or_else(|| RouterError::MissingSession(user.clone()))?
524                .producers,
525        );
526        for consumers in producers.values_mut() {
527            self.remove_dependents(session.placement.router, mem::take(consumers));
528        }
529        for local in self.routers.values_mut() {
530            let Some(removed) = local.sessions.remove(&session.connection) else {
531                continue;
532            };
533            for (consumer, producer) in &removed.consumers {
534                let routed = RoutedConsumerId(producer.router_id(), session.connection, *consumer);
535                if let Some(dependents) = local
536                    .sessions
537                    .get_mut(&producer.connection_id())
538                    .and_then(|source| source.producers.get_mut(&producer.producer_id()))
539                {
540                    dependents.remove(&routed);
541                }
542            }
543        }
544        self.sessions.remove(user);
545        self.users.remove(&session.connection);
546        Ok(())
547    }
548
549    fn remove_dependents(&mut self, router: RouterId, consumers: BTreeSet<RoutedConsumerId>) {
550        for consumer in consumers {
551            if let Some(session) = self
552                .routers
553                .get_mut(&router)
554                .and_then(|local| local.sessions.get_mut(&consumer.connection_id()))
555            {
556                session.consumers.remove(&consumer.consumer_id());
557            }
558            self.prune_foreign_session(router, consumer.connection_id());
559        }
560    }
561
562    fn attach_placement(&mut self, placement: RouterPlacement) {
563        match &mut self.placements {
564            Some(placements) => placements.upsert(placement),
565            None => self.placements = Some(RouterPlacements::new(placement, Vec::new())),
566        }
567        self.routers.entry(placement.router).or_default();
568    }
569
570    fn validate_placement(&self, placement: RouterPlacement) -> Result<(), RouterError> {
571        let Some(placements) = &self.placements else {
572            return if placement.router == self.primary {
573                Ok(())
574            } else {
575                Err(RouterError::PrimaryRouterMismatch {
576                    expected: self.primary,
577                    actual: placement.router,
578                })
579            };
580        };
581        if let Some(existing) = placements
582            .iter()
583            .find(|existing| existing.router == placement.router)
584            && existing.media_worker != placement.media_worker
585        {
586            return Err(RouterError::MediaWorkerMismatch {
587                router: placement.router,
588                expected: existing.media_worker,
589                actual: placement.media_worker,
590            });
591        }
592        Ok(())
593    }
594
595    fn require_session(&self, user: &UserId) -> Result<&CommittedSession, RouterError> {
596        self.sessions
597            .get(user)
598            .ok_or_else(|| RouterError::MissingSession(user.clone()))
599    }
600
601    fn prune_foreign_session(&mut self, router: RouterId, connection: ConnectionId) {
602        let home = self
603            .users
604            .get(&connection)
605            .and_then(|user| self.sessions.get(user))
606            .map(|session| session.placement.router);
607        if home == Some(router) {
608            return;
609        }
610        let Some(local) = self.routers.get_mut(&router) else {
611            return;
612        };
613        if local
614            .sessions
615            .get(&connection)
616            .is_some_and(LocalSession::is_empty)
617        {
618            local.sessions.remove(&connection);
619        }
620    }
621}
o_sfu_router/model/topology/mod.rs

o_sfu_router/model/topology/
mod.rs
