mumble_server_runtime_shard/
routing.rs

1//! The audio relation, and the routing table compiled from it.
2//!
3//! Scopes drive the **shared visual** only. Audio is a directed relation the
4//! flavor declares outright - one way, both ways, or not at all - because
5//! "spectators hear the team without being heard" is not expressible as a tree
6//! position, and forcing it to be is what makes visibility models grow special
7//! cases.
8//!
9//! | primitive | cost | for |
10//! |---|---|---|
11//! | [`AudioRelation::domain`] | O(members squared) | the bulk: a channel, a team |
12//! | [`AudioRelation::listen`] | O(members) | admin, spectator: hears without being heard |
13//! | [`AudioRelation::edge`] | O(1) | full generality, at the flavor's cost |
14//!
15//! REF: docs/design/guide-implementation.md 3.5, 8.3
16
17use std::collections::{BTreeMap, BTreeSet};
18
19use crate::ids::{ConnectionId, SessionId};
20use crate::view::UserFlags;
21
22/// Who may not speak, and who may not hear, as the rendered view says it.
23///
24/// The audio plane is the only place these flags mean anything: everywhere else
25/// they are an icon. Compiling them into the table rather than testing them per
26/// packet keeps the hot path free of the question, and makes a mute take effect
27/// on the very turn that announces it - the table is published before any view.
28///
29/// A flavor owns the flags, as it owns everything else it renders. What it does
30/// **not** own is whether a user it renders as muted can still be heard: a client
31/// showing a crossed-out microphone next to someone whose voice comes through is
32/// a lie the runtime would be telling on the flavor's behalf.
33#[derive(Debug, Clone, PartialEq, Eq, Default)]
34pub struct Silence {
35    muted: BTreeSet<SessionId>,
36    deafened: BTreeSet<SessionId>,
37}
38
39impl Silence {
40    /// Record what one rendered user's flags mean for the audio plane.
41    ///
42    /// The two predicates are the reference server's own, field for field.
43    ///
44    /// REF: references/mumble/src/murmur/Server.cpp : `processMsg` drops the
45    ///   packet before anything else when the speaker is
46    ///   `bMute || bSuppress || bSelfMute`.
47    /// REF: references/mumble/src/murmur/AudioReceiverBuffer.cpp : `addReceiver`
48    ///   refuses a receiver that is `bDeaf || bSelfDeaf`.
49    pub fn record(&mut self, session: SessionId, flags: UserFlags) {
50        if flags.self_mute || flags.mute || flags.suppress {
51            self.muted.insert(session);
52        }
53        if flags.self_deaf || flags.deaf {
54            self.deafened.insert(session);
55        }
56    }
57
58    /// Whether this session's voice may leave the server at all.
59    #[must_use]
60    pub fn may_speak(&self, session: SessionId) -> bool {
61        !self.muted.contains(&session)
62    }
63
64    /// Whether this session may be given anyone's voice.
65    #[must_use]
66    pub fn may_hear(&self, session: SessionId) -> bool {
67        !self.deafened.contains(&session)
68    }
69}
70
71/// A named group of connections that all hear each other.
72#[derive(Debug, Clone, Copy, PartialEq, Eq, PartialOrd, Ord, Hash)]
73pub struct DomainId(pub u64);
74
75/// The directed "who may hear whom" relation, as the flavor declared it.
76///
77/// Accumulated during a render and compiled once at the end. Kept in terms of
78/// [`ConnectionId`] because that is the vocabulary a flavor thinks in; the
79/// translation to [`SessionId`] happens at compile time, against the view.
80#[derive(Debug, Clone, PartialEq, Eq, Default)]
81pub struct AudioRelation {
82    domains: BTreeMap<DomainId, BTreeSet<ConnectionId>>,
83    listeners: BTreeSet<(ConnectionId, DomainId)>,
84    edges: BTreeSet<(ConnectionId, ConnectionId)>,
85}
86
87impl AudioRelation {
88    /// Declare a symmetric group: every member hears every other member.
89    ///
90    /// Repeated calls with the same domain accumulate members rather than
91    /// replacing them, so a flavor may build a domain across several loops.
92    pub fn domain(&mut self, domain: DomainId, members: &[ConnectionId]) {
93        self.domains
94            .entry(domain)
95            .or_default()
96            .extend(members.iter().copied());
97    }
98
99    /// Declare a one-way exception: `listener` hears the domain, and is not
100    /// heard by it.
101    pub fn listen(&mut self, listener: ConnectionId, domain: DomainId) {
102        self.listeners.insert((listener, domain));
103    }
104
105    /// Declare a single directed edge.
106    pub fn edge(&mut self, sender: ConnectionId, receiver: ConnectionId) {
107        self.edges.insert((sender, receiver));
108    }
109
110    /// The members of each declared domain.
111    pub fn domains(&self) -> impl Iterator<Item = (DomainId, &BTreeSet<ConnectionId>)> {
112        self.domains.iter().map(|(id, members)| (*id, members))
113    }
114
115    /// Each listen declaration, paired with the domain's members.
116    ///
117    /// A listen on a domain that was never declared yields nothing: failing
118    /// closed means a typo costs silence rather than a leak.
119    pub fn listeners(&self) -> impl Iterator<Item = (ConnectionId, &BTreeSet<ConnectionId>)> {
120        self.listeners
121            .iter()
122            .filter_map(|(listener, domain)| Some((*listener, self.domains.get(domain)?)))
123    }
124
125    /// Each listen declaration as written, including ones naming a domain that
126    /// does not exist.
127    pub fn listen_declarations(&self) -> impl Iterator<Item = (ConnectionId, DomainId)> + '_ {
128        self.listeners.iter().copied()
129    }
130
131    /// The edges declared one at a time.
132    pub fn explicit_edges(&self) -> impl Iterator<Item = (ConnectionId, ConnectionId)> + '_ {
133        self.edges.iter().copied()
134    }
135
136    /// Every directed edge implied by the declarations, as `(sender, receiver)`.
137    ///
138    /// A member never hears itself: the client plays back its own voice locally,
139    /// and echoing it from the server is the classic doubled-voice bug. The
140    /// explicit server loopback target is a separate mechanism and is not a
141    /// route.
142    ///
143    /// This is the **definition** of the relation, and it materializes every
144    /// pair - which is quadratic for a full-mesh domain. Nothing on a shard's
145    /// turn calls it: [`compile`] walks the same structure without a tree, and
146    /// the render's own checks walk it by distinct scope. It stays because it
147    /// says plainly what the relation means, and a test pins [`compile`] to it.
148    #[must_use]
149    pub fn resolve(&self) -> BTreeSet<(ConnectionId, ConnectionId)> {
150        let mut resolved = self.edges.clone();
151
152        for members in self.domains.values() {
153            for sender in members {
154                for receiver in members {
155                    if sender != receiver {
156                        resolved.insert((*sender, *receiver));
157                    }
158                }
159            }
160        }
161
162        for (listener, members) in self.listeners() {
163            for sender in members {
164                if *sender != listener {
165                    resolved.insert((*sender, listener));
166                }
167            }
168        }
169
170        resolved
171    }
172}
173
174/// The compiled table the voice plane reads.
175///
176/// Replaced whole, never mutated in place: the voice plane loads it without a
177/// lock, and mutating it would put a lock on the audio path.
178///
179/// The transport half of the guide's `Delivery` - the live UDP address, the OCB2
180/// state, the output queue - is deliberately absent. Binding sessions to live
181/// transports is the voice plane's job (guide 9.7, build step 8); what a shard
182/// owes it is *who may hear whom, and from which version*.
183#[derive(Debug, Clone, PartialEq, Eq, Default)]
184pub struct AudioRouting {
185    receivers: BTreeMap<SessionId, Vec<SessionId>>,
186    since: BTreeMap<SessionId, u64>,
187    /// Carried through so the one delivery that is **not** a route - the server
188    /// loopback a client asks for explicitly - can ask the same question.
189    silence: Silence,
190}
191
192impl AudioRouting {
193    /// The sessions that may hear `sender`, or an empty slice.
194    #[must_use]
195    pub fn receivers(&self, sender: SessionId) -> &[SessionId] {
196        self.receivers.get(&sender).map_or(&[], Vec::as_slice)
197    }
198
199    /// The shard version at which `session` became visible.
200    ///
201    /// The voice plane gates on this: a receiver whose cursor has not reached
202    /// the sender's `since` does not get the packet, because it has not been
203    /// told the sender exists yet and would discard the audio anyway.
204    ///
205    /// It is per participant rather than per pair, which is what keeps the table
206    /// O(N) instead of O(N squared).
207    #[must_use]
208    pub fn since(&self, session: SessionId) -> Option<u64> {
209        self.since.get(&session).copied()
210    }
211
212    /// Whether `receiver` may hear `sender` at all, ignoring the cursor gate.
213    #[must_use]
214    pub fn may_hear(&self, sender: SessionId, receiver: SessionId) -> bool {
215        self.receivers(sender).contains(&receiver)
216    }
217
218    /// Whether this session's voice may leave the server at all.
219    ///
220    /// Every route already answers it - a muted sender has no receivers - so
221    /// this exists for the one delivery that goes through no route: the server
222    /// loopback. A muted microphone that still echoes back would tell its owner
223    /// the line is open when nobody else can hear a thing.
224    #[must_use]
225    pub fn may_speak(&self, sender: SessionId) -> bool {
226        self.silence.may_speak(sender)
227    }
228
229    /// Every sender that has at least one receiver.
230    pub fn senders(&self) -> impl Iterator<Item = SessionId> + '_ {
231        self.receivers.keys().copied()
232    }
233}
234
235/// Compile the declared relation into the table the voice plane reads.
236///
237/// `session_of` resolves a connection to the session it is rendered under.
238/// Connections with no rendered user are dropped from the table: a sender nobody
239/// can see is a sender whose audio the client would discard (guide 1.2), and a
240/// receiver with no session has nothing to deliver to.
241///
242/// `since` carries the version each session first appeared at, threaded through
243/// from the shard so it survives recompilation.
244///
245/// `silence` removes muted senders and deafened receivers before any route is
246/// written down, rather than after: a muted speaker in a domain of fifty costs
247/// nothing at all here, where filtering the finished table would cost the fifty
248/// routes it should never have had.
249///
250/// # Cost
251///
252/// The compiled table is quadratic in a domain's size **by construction**: a
253/// full mesh of M members really does have M·(M-1) directed routes, and no
254/// representation of an adjacency list avoids writing them down. What this
255/// avoids is doing so through an ordered set: the members are resolved once per
256/// domain and the receiver lists are appended to directly, so the cost is a
257/// quadratic number of `Vec` pushes rather than of tree insertions. That is the
258/// difference between microseconds and tens of milliseconds at 500 connections.
259pub fn compile(
260    relation: &AudioRelation,
261    session_of: &BTreeMap<ConnectionId, SessionId>,
262    since: &BTreeMap<SessionId, u64>,
263    silence: &Silence,
264) -> AudioRouting {
265    let mut receivers: BTreeMap<SessionId, Vec<SessionId>> = BTreeMap::new();
266    let resolve = |connections: &BTreeSet<ConnectionId>| -> Vec<SessionId> {
267        connections
268            .iter()
269            .filter_map(|connection| session_of.get(connection).copied())
270            .collect()
271    };
272
273    let mut listening_on: BTreeMap<DomainId, Vec<SessionId>> = BTreeMap::new();
274    for (listener, domain) in relation.listen_declarations() {
275        if let Some(session) = session_of.get(&listener).filter(|s| silence.may_hear(**s)) {
276            listening_on.entry(domain).or_default().push(*session);
277        }
278    }
279
280    for (domain, members) in relation.domains() {
281        let sessions = resolve(members);
282        // Resolved once per domain rather than per pair: deafening one member of
283        // a domain of M costs one filtered pass, not M tests inside the loop
284        // that writes M squared routes.
285        let audience: Vec<SessionId> = sessions
286            .iter()
287            .copied()
288            .filter(|session| silence.may_hear(*session))
289            .collect();
290        let empty: Vec<SessionId> = Vec::new();
291        let listening = listening_on.get(&domain).unwrap_or(&empty);
292
293        for sender in sessions.iter().filter(|s| silence.may_speak(**s)) {
294            let list = receivers.entry(*sender).or_default();
295            // Compared by value rather than by position, because `audience` is
296            // no longer index-aligned with `sessions`. Two connections never
297            // share a session, so the two tests are the same test.
298            list.extend(audience.iter().filter(|receiver| *receiver != sender));
299            list.extend(listening.iter().filter(|listener| *listener != sender));
300        }
301    }
302
303    for (sender, receiver) in relation.explicit_edges() {
304        let (Some(sender), Some(receiver)) = (session_of.get(&sender), session_of.get(&receiver))
305        else {
306            continue;
307        };
308        if !silence.may_speak(*sender) || !silence.may_hear(*receiver) {
309            continue;
310        }
311        receivers.entry(*sender).or_default().push(*receiver);
312    }
313
314    for list in receivers.values_mut() {
315        // Sorted for determinism, deduplicated because a connection reachable
316        // through both a domain and an explicit edge is still one receiver.
317        list.sort_unstable();
318        list.dedup();
319    }
320    receivers.retain(|_, list| !list.is_empty());
321
322    AudioRouting {
323        receivers,
324        since: since.clone(),
325        silence: silence.clone(),
326    }
327}
328
329#[cfg(test)]
330mod tests {
331    #![allow(clippy::expect_used)]
332
333    use super::*;
334
335    fn sessions(pairs: &[(u64, u32)]) -> BTreeMap<ConnectionId, SessionId> {
336        pairs
337            .iter()
338            .map(|(connection, session)| (ConnectionId(*connection), SessionId(*session)))
339            .collect()
340    }
341
342    #[test]
343    fn a_domain_is_symmetric_and_excludes_self() {
344        let mut relation = AudioRelation::default();
345        relation.domain(
346            DomainId(1),
347            &[ConnectionId(1), ConnectionId(2), ConnectionId(3)],
348        );
349
350        let edges = relation.resolve();
351        assert!(edges.contains(&(ConnectionId(1), ConnectionId(2))));
352        assert!(edges.contains(&(ConnectionId(2), ConnectionId(1))));
353        assert!(
354            !edges.contains(&(ConnectionId(1), ConnectionId(1))),
355            "echoing a speaker back to itself is the doubled-voice bug"
356        );
357        assert_eq!(edges.len(), 6, "three members, every ordered pair but self");
358    }
359
360    #[test]
361    fn a_listener_hears_without_being_heard() {
362        let mut relation = AudioRelation::default();
363        relation.domain(DomainId(1), &[ConnectionId(1), ConnectionId(2)]);
364        relation.listen(ConnectionId(9), DomainId(1));
365
366        let edges = relation.resolve();
367        assert!(edges.contains(&(ConnectionId(1), ConnectionId(9))));
368        assert!(edges.contains(&(ConnectionId(2), ConnectionId(9))));
369        assert!(
370            !edges.contains(&(ConnectionId(9), ConnectionId(1))),
371            "a spectator must stay silent"
372        );
373    }
374
375    #[test]
376    fn listening_to_an_undeclared_domain_grants_nothing() {
377        let mut relation = AudioRelation::default();
378        relation.listen(ConnectionId(9), DomainId(404));
379
380        assert!(
381            relation.resolve().is_empty(),
382            "a typo must cost silence, never a leak"
383        );
384    }
385
386    #[test]
387    fn compiling_drops_connections_with_no_rendered_user() {
388        let mut relation = AudioRelation::default();
389        relation.domain(DomainId(1), &[ConnectionId(1), ConnectionId(2)]);
390
391        // Only connection 1 is rendered, so no edge survives: it has nobody
392        // visible to talk to.
393        let routing = compile(
394            &relation,
395            &sessions(&[(1, 100)]),
396            &BTreeMap::new(),
397            &Silence::default(),
398        );
399        assert!(routing.receivers(SessionId(100)).is_empty());
400    }
401
402    #[test]
403    fn a_receiver_reachable_twice_is_listed_once() {
404        let mut relation = AudioRelation::default();
405        relation.domain(DomainId(1), &[ConnectionId(1), ConnectionId(2)]);
406        relation.edge(ConnectionId(1), ConnectionId(2));
407
408        let routing = compile(
409            &relation,
410            &sessions(&[(1, 100), (2, 200)]),
411            &BTreeMap::new(),
412            &Silence::default(),
413        );
414        assert_eq!(routing.receivers(SessionId(100)), &[SessionId(200)]);
415    }
416
417    /// A `Silence` built the way a shard builds it: from rendered flags.
418    fn silenced(flags: &[(u32, UserFlags)]) -> Silence {
419        let mut silence = Silence::default();
420        for (session, flags) in flags {
421            silence.record(SessionId(*session), *flags);
422        }
423        silence
424    }
425
426    fn muted() -> UserFlags {
427        UserFlags {
428            self_mute: true,
429            ..UserFlags::default()
430        }
431    }
432
433    fn deafened() -> UserFlags {
434        UserFlags {
435            self_deaf: true,
436            ..UserFlags::default()
437        }
438    }
439
440    #[test]
441    fn a_muted_speaker_has_no_receivers_at_all() {
442        let mut relation = AudioRelation::default();
443        relation.domain(
444            DomainId(1),
445            &[ConnectionId(1), ConnectionId(2), ConnectionId(3)],
446        );
447        relation.edge(ConnectionId(1), ConnectionId(4));
448        relation.listen(ConnectionId(4), DomainId(1));
449
450        let session_of = sessions(&[(1, 10), (2, 20), (3, 30), (4, 40)]);
451        let routing = compile(
452            &relation,
453            &session_of,
454            &BTreeMap::new(),
455            &silenced(&[(10, muted())]),
456        );
457
458        assert!(
459            routing.receivers(SessionId(10)).is_empty(),
460            "a muted microphone must have no line to anyone, by any primitive"
461        );
462        assert!(!routing.may_speak(SessionId(10)));
463        assert_eq!(
464            routing.receivers(SessionId(20)),
465            &[SessionId(10), SessionId(30), SessionId(40)],
466            "muting silences a microphone, not an ear: the muted one is still a \
467             receiver, and the others keep every line they had"
468        );
469    }
470
471    #[test]
472    fn a_deafened_receiver_appears_in_nobody_s_list() {
473        let mut relation = AudioRelation::default();
474        relation.domain(DomainId(1), &[ConnectionId(1), ConnectionId(2)]);
475        relation.edge(ConnectionId(3), ConnectionId(2));
476        relation.listen(ConnectionId(2), DomainId(1));
477
478        let session_of = sessions(&[(1, 10), (2, 20), (3, 30)]);
479        let routing = compile(
480            &relation,
481            &session_of,
482            &BTreeMap::new(),
483            &silenced(&[(20, deafened())]),
484        );
485
486        for sender in [SessionId(10), SessionId(30)] {
487            assert!(
488                !routing.receivers(sender).contains(&SessionId(20)),
489                "a deafened session must not be a receiver of {sender:?}"
490            );
491        }
492        assert_eq!(
493            routing.receivers(SessionId(20)),
494            &[SessionId(10)],
495            "deafening silences the ear, not the microphone"
496        );
497        assert!(routing.may_speak(SessionId(20)));
498    }
499
500    #[test]
501    fn the_server_flags_silence_exactly_as_their_self_counterparts_do() {
502        // `mute` and `suppress` are the moderation equivalents of `self_mute`,
503        // and `deaf` of `self_deaf`. Rendering one of them while the voice still
504        // flows would put an icon on a lie.
505        let server_muted = UserFlags {
506            mute: true,
507            ..UserFlags::default()
508        };
509        let suppressed = UserFlags {
510            suppress: true,
511            ..UserFlags::default()
512        };
513        let server_deafened = UserFlags {
514            deaf: true,
515            ..UserFlags::default()
516        };
517
518        let silence = silenced(&[(10, server_muted), (20, suppressed), (30, server_deafened)]);
519        assert!(!silence.may_speak(SessionId(10)));
520        assert!(!silence.may_speak(SessionId(20)));
521        assert!(!silence.may_hear(SessionId(30)));
522        assert!(
523            silence.may_speak(SessionId(30)),
524            "server-deafening is not server-muting"
525        );
526    }
527
528    #[test]
529    fn compile_agrees_with_the_relations_definition() {
530        // `compile` walks the structure directly for speed while `resolve`
531        // spells out what the relation means. Nothing keeps them together
532        // except this: a shape with overlapping domains, a listener, a shared
533        // member and a stray edge, compiled both ways.
534        let mut relation = AudioRelation::default();
535        relation.domain(
536            DomainId(1),
537            &[ConnectionId(1), ConnectionId(2), ConnectionId(3)],
538        );
539        relation.domain(DomainId(2), &[ConnectionId(3), ConnectionId(4)]);
540        relation.listen(ConnectionId(5), DomainId(1));
541        relation.listen(ConnectionId(5), DomainId(2));
542        relation.listen(ConnectionId(3), DomainId(2));
543        relation.listen(ConnectionId(6), DomainId(404));
544        relation.edge(ConnectionId(4), ConnectionId(1));
545        relation.edge(ConnectionId(1), ConnectionId(2));
546
547        let session_of = sessions(&[(1, 10), (2, 20), (3, 30), (4, 40), (5, 50), (6, 60)]);
548        let compiled = compile(
549            &relation,
550            &session_of,
551            &BTreeMap::new(),
552            &Silence::default(),
553        );
554
555        let mut expected: BTreeMap<SessionId, Vec<SessionId>> = BTreeMap::new();
556        for (sender, receiver) in relation.resolve() {
557            let (Some(sender), Some(receiver)) =
558                (session_of.get(&sender), session_of.get(&receiver))
559            else {
560                continue;
561            };
562            expected.entry(*sender).or_default().push(*receiver);
563        }
564        for list in expected.values_mut() {
565            list.sort_unstable();
566            list.dedup();
567        }
568
569        for (sender, receivers) in &expected {
570            assert_eq!(
571                compiled.receivers(*sender),
572                receivers.as_slice(),
573                "compile disagrees with resolve for sender {sender:?}"
574            );
575        }
576        assert_eq!(
577            compiled.senders().collect::<Vec<SessionId>>(),
578            expected.keys().copied().collect::<Vec<SessionId>>(),
579            "compile and resolve must agree on which senders exist at all"
580        );
581    }
582
583    #[test]
584    fn since_is_carried_per_participant() {
585        let since = BTreeMap::from([(SessionId(100), 7), (SessionId(200), 9)]);
586        let routing = compile(
587            &AudioRelation::default(),
588            &BTreeMap::new(),
589            &since,
590            &Silence::default(),
591        );
592
593        assert_eq!(routing.since(SessionId(100)), Some(7));
594        assert_eq!(routing.since(SessionId(200)), Some(9));
595        assert_eq!(routing.since(SessionId(300)), None);
596    }
597}
mumble_server_runtime_shard/routing.rs

mumble_server_runtime_shard/
routing.rs
