ozzy_replication/
clock.rs

1//! Whether this broker's wall clock agrees with the other brokers. No I/O: the
2//! owner supplies each sample and reads both clocks. Monotonic time alone decides
3//! which samples still count, so a wall clock step cannot revive an old one.
4
5use std::time::Duration;
6
7use ozzy_proto::NodeId;
8
9/// How often a broker sends its wall time to each peer.
10pub const SAMPLE_INTERVAL: Duration = Duration::from_secs(1);
11
12/// How long a received sample counts.
13pub const SAMPLE_FRESH: Duration = Duration::from_secs(5);
14
15/// What the fresh samples say about the local clock.
16#[derive(Debug, Clone, Copy, PartialEq, Eq)]
17pub enum ClockVerdict {
18    /// The clocks agree within the tolerance.
19    Agreed,
20    /// Fresh samples put the local clock outside the tolerance.
21    Disagreed,
22    /// No fresh sample: nothing to compare with.
23    Unknown,
24}
25
26#[derive(Debug, Clone, Copy)]
27struct Sample {
28    offset_millis: i64,
29    arrived: Duration,
30}
31
32/// The latest sample of each peer. A delayed sample looks like a slow peer, so
33/// it can pause decisions but never approve a wrong clock: the median needs a
34/// majority of the clocks within the tolerance.
35#[derive(Debug, Clone)]
36pub struct ClockGuard {
37    tolerance_millis: u64,
38    peers: Vec<(NodeId, Option<Sample>)>,
39}
40
41impl ClockGuard {
42    /// Compare with `peers`, allowing `tolerance_millis` of disagreement.
43    pub fn new(peers: impl IntoIterator<Item = NodeId>, tolerance_millis: u64) -> Self {
44        Self {
45            tolerance_millis,
46            peers: peers.into_iter().map(|peer| (peer, None)).collect(),
47        }
48    }
49
50    /// Keep `peer`'s sample: its wall time as sent, the local wall time at
51    /// receipt, and the local monotonic time. Unconfigured peers are ignored.
52    pub fn observe(
53        &mut self,
54        peer: NodeId,
55        peer_wall_millis: u64,
56        local_wall_millis: u64,
57        now: Duration,
58    ) {
59        let difference = i128::from(peer_wall_millis) - i128::from(local_wall_millis);
60        let offset_millis =
61            i64::try_from(difference).unwrap_or(if difference < 0 { i64::MIN } else { i64::MAX });
62        if let Some((_, slot)) = self.peers.iter_mut().find(|(id, _)| *id == peer) {
63            *slot = Some(Sample {
64                offset_millis,
65                arrived: now,
66            });
67        }
68    }
69
70    /// Drop every sample. Their offsets were measured against a clock that has
71    /// since stepped.
72    pub fn forget(&mut self) {
73        for (_, sample) in &mut self.peers {
74            *sample = None;
75        }
76    }
77
78    /// Each peer's fresh offset in milliseconds, peer minus local.
79    pub fn offsets(&self, now: Duration) -> impl Iterator<Item = (NodeId, Option<i64>)> + '_ {
80        self.peers.iter().map(move |(peer, sample)| {
81            let fresh = sample.filter(|sample| now.saturating_sub(sample.arrived) <= SAMPLE_FRESH);
82            (*peer, fresh.map(|sample| sample.offset_millis))
83        })
84    }
85
86    /// Take the median of zero (this clock) and the fresh offsets. With two
87    /// clocks both count, because neither can outvote the other.
88    pub fn verdict(&self, now: Duration) -> ClockVerdict {
89        let mut clocks: Vec<i64> = self
90            .offsets(now)
91            .filter_map(|(_, offset)| offset)
92            .chain([0])
93            .collect();
94        if clocks.len() < 2 {
95            return ClockVerdict::Unknown;
96        }
97        clocks.sort_unstable();
98        let lower = (clocks.len() - 1) / 2;
99        let upper = clocks.len() / 2;
100        let agreed = [lower, upper].iter().all(|&index| {
101            clocks
102                .get(index)
103                .is_some_and(|offset| offset.unsigned_abs() <= self.tolerance_millis)
104        });
105        if agreed {
106            ClockVerdict::Agreed
107        } else {
108            ClockVerdict::Disagreed
109        }
110    }
111}
112
113#[cfg(test)]
114mod tests {
115    use super::*;
116
117    const HOUR: u64 = 3_600_000;
118    const WALL: u64 = 10 * HOUR;
119    const TOLERANCE: u64 = 1000;
120
121    fn node(byte: u8) -> NodeId {
122        NodeId::from_bytes([byte; 16])
123    }
124
125    fn at(seconds: u64) -> Duration {
126        Duration::from_secs(seconds)
127    }
128
129    fn guard() -> ClockGuard {
130        ClockGuard::new([node(1), node(2)], TOLERANCE)
131    }
132
133    #[test]
134    fn no_fresh_sample_is_unknown() {
135        assert_eq!(guard().verdict(at(0)), ClockVerdict::Unknown);
136    }
137
138    #[test]
139    fn a_fast_local_clock_disagrees_with_both_peers() {
140        let mut guard = guard();
141        guard.observe(node(1), WALL, WALL + HOUR, at(0));
142        guard.observe(node(2), WALL, WALL + HOUR, at(0));
143        assert_eq!(guard.verdict(at(0)), ClockVerdict::Disagreed);
144    }
145
146    #[test]
147    fn a_slow_local_clock_disagrees_with_both_peers() {
148        let mut guard = guard();
149        guard.observe(node(1), WALL + HOUR, WALL, at(0));
150        guard.observe(node(2), WALL + HOUR, WALL, at(0));
151        assert_eq!(guard.verdict(at(0)), ClockVerdict::Disagreed);
152    }
153
154    #[test]
155    fn one_bad_peer_is_outvoted_whichever_way_it_is_wrong() {
156        for bad in [WALL + HOUR, WALL - HOUR] {
157            let mut guard = guard();
158            guard.observe(node(1), bad, WALL, at(0));
159            guard.observe(node(2), WALL, WALL, at(0));
160            assert_eq!(guard.verdict(at(0)), ClockVerdict::Agreed);
161        }
162    }
163
164    #[test]
165    fn a_clock_between_two_bad_peers_agrees() {
166        let mut guard = guard();
167        guard.observe(node(1), WALL + HOUR, WALL, at(0));
168        guard.observe(node(2), WALL - HOUR, WALL, at(0));
169        assert_eq!(guard.verdict(at(0)), ClockVerdict::Agreed);
170    }
171
172    #[test]
173    fn one_sample_cannot_tell_which_clock_is_wrong() {
174        let mut guard = guard();
175        guard.observe(node(1), WALL + HOUR, WALL, at(0));
176        assert_eq!(guard.verdict(at(0)), ClockVerdict::Disagreed);
177        guard.observe(node(1), WALL - HOUR, WALL, at(0));
178        assert_eq!(guard.verdict(at(0)), ClockVerdict::Disagreed);
179        guard.observe(node(1), WALL + TOLERANCE, WALL, at(0));
180        assert_eq!(guard.verdict(at(0)), ClockVerdict::Agreed);
181    }
182
183    #[test]
184    fn the_tolerance_is_inclusive() {
185        for (peer, expected) in [
186            (WALL + TOLERANCE, ClockVerdict::Agreed),
187            (WALL - TOLERANCE, ClockVerdict::Agreed),
188            (WALL + TOLERANCE + 1, ClockVerdict::Disagreed),
189            (WALL - TOLERANCE - 1, ClockVerdict::Disagreed),
190        ] {
191            let mut guard = guard();
192            guard.observe(node(1), peer, WALL, at(0));
193            guard.observe(node(2), peer, WALL, at(0));
194            assert_eq!(guard.verdict(at(0)), expected, "peer wall {peer}");
195        }
196    }
197
198    #[test]
199    fn one_slow_link_never_pauses() {
200        let mut guard = guard();
201        // Sent an hour before receipt, against a prompt sample from the other.
202        guard.observe(node(1), WALL - HOUR, WALL, at(0));
203        guard.observe(node(2), WALL - 5, WALL, at(0));
204        assert_eq!(guard.verdict(at(0)), ClockVerdict::Agreed);
205    }
206
207    #[test]
208    fn stale_samples_stop_counting() {
209        let mut guard = guard();
210        guard.observe(node(1), WALL, WALL, at(0));
211        guard.observe(node(2), WALL, WALL, at(3));
212        assert_eq!(guard.verdict(at(5)), ClockVerdict::Agreed);
213        // Only the newer one is left: it alone decides.
214        assert_eq!(guard.verdict(at(6)), ClockVerdict::Agreed);
215        assert_eq!(
216            guard.offsets(at(6)).collect::<Vec<_>>(),
217            [(node(1), None), (node(2), Some(0))]
218        );
219        assert_eq!(guard.verdict(at(9)), ClockVerdict::Unknown);
220    }
221
222    #[test]
223    fn a_stale_bad_sample_leaves_the_good_one_alone() {
224        let mut guard = guard();
225        guard.observe(node(1), WALL + HOUR, WALL, at(0));
226        guard.observe(node(2), WALL, WALL, at(4));
227        assert_eq!(guard.verdict(at(6)), ClockVerdict::Agreed);
228    }
229
230    #[test]
231    fn a_wall_clock_step_does_not_revive_a_stale_sample() {
232        let mut guard = guard();
233        guard.observe(node(1), WALL, WALL, at(0));
234        guard.observe(node(2), WALL, WALL, at(0));
235        // Freshness only reads the monotonic time passed in.
236        assert_eq!(guard.verdict(at(6)), ClockVerdict::Unknown);
237        guard.observe(node(1), WALL, WALL - HOUR, at(6));
238        assert_eq!(guard.verdict(at(6)), ClockVerdict::Disagreed);
239    }
240
241    #[test]
242    fn a_new_sample_replaces_the_old_one() {
243        let mut guard = guard();
244        guard.observe(node(1), WALL + HOUR, WALL, at(0));
245        guard.observe(node(2), WALL + HOUR, WALL, at(0));
246        assert_eq!(guard.verdict(at(0)), ClockVerdict::Disagreed);
247        guard.observe(node(1), WALL, WALL, at(1));
248        guard.observe(node(2), WALL, WALL, at(1));
249        assert_eq!(guard.verdict(at(1)), ClockVerdict::Agreed);
250    }
251
252    #[test]
253    fn extreme_wall_times_saturate() {
254        let mut guard = guard();
255        guard.observe(node(1), u64::MAX, 0, at(0));
256        guard.observe(node(2), 0, u64::MAX, at(0));
257        assert_eq!(
258            guard.offsets(at(0)).collect::<Vec<_>>(),
259            [(node(1), Some(i64::MAX)), (node(2), Some(i64::MIN))]
260        );
261        assert_eq!(guard.verdict(at(0)), ClockVerdict::Agreed);
262        guard.observe(node(2), u64::MAX, 0, at(0));
263        assert_eq!(guard.verdict(at(0)), ClockVerdict::Disagreed);
264    }
265
266    #[test]
267    fn an_unconfigured_peer_is_ignored() {
268        let mut guard = guard();
269        guard.observe(node(9), WALL, WALL, at(0));
270        assert_eq!(guard.verdict(at(0)), ClockVerdict::Unknown);
271    }
272}
ozzy_replication/clock.rs

ozzy_replication/
clock.rs
