librotenado/
clock.rs

1// SPDX-License-Identifier: MIT
2//! Hybrid logical clocks: an order of events across devices that respects causality and stays close to real time.
3
4use std::{fmt, str::FromStr};
5
6use serde::{Deserialize, Deserializer, Serialize, Serializer, de};
7
8use crate::{Object, ParseError};
9
10/// The name of a device that writes events, unique within a set of books: 1–64 characters of `a-z`, `0-9`, `-`.
11#[derive(Debug, Clone, PartialEq, Eq, PartialOrd, Ord, Hash)]
12pub struct DeviceId(String);
13
14impl DeviceId {
15    /// The name as text.
16    pub fn as_str(&self) -> &str {
17        &self.0
18    }
19}
20
21impl FromStr for DeviceId {
22    type Err = ParseError;
23
24    fn from_str(s: &str) -> Result<Self, Self::Err> {
25        let valid =
26            (1..=64).contains(&s.len()) && s.bytes().all(|b| b.is_ascii_lowercase() || b.is_ascii_digit() || b == b'-');
27        if valid { Ok(DeviceId(s.to_owned())) } else { Err(ParseError::Device(s.to_owned())) }
28    }
29}
30
31impl fmt::Display for DeviceId {
32    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
33        f.write_str(&self.0)
34    }
35}
36
37impl Serialize for DeviceId {
38    fn serialize<S: Serializer>(&self, serializer: S) -> Result<S::Ok, S::Error> {
39        serializer.serialize_str(&self.0)
40    }
41}
42
43impl<'de> Deserialize<'de> for DeviceId {
44    fn deserialize<D: Deserializer<'de>>(deserializer: D) -> Result<Self, D::Error> {
45        String::deserialize(deserializer)?.parse().map_err(de::Error::custom)
46    }
47}
48
49/// A timestamp of a hybrid logical clock.
50///
51/// Ordered by wall time, then counter, then device – a total order on which every device agrees. The wall time is
52/// in milliseconds since 1970-01-01 UTC; the counter orders events within the same millisecond and lets the clock
53/// move on when a device's own clock is behind a time it has seen.
54#[derive(Debug, Clone, PartialEq, Eq, PartialOrd, Ord, Hash, Serialize, Deserialize)]
55pub struct Hlc {
56    /// Milliseconds since 1970-01-01 UTC.
57    #[serde(deserialize_with = "crate::safe_u64")]
58    pub wall_ms: u64,
59    /// Order within the same millisecond.
60    pub counter: u32,
61    /// The device that made the timestamp.
62    pub device: DeviceId,
63}
64
65impl Object for Hlc {
66    const TYPE: &'static str = "timestamp";
67}
68
69/// The hybrid logical clock of one device.
70#[derive(Debug, Clone)]
71pub struct Clock {
72    device: DeviceId,
73    wall_ms: u64,
74    counter: u32,
75}
76
77impl Clock {
78    /// A clock for a device that has not made or seen any timestamp yet.
79    pub fn new(device: DeviceId) -> Self {
80        Clock { device, wall_ms: 0, counter: 0 }
81    }
82
83    /// A timestamp for a new local event, given the device's physical time in milliseconds.
84    pub fn now(&mut self, physical_ms: u64) -> Hlc {
85        if physical_ms > self.wall_ms {
86            self.wall_ms = physical_ms;
87            self.counter = 0;
88        } else {
89            self.counter += 1;
90        }
91        self.stamp()
92    }
93
94    /// Takes note of a timestamp received from another device, so that every later local timestamp is after it.
95    pub fn observe(&mut self, seen: &Hlc, physical_ms: u64) {
96        let wall_ms = self.wall_ms.max(seen.wall_ms).max(physical_ms);
97        self.counter = match (wall_ms == self.wall_ms, wall_ms == seen.wall_ms) {
98            (true, true) => self.counter.max(seen.counter),
99            (true, false) => self.counter,
100            (false, true) => seen.counter,
101            (false, false) => 0,
102        };
103        self.wall_ms = wall_ms;
104    }
105
106    fn stamp(&self) -> Hlc {
107        Hlc { wall_ms: self.wall_ms, counter: self.counter, device: self.device.clone() }
108    }
109}
110
111#[cfg(test)]
112mod tests {
113    use super::*;
114
115    fn device(name: &str) -> DeviceId {
116        name.parse().unwrap()
117    }
118
119    #[test]
120    fn device_ids() {
121        assert!("laptop-2".parse::<DeviceId>().is_ok());
122        for s in ["", "Laptop", "my laptop", &"x".repeat(65)] {
123            assert!(s.parse::<DeviceId>().is_err(), "{s:?}");
124        }
125    }
126
127    #[test]
128    fn local_timestamps_always_increase() {
129        let mut clock = Clock::new(device("a"));
130        let first = clock.now(1000);
131        let same_ms = clock.now(1000);
132        let clock_went_back = clock.now(500);
133        assert!(first < same_ms && same_ms < clock_went_back);
134        assert_eq!(clock_went_back.wall_ms, 1000);
135    }
136
137    #[test]
138    fn after_observing_a_later_timestamp_the_next_one_is_later_still() {
139        let mut ahead = Clock::new(device("ahead"));
140        let mut behind = Clock::new(device("behind"));
141        let remote = ahead.now(5000);
142        behind.observe(&remote, 1000);
143        let local = behind.now(1000);
144        assert!(local > remote);
145        assert_eq!(local.wall_ms, 5000);
146    }
147
148    #[test]
149    fn ties_are_broken_by_device() {
150        let a = Clock::new(device("a")).now(1000);
151        let b = Clock::new(device("b")).now(1000);
152        assert!(a < b);
153    }
154}
librotenado/clock.rs

librotenado/
clock.rs
