mumble_server_runtime_shard/
compose.rs

1//! Composing one connection's transition out of the shared delta.
2//!
3//! This is the heart of the runtime, and it is about fifteen lines:
4//!
5//! ```text
6//! let shared  = filter(journal.replay(cursor, head), see);
7//! let private = plan_elements(&overlay_sent, &overlay_new);
8//! let mut ops = splice(shared, private);
9//! collapse(&mut ops);
10//! ```
11//!
12//! # What must not be done instead
13//!
14//! `diff(overlay + shared, committed)` is the honest formulation and it is always
15//! correct - and it costs **O(V) per connection**, so O(N·W) overall. That is
16//! precisely the quadratic this design exists to escape. Composition costs
17//! `O(|D|) + O(|overlay|) + O(|ops|)`.
18//!
19//! REF: docs/design/guide-implementation.md 6
20
21use std::collections::HashSet;
22
23use crate::plan::{ElementId, OverlayOps, PlanOp, PlannedOp};
24use crate::scope::ScopeSet;
25
26/// Keep the operations this connection can see.
27///
28/// **Filtering preserves validity for free.** Every ordering rule in the plan is
29/// a constraint of the form "X before Y", and dropping elements from a sequence
30/// violates none of them. So a valid plan, filtered, is still a valid plan and
31/// there is nothing to replan. The closure theorem does the real work: it is what
32/// guarantees no reference is left dangling by the elements that were dropped.
33///
34/// Takes anything that iterates borrowed operations, so a freshly planned
35/// `Vec<PlannedOp>` and a `Vec<&PlannedOp>` replayed out of the journal both go
36/// through the same function.
37#[must_use]
38pub fn filter<'a>(ops: impl IntoIterator<Item = &'a PlannedOp>, see: ScopeSet) -> Vec<PlanOp> {
39    ops.into_iter()
40        .filter(|planned| see.sees(planned.scope))
41        .map(|planned| planned.op.clone())
42        .collect()
43}
44
45/// Insert the overlay's operations **inside** the shared phases.
46///
47/// The overlay does not append. The counter-example is immediate: suppose the
48/// shared delta removes channel C, and the overlay had placed the admin in it.
49/// With the private operations last, `RemoveChannel(C)` goes out *before* the
50/// admin is withdrawn, and an occupied channel is deleted.
51///
52/// So the insertion point is the boundary between the shared additions
53/// (P1 to P5) and the shared removals (P6, P7):
54///
55/// ```text
56/// P1..P5   shared additions, filtered
57///          ├─ overlay removals      <- before: they vacate a channel about to die
58///          └─ overlay additions     <- after:  they may target a brand new channel
59/// P6..P7   shared removals, filtered
60/// ```
61#[must_use]
62pub fn splice(shared: Vec<PlanOp>, private: OverlayOps) -> Vec<PlanOp> {
63    let boundary = shared
64        .iter()
65        .position(PlanOp::is_removal)
66        .unwrap_or(shared.len());
67
68    let mut composed =
69        Vec::with_capacity(shared.len() + private.additions.len() + private.removals.len());
70    let mut shared = shared;
71    let tail = shared.split_off(boundary);
72    composed.append(&mut shared);
73    composed.extend(private.removals);
74    composed.extend(private.additions);
75    composed.extend(tail);
76    composed
77}
78
79/// Drop every removal of an element that is also added in the same transition.
80///
81/// One pass, three jobs, because in all three the right answer is the same:
82/// **an element that has an `Add` somewhere still exists, so the `Remove` is
83/// wrong.** A full `AddUser`/`CreateChannel` carries the complete state, and the
84/// client merges `UserState`/`ChannelState` rather than replacing, so re-sending
85/// the element whole simply updates it.
86///
87/// | situation | operations produced | after `collapse` |
88/// |---|---|---|
89/// | the element changes scope | Remove(old) + Add(new) | Add alone |
90/// | vanish to unvanish | Remove(overlay) + Add(shared) | Add alone, it changes channel |
91/// | unvanish to vanish | Remove(shared) + Add(overlay) | Add alone |
92/// | dropped from the overlay, absent from the shared view | Remove alone | Remove kept |
93/// | gone everywhere | Remove alone | Remove kept |
94///
95/// Three problems that looked distinct, one function: that is the sign it sits
96/// at the right level.
97pub fn collapse(ops: &mut Vec<PlanOp>) {
98    let added: HashSet<ElementId> = ops.iter().filter_map(PlanOp::added).collect();
99    ops.retain(|op| !matches!(op.removed(), Some(id) if added.contains(&id)));
100}
101
102#[cfg(test)]
103mod tests {
104    #![allow(clippy::expect_used)]
105
106    use std::collections::{BTreeMap, BTreeSet};
107
108    use super::*;
109    use crate::ids::{ChannelId, ChannelKey, ConnectionId, Occupant, SessionId};
110    use crate::plan::{plan, plan_elements};
111    use crate::scope::Scope;
112    use crate::view::{Channel, Overlay, ShardView, User, UserFlags};
113
114    fn scope(segments: &[u32]) -> Scope {
115        let mut scope = Scope::ROOT;
116        for segment in segments {
117            scope = scope.child(*segment).expect("within MAX_DEPTH");
118        }
119        scope
120    }
121
122    fn channel(id: u32, parent: u32, scope: Scope) -> Channel {
123        Channel {
124            key: ChannelKey(u64::from(id)),
125            id: ChannelId(id),
126            parent: ChannelId(parent),
127            scope,
128            name: format!("channel-{id}"),
129            position: 0,
130            can_enter: true,
131            can_text: true,
132            links: BTreeSet::new(),
133        }
134    }
135
136    fn user(session: u32, channel: u32, scope: Scope) -> User {
137        User {
138            occupant: Occupant::Connection(ConnectionId(u64::from(session))),
139            session: SessionId(session),
140            channel: ChannelId(channel),
141            scope,
142            name: format!("user-{session}"),
143            flags: UserFlags::default(),
144        }
145    }
146
147    /// Root, game, two teams, one player in each.
148    fn two_teams() -> ShardView {
149        let mut view = ShardView::empty();
150        view.channels
151            .insert(ChannelId::ROOT, channel(0, 0, Scope::ROOT));
152        view.channels
153            .insert(ChannelId(1), channel(1, 0, scope(&[7])));
154        view.channels
155            .insert(ChannelId(2), channel(2, 1, scope(&[7, 2])));
156        view.channels
157            .insert(ChannelId(3), channel(3, 1, scope(&[7, 3])));
158        view.users
159            .insert(SessionId(10), user(10, 2, scope(&[7, 2])));
160        view.users
161            .insert(SessionId(11), user(11, 3, scope(&[7, 3])));
162        view
163    }
164
165    fn sessions_touched(ops: &[PlanOp]) -> Vec<(&'static str, u32)> {
166        ops.iter()
167            .filter_map(|op| match op {
168                PlanOp::AddUser(user) => Some(("add", user.session.0)),
169                PlanOp::RemoveUser(session) => Some(("remove", session.0)),
170                PlanOp::MoveUser { session, .. } => Some(("move", session.0)),
171                _ => None,
172            })
173            .collect()
174    }
175
176    #[test]
177    fn a_scope_change_reaches_each_observer_as_the_half_that_concerns_it() {
178        // Player 11 leaves team 3 for team 2.
179        let before = two_teams();
180        let mut after = before.clone();
181        after
182            .users
183            .insert(SessionId(11), user(11, 2, scope(&[7, 2])));
184        let ops = plan(&before, &after);
185
186        let team_three = ScopeSet::new(&[scope(&[7, 3])]).expect("one scope");
187        let team_two = ScopeSet::new(&[scope(&[7, 2])]).expect("one scope");
188        let spectator = ScopeSet::new(&[scope(&[7])]).expect("one scope");
189        let elsewhere = ScopeSet::new(&[scope(&[8])]).expect("one scope");
190
191        // The teammate left behind only learns of the departure.
192        assert_eq!(
193            sessions_touched(&filter(&ops, team_three)),
194            vec![("remove", 11)]
195        );
196        // The new team only learns of the arrival.
197        assert_eq!(sessions_touched(&filter(&ops, team_two)), vec![("add", 11)]);
198        // Another game learns nothing at all.
199        assert!(filter(&ops, elsewhere).is_empty());
200
201        // The spectator sees both halves, and collapse settles it into a move.
202        let mut both = filter(&ops, spectator);
203        assert_eq!(
204            sessions_touched(&both),
205            vec![("add", 11), ("remove", 11)],
206            "the raw filter yields both halves"
207        );
208        collapse(&mut both);
209        assert_eq!(
210            sessions_touched(&both),
211            vec![("add", 11)],
212            "a full AddUser merges client-side, which is the move"
213        );
214    }
215
216    #[test]
217    fn a_move_within_one_scope_stays_a_move() {
218        let before = two_teams();
219        let mut after = before.clone();
220        // Same scope, different channel: a plain relocation, not a departure.
221        after
222            .users
223            .insert(SessionId(10), user(10, 1, scope(&[7, 2])));
224
225        let ops = plan(&before, &after);
226        let team_two = ScopeSet::new(&[scope(&[7, 2])]).expect("one scope");
227        assert_eq!(
228            sessions_touched(&filter(&ops, team_two)),
229            vec![("move", 10)]
230        );
231    }
232
233    #[test]
234    fn filtering_preserves_order() {
235        let before = two_teams();
236        let mut after = before.clone();
237        after
238            .channels
239            .insert(ChannelId(4), channel(4, 2, scope(&[7, 2])));
240        after.users.remove(&SessionId(10));
241
242        let ops = plan(&before, &after);
243        let everything = ScopeSet::new(&[Scope::ROOT]).expect("one scope");
244        let filtered = filter(&ops, everything);
245
246        let unfiltered: Vec<&PlanOp> = ops.iter().map(|planned| &planned.op).collect();
247        let kept: Vec<&PlanOp> = filtered.iter().collect();
248        assert_eq!(unfiltered, kept, "an unrestricted filter is the identity");
249
250        // And a real filter is a subsequence, never a reordering.
251        let team_two = ScopeSet::new(&[scope(&[7, 2])]).expect("one scope");
252        let restricted = filter(&ops, team_two);
253        let mut remaining = restricted.iter();
254        for op in &ops {
255            if remaining.clone().next() == Some(&op.op) {
256                let _ = remaining.next();
257            }
258        }
259        assert_eq!(remaining.count(), 0, "filtering must not reorder");
260    }
261
262    #[test]
263    fn the_overlay_is_spliced_before_the_shared_removals() {
264        // The shared delta deletes the channel the overlay was using.
265        let shared = vec![
266            PlanOp::CreateChannel(channel(5, 0, Scope::ROOT)),
267            PlanOp::RemoveChannel(ChannelId(2)),
268        ];
269        let private = OverlayOps {
270            additions: vec![PlanOp::AddUser(user(99, 5, Scope::ROOT))],
271            removals: vec![PlanOp::RemoveUser(SessionId(98))],
272        };
273
274        let composed = splice(shared, private);
275        let shapes: Vec<&str> = composed
276            .iter()
277            .map(|op| match op {
278                PlanOp::CreateChannel(_) => "create-channel",
279                PlanOp::AddUser(_) => "add-user",
280                PlanOp::RemoveUser(_) => "remove-user",
281                PlanOp::RemoveChannel(_) => "remove-channel",
282                _ => "other",
283            })
284            .collect();
285
286        assert_eq!(
287            shapes,
288            vec![
289                "create-channel",
290                "remove-user",
291                "add-user",
292                "remove-channel"
293            ],
294            "the overlay's withdrawal must precede the shared channel removal"
295        );
296    }
297
298    #[test]
299    fn an_overlay_addition_may_target_a_channel_created_in_the_same_turn() {
300        let shared = vec![PlanOp::CreateChannel(channel(5, 0, Scope::ROOT))];
301        let private = OverlayOps {
302            additions: vec![PlanOp::AddUser(user(99, 5, Scope::ROOT))],
303            removals: Vec::new(),
304        };
305
306        let composed = splice(shared, private);
307        let create = composed
308            .iter()
309            .position(|op| matches!(op, PlanOp::CreateChannel(_)))
310            .expect("the creation is present");
311        let add = composed
312            .iter()
313            .position(|op| matches!(op, PlanOp::AddUser(_)))
314            .expect("the addition is present");
315        assert!(create < add);
316    }
317
318    #[test]
319    fn vanishing_and_unvanishing_both_collapse_to_a_single_addition() {
320        let admin = Occupant::Connection(ConnectionId(99));
321        let placed = User {
322            occupant: admin,
323            session: SessionId(99),
324            channel: ChannelId(2),
325            scope: scope(&[7, 2]),
326            name: "admin".to_owned(),
327            flags: UserFlags::default(),
328        };
329
330        // unvanish: the overlay drops them, the shared view picks them up.
331        let mut overlay_before = Overlay::default();
332        overlay_before.users.insert(SessionId(99), placed.clone());
333        let private = plan_elements(&overlay_before, &Overlay::default());
334        let mut ops = splice(vec![PlanOp::AddUser(placed.clone())], private);
335        collapse(&mut ops);
336        assert_eq!(sessions_touched(&ops), vec![("add", 99)]);
337
338        // vanish: the shared view drops them, the overlay picks them up.
339        let mut overlay_after = Overlay::default();
340        overlay_after.users.insert(SessionId(99), placed);
341        let private = plan_elements(&Overlay::default(), &overlay_after);
342        let mut ops = splice(vec![PlanOp::RemoveUser(SessionId(99))], private);
343        collapse(&mut ops);
344        assert_eq!(sessions_touched(&ops), vec![("add", 99)]);
345    }
346
347    #[test]
348    fn a_removal_with_no_matching_addition_survives_collapse() {
349        let mut ops = vec![
350            PlanOp::RemoveUser(SessionId(10)),
351            PlanOp::RemoveChannel(ChannelId(2)),
352        ];
353        collapse(&mut ops);
354        assert_eq!(ops.len(), 2, "an element that is gone must stay gone");
355    }
356
357    #[test]
358    fn splicing_into_an_all_removal_plan_still_lands_before_them() {
359        let shared = vec![PlanOp::RemoveUser(SessionId(10))];
360        let private = OverlayOps {
361            additions: vec![PlanOp::AddUser(user(99, 2, Scope::ROOT))],
362            removals: Vec::new(),
363        };
364        let composed = splice(shared, private);
365        assert!(matches!(composed[0], PlanOp::AddUser(_)));
366    }
367
368    #[test]
369    fn an_empty_overlay_leaves_the_shared_plan_untouched() {
370        let shared = vec![
371            PlanOp::CreateChannel(channel(5, 0, Scope::ROOT)),
372            PlanOp::RemoveUser(SessionId(10)),
373        ];
374        let composed = splice(shared.clone(), OverlayOps::default());
375        assert_eq!(composed, shared);
376    }
377
378    #[test]
379    fn an_unobserved_delta_filters_to_nothing() {
380        let before = two_teams();
381        let mut after = before.clone();
382        after.users.remove(&SessionId(10));
383        let ops = plan(&before, &after);
384
385        assert!(filter(&ops, ScopeSet::NONE).is_empty());
386        assert!(BTreeMap::<u32, u32>::new().is_empty());
387    }
388}
mumble_server_runtime_shard/compose.rs

mumble_server_runtime_shard/
compose.rs
