mumble_server_runtime_shard/
journal.rs

1//! The shard's delta journal.
2//!
3//! A shard renders once and journals one delta per version. Connections replay
4//! the slice they have not seen yet, which is what makes a delta cost
5//! `O(N·|D|)` to distribute rather than `O(N·W)`.
6//!
7//! # There is no snapshot mechanism to write
8//!
9//! A newcomer receives `plan(empty view, current view)` - the ordinary planner.
10//! A departure is the inverse. A migration is both. Nothing here needs to know
11//! about any of that, which is why this module has no view vocabulary at all.
12//!
13//! # Falling off the tail closes the connection
14//!
15//! A connection whose cursor has dropped below [`Journal::tail`] cannot be
16//! repaired from deltas, and it is dying anyway: its output queue holds a
17//! thousand messages. It is closed and reconnects from scratch.
18//!
19//! REF: docs/design/guide-implementation.md 8.1
20
21use std::collections::VecDeque;
22
23use thiserror::Error;
24
25use crate::plan::PlannedOp;
26
27/// How many versions of history the journal keeps.
28///
29/// A starting point, not a truth (guide 17.3). It bounds memory, and it is the
30/// distance beyond which a connection is declared unrecoverable.
31pub const DEPTH: usize = 256;
32
33/// A connection asked for history the journal no longer holds.
34#[derive(Debug, Clone, Copy, PartialEq, Eq, Error)]
35#[error(
36    "cursor {cursor} is below the journal tail {tail}: this connection has missed too much to be \
37     repaired from deltas"
38)]
39pub struct TooFarBehind {
40    pub cursor: u64,
41    pub tail: u64,
42}
43
44/// Deltas for versions `tail + 1 ..= head`.
45#[derive(Debug, Default)]
46pub struct Journal {
47    tail: u64,
48    head: u64,
49    entries: VecDeque<Vec<PlannedOp>>,
50}
51
52impl Journal {
53    /// An empty journal at version zero.
54    #[must_use]
55    pub fn new() -> Journal {
56        Journal {
57            tail: 0,
58            head: 0,
59            entries: VecDeque::new(),
60        }
61    }
62
63    /// The newest version.
64    #[must_use]
65    pub fn head(&self) -> u64 {
66        self.head
67    }
68
69    /// The oldest version still replayable *from*.
70    #[must_use]
71    pub fn tail(&self) -> u64 {
72        self.tail
73    }
74
75    /// Append a delta, returning the version it became.
76    ///
77    /// Evicts from the front past [`DEPTH`], which is the only thing that moves
78    /// the tail.
79    pub fn push(&mut self, ops: Vec<PlannedOp>) -> u64 {
80        self.head = self.head.saturating_add(1);
81        self.entries.push_back(ops);
82        while self.entries.len() > DEPTH {
83            let _evicted = self.entries.pop_front();
84            self.tail = self.tail.saturating_add(1);
85        }
86        self.head
87    }
88
89    /// Every operation in versions `from + 1 ..= to`, in order.
90    ///
91    /// # Errors
92    ///
93    /// [`TooFarBehind`] when `from` is below the tail. A `to` above the head, or
94    /// a `to` below `from`, yields nothing: both mean the caller is already up to
95    /// date or ahead, which is not an error a connection can act on.
96    pub fn replay(&self, from: u64, to: u64) -> Result<Vec<&PlannedOp>, TooFarBehind> {
97        if from < self.tail {
98            return Err(TooFarBehind {
99                cursor: from,
100                tail: self.tail,
101            });
102        }
103        let to = to.min(self.head);
104        if to <= from {
105            return Ok(Vec::new());
106        }
107
108        // Version v lives at index `v - tail - 1`; both subtractions are safe
109        // because `from >= tail` and `v > from`.
110        let start = usize::try_from(from - self.tail).unwrap_or(usize::MAX);
111        let end = usize::try_from(to - self.tail).unwrap_or(usize::MAX);
112        Ok(self
113            .entries
114            .iter()
115            .take(end)
116            .skip(start)
117            .flatten()
118            .collect())
119    }
120}
121
122#[cfg(test)]
123mod tests {
124    #![allow(clippy::expect_used)]
125
126    use super::*;
127    use crate::ids::{ChannelId, SessionId};
128    use crate::plan::PlanOp;
129    use crate::scope::Scope;
130
131    fn delta(marker: u32) -> Vec<PlannedOp> {
132        vec![PlannedOp {
133            op: PlanOp::RemoveUser(SessionId(marker)),
134            scope: Scope::ROOT,
135        }]
136    }
137
138    fn markers(ops: &[&PlannedOp]) -> Vec<u32> {
139        ops.iter()
140            .filter_map(|planned| match planned.op {
141                PlanOp::RemoveUser(session) => Some(session.0),
142                _ => None,
143            })
144            .collect()
145    }
146
147    #[test]
148    fn replaying_in_pieces_equals_replaying_at_once() {
149        let mut journal = Journal::new();
150        for marker in 1..=10 {
151            journal.push(delta(marker));
152        }
153
154        let whole = journal.replay(0, 10).expect("within the journal");
155        let mut pieced: Vec<u32> = Vec::new();
156        for (from, to) in [(0, 3), (3, 3), (3, 7), (7, 10)] {
157            pieced.extend(markers(
158                &journal.replay(from, to).expect("within the journal"),
159            ));
160        }
161
162        assert_eq!(markers(&whole), pieced);
163        assert_eq!(markers(&whole), (1..=10).collect::<Vec<u32>>());
164    }
165
166    #[test]
167    fn a_cursor_at_the_head_replays_nothing() {
168        let mut journal = Journal::new();
169        journal.push(delta(1));
170        assert!(
171            journal
172                .replay(journal.head(), journal.head())
173                .expect("within the journal")
174                .is_empty()
175        );
176    }
177
178    #[test]
179    fn a_cursor_below_the_tail_is_detected_rather_than_silently_truncated() {
180        let mut journal = Journal::new();
181        for marker in 0..u32::try_from(DEPTH).unwrap_or(256) + 5 {
182            journal.push(delta(marker));
183        }
184
185        assert_eq!(journal.tail(), 5, "five versions were evicted");
186        assert_eq!(
187            journal.replay(4, journal.head()),
188            Err(TooFarBehind { cursor: 4, tail: 5 })
189        );
190        assert!(
191            journal.replay(5, journal.head()).is_ok(),
192            "the tail itself is still a valid cursor"
193        );
194    }
195
196    #[test]
197    fn the_journal_never_grows_past_its_depth() {
198        let mut journal = Journal::new();
199        for marker in 0..u32::try_from(DEPTH).unwrap_or(256) * 3 {
200            journal.push(delta(marker));
201        }
202        assert_eq!(journal.entries.len(), DEPTH);
203        assert_eq!(journal.head() - journal.tail(), DEPTH as u64);
204    }
205
206    #[test]
207    fn asking_beyond_the_head_stops_at_the_head() {
208        let mut journal = Journal::new();
209        journal.push(delta(1));
210        journal.push(delta(2));
211
212        let ops = journal.replay(0, 999).expect("within the journal");
213        assert_eq!(markers(&ops), vec![1, 2]);
214    }
215
216    #[test]
217    fn an_empty_delta_still_advances_the_version() {
218        let mut journal = Journal::new();
219        let version = journal.push(Vec::new());
220        assert_eq!(version, 1);
221        assert_eq!(journal.head(), 1);
222        assert_eq!(
223            journal.replay(0, 1).expect("within the journal").len(),
224            0,
225            "a version with no operations replays as nothing"
226        );
227        let _ = ChannelId::ROOT;
228    }
229}
mumble_server_runtime_shard/journal.rs

mumble_server_runtime_shard/
journal.rs
