baukit_sync/
hlc.rs

1//! Hybrid logical timestamps compatible with JavaScript safe integers.
2//!
3//! An encoded timestamp is `wall_time_ms * 1_000 + counter + 1`. The added one
4//! reserves zero as invalid. [`HybridLogicalClock`] combines an injected
5//! physical clock with a logical counter, so local timestamps keep increasing
6//! when physical time stalls or moves backward.
7
8use std::cmp::Ordering;
9
10use serde::{Deserialize, Serialize};
11use thiserror::Error;
12
13/// Number of logical counter values available in one physical millisecond.
14pub const COUNTERS_PER_MILLISECOND: i64 = 1_000;
15
16/// Largest encoded timestamp that every JavaScript runtime can represent exactly.
17pub const MAX_ENCODED_TIMESTAMP: i64 = 9_007_199_254_740_991;
18
19/// Serializable state needed to restore a clock.
20#[derive(Clone, Debug, Eq, PartialEq, Serialize, Deserialize)]
21#[serde(rename_all = "camelCase")]
22pub struct HybridLogicalClockState {
23    /// Physical millisecond component retained by the clock.
24    pub wall_time_ms: i64,
25    /// Logical counter within `wall_time_ms`.
26    pub counter: i64,
27    /// Caller-supplied identifier used to reject state from another clock.
28    pub device_id: String,
29}
30
31/// Invalid input or exhausted JavaScript-safe timestamp space.
32#[derive(Clone, Copy, Debug, Eq, Error, PartialEq)]
33#[non_exhaustive]
34pub enum HlcError {
35    /// A wall-time or counter component is outside its accepted range.
36    #[error("invalid hybrid logical timestamp component")]
37    InvalidComponent,
38    /// An encoded timestamp is not a positive JavaScript safe integer.
39    #[error("invalid encoded hybrid logical timestamp")]
40    InvalidTimestamp,
41    /// Encoding or logical rollover would exceed the JavaScript safe-integer limit.
42    #[error("hybrid logical timestamp exceeds safe integer range")]
43    ExceedsSafeInteger,
44    /// The injected physical clock returned a negative or unsafe value.
45    #[error("physical clock must return a non-negative safe integer")]
46    InvalidPhysicalClock,
47    /// The caller supplied a blank clock identifier.
48    #[error("HLC device id must not be empty")]
49    EmptyDeviceId,
50}
51
52/// Encodes physical milliseconds and a logical counter into one timestamp.
53///
54/// # Errors
55///
56/// Returns [`HlcError::InvalidComponent`] for a negative wall time or a counter
57/// outside `0..1000`. Returns [`HlcError::ExceedsSafeInteger`] when the result
58/// is above [`MAX_ENCODED_TIMESTAMP`].
59pub fn encode(wall_time_ms: i64, counter: i64) -> Result<i64, HlcError> {
60    if wall_time_ms < 0 || !(0..COUNTERS_PER_MILLISECOND).contains(&counter) {
61        return Err(HlcError::InvalidComponent);
62    }
63
64    let encoded = wall_time_ms
65        .checked_mul(COUNTERS_PER_MILLISECOND)
66        .and_then(|value| value.checked_add(counter))
67        .and_then(|value| value.checked_add(1))
68        .ok_or(HlcError::ExceedsSafeInteger)?;
69    if encoded > MAX_ENCODED_TIMESTAMP {
70        return Err(HlcError::ExceedsSafeInteger);
71    }
72    Ok(encoded)
73}
74
75/// Decodes a positive JavaScript-safe timestamp into physical and logical parts.
76///
77/// # Errors
78///
79/// Returns [`HlcError::InvalidTimestamp`] for zero, negative values, or values
80/// above [`MAX_ENCODED_TIMESTAMP`].
81pub fn decode(timestamp: i64) -> Result<(i64, i64), HlcError> {
82    if !(1..=MAX_ENCODED_TIMESTAMP).contains(&timestamp) {
83        return Err(HlcError::InvalidTimestamp);
84    }
85
86    let zero_based = timestamp - 1;
87    Ok((
88        zero_based / COUNTERS_PER_MILLISECOND,
89        zero_based % COUNTERS_PER_MILLISECOND,
90    ))
91}
92
93/// Compares two valid encoded timestamps.
94///
95/// Returns `None` when either input is not a positive JavaScript-safe encoded
96/// timestamp.
97#[must_use]
98pub fn compare(left: i64, right: i64) -> Option<Ordering> {
99    (decode(left).is_ok() && decode(right).is_ok()).then(|| left.cmp(&right))
100}
101
102/// Stateful hybrid logical clock backed by a caller-supplied physical clock.
103pub struct HybridLogicalClock<C> {
104    device_id: String,
105    physical_clock: C,
106    state: HybridLogicalClockState,
107}
108
109impl<C> HybridLogicalClock<C>
110where
111    C: FnMut() -> i64,
112{
113    /// Opens a clock from optional persisted state.
114    ///
115    /// State with invalid components, an unencodable value, or a different
116    /// `device_id` is treated as corrupt and replaced with zero state. The
117    /// caller owns loading and saving the snapshot.
118    ///
119    /// # Errors
120    ///
121    /// Returns [`HlcError::EmptyDeviceId`] when `device_id` is blank.
122    pub fn open(
123        device_id: impl Into<String>,
124        physical_clock: C,
125        persisted: Option<HybridLogicalClockState>,
126    ) -> Result<Self, HlcError> {
127        let device_id = device_id.into();
128        if device_id.trim().is_empty() {
129            return Err(HlcError::EmptyDeviceId);
130        }
131
132        let state = match persisted {
133            Some(state)
134                if state.device_id == device_id
135                    && encode(state.wall_time_ms, state.counter).is_ok() =>
136            {
137                state
138            }
139            _ => HybridLogicalClockState {
140                wall_time_ms: 0,
141                counter: 0,
142                device_id: device_id.clone(),
143            },
144        };
145
146        Ok(Self {
147            device_id,
148            physical_clock,
149            state,
150        })
151    }
152
153    /// Returns the next local timestamp and advances the clock.
154    ///
155    /// # Errors
156    ///
157    /// Returns [`HlcError::InvalidPhysicalClock`] for an invalid clock reading.
158    /// Returns [`HlcError::ExceedsSafeInteger`] when no later safe timestamp is
159    /// available. State remains unchanged on error.
160    pub fn now(&mut self) -> Result<i64, HlcError> {
161        let physical = self.read_physical_time()?;
162        let wall_time_ms = physical.max(self.state.wall_time_ms);
163        let counter = if physical > self.state.wall_time_ms {
164            0
165        } else {
166            self.state.counter + 1
167        };
168        self.advance(wall_time_ms, counter)
169    }
170
171    /// Observes a remote timestamp and returns a timestamp ordered after it.
172    ///
173    /// # Errors
174    ///
175    /// Returns [`HlcError::InvalidTimestamp`] for invalid remote input,
176    /// [`HlcError::InvalidPhysicalClock`] for an invalid local clock reading,
177    /// or [`HlcError::ExceedsSafeInteger`] when no later timestamp is available.
178    /// State remains unchanged on error.
179    pub fn observe(&mut self, remote_timestamp: i64) -> Result<i64, HlcError> {
180        let (remote_wall_time_ms, remote_counter) = decode(remote_timestamp)?;
181        let physical = self.read_physical_time()?;
182        let wall_time_ms = physical
183            .max(self.state.wall_time_ms)
184            .max(remote_wall_time_ms);
185        let counter =
186            if wall_time_ms == self.state.wall_time_ms && wall_time_ms == remote_wall_time_ms {
187                self.state.counter.max(remote_counter) + 1
188            } else if wall_time_ms == self.state.wall_time_ms {
189                self.state.counter + 1
190            } else if wall_time_ms == remote_wall_time_ms {
191                remote_counter + 1
192            } else {
193                0
194            };
195        self.advance(wall_time_ms, counter)
196    }
197
198    /// Returns an owned snapshot suitable for caller-controlled persistence.
199    #[must_use]
200    pub fn snapshot(&self) -> HybridLogicalClockState {
201        self.state.clone()
202    }
203
204    fn read_physical_time(&mut self) -> Result<i64, HlcError> {
205        let value = (self.physical_clock)();
206        if !(0..=MAX_ENCODED_TIMESTAMP).contains(&value) {
207            return Err(HlcError::InvalidPhysicalClock);
208        }
209        Ok(value)
210    }
211
212    fn advance(&mut self, mut wall_time_ms: i64, mut counter: i64) -> Result<i64, HlcError> {
213        if counter >= COUNTERS_PER_MILLISECOND {
214            wall_time_ms = wall_time_ms
215                .checked_add(1)
216                .ok_or(HlcError::ExceedsSafeInteger)?;
217            counter = 0;
218        }
219
220        let timestamp = encode(wall_time_ms, counter)?;
221        self.state = HybridLogicalClockState {
222            wall_time_ms,
223            counter,
224            device_id: self.device_id.clone(),
225        };
226        Ok(timestamp)
227    }
228}
229
230#[cfg(test)]
231mod tests {
232    use std::{cell::Cell, cmp::Ordering};
233
234    use serde::Deserialize;
235
236    use super::{
237        HlcError, HybridLogicalClock, HybridLogicalClockState, MAX_ENCODED_TIMESTAMP, compare,
238        decode, encode,
239    };
240
241    #[derive(Deserialize)]
242    #[serde(rename_all = "camelCase")]
243    struct Vectors {
244        encode: Vec<EncodeVector>,
245        compare: Vec<CompareVector>,
246        merge: Vec<MergeVector>,
247    }
248
249    #[derive(Deserialize)]
250    #[serde(rename_all = "camelCase")]
251    struct EncodeVector {
252        wall_time_ms: i64,
253        counter: i64,
254        timestamp: i64,
255    }
256
257    #[derive(Deserialize)]
258    struct CompareVector {
259        left: i64,
260        right: i64,
261        ordering: i8,
262    }
263
264    #[derive(Deserialize)]
265    #[serde(rename_all = "camelCase")]
266    struct MergeVector {
267        device_id: String,
268        initial: HybridLogicalClockState,
269        physical_time_ms: i64,
270        remote_timestamp: i64,
271        expected_timestamp: i64,
272        expected_state: HybridLogicalClockState,
273    }
274
275    #[test]
276    fn advances_when_physical_time_stalls_or_moves_backward() {
277        let physical = Cell::new(1_000);
278        let mut clock = HybridLogicalClock::open("device-a", || physical.get(), None)
279            .expect("clock should open");
280
281        let first = clock.now().expect("first timestamp");
282        let stalled = clock.now().expect("stalled timestamp");
283        physical.set(900);
284        let backward = clock.now().expect("backward timestamp");
285
286        assert!(first < stalled && stalled < backward);
287        assert_eq!(decode(backward), Ok((1_000, 2)));
288    }
289
290    #[test]
291    fn rolls_the_logical_counter_into_the_next_millisecond() {
292        let state = HybridLogicalClockState {
293            wall_time_ms: 1_000,
294            counter: 999,
295            device_id: "device-a".to_owned(),
296        };
297        let mut clock =
298            HybridLogicalClock::open("device-a", || 1_000, Some(state)).expect("clock should open");
299
300        assert_eq!(clock.now(), Ok(1_001_001));
301        assert_eq!(
302            clock.snapshot(),
303            HybridLogicalClockState {
304                wall_time_ms: 1_001,
305                counter: 0,
306                device_id: "device-a".to_owned(),
307            }
308        );
309    }
310
311    #[test]
312    fn restores_matching_state_and_rejects_corrupt_state() {
313        let restored = HybridLogicalClockState {
314            wall_time_ms: 2_000,
315            counter: 7,
316            device_id: "device-a".to_owned(),
317        };
318        let mut clock = HybridLogicalClock::open("device-a", || 1_000, Some(restored))
319            .expect("clock should open");
320        assert_eq!(clock.now(), Ok(2_000_009));
321
322        let corrupt = HybridLogicalClockState {
323            wall_time_ms: MAX_ENCODED_TIMESTAMP,
324            counter: 0,
325            device_id: "device-a".to_owned(),
326        };
327        let mut reset = HybridLogicalClock::open("device-a", || 10, Some(corrupt))
328            .expect("corrupt state should reset");
329        assert_eq!(reset.now(), Ok(10_001));
330    }
331
332    #[test]
333    fn observes_remote_time_and_preserves_state_on_invalid_input() {
334        let mut clock =
335            HybridLogicalClock::open("device-a", || 900, None).expect("clock should open");
336
337        assert_eq!(clock.observe(1_000_006), Ok(1_000_007));
338        let snapshot = clock.snapshot();
339        assert_eq!(clock.observe(0), Err(HlcError::InvalidTimestamp));
340        assert_eq!(clock.snapshot(), snapshot);
341        assert_eq!(compare(0, 1), None);
342    }
343
344    #[test]
345    fn rejects_blank_device_ids_and_invalid_physical_time() {
346        assert!(matches!(
347            HybridLogicalClock::open("   ", || 0, None),
348            Err(HlcError::EmptyDeviceId)
349        ));
350
351        let mut clock =
352            HybridLogicalClock::open("device-a", || -1, None).expect("clock should open");
353        assert_eq!(clock.now(), Err(HlcError::InvalidPhysicalClock));
354        assert_eq!(
355            clock.snapshot(),
356            HybridLogicalClockState {
357                wall_time_ms: 0,
358                counter: 0,
359                device_id: "device-a".to_owned(),
360            }
361        );
362    }
363
364    #[test]
365    fn supports_the_maximum_encoded_value_and_reports_exhaustion() {
366        assert_eq!(encode(9_007_199_254_740, 990), Ok(MAX_ENCODED_TIMESTAMP));
367        assert_eq!(decode(MAX_ENCODED_TIMESTAMP), Ok((9_007_199_254_740, 990)));
368        assert_eq!(
369            encode(9_007_199_254_740, 991),
370            Err(HlcError::ExceedsSafeInteger)
371        );
372
373        let state = HybridLogicalClockState {
374            wall_time_ms: 9_007_199_254_740,
375            counter: 990,
376            device_id: "device-a".to_owned(),
377        };
378        let mut clock = HybridLogicalClock::open("device-a", || 0, Some(state))
379            .expect("maximum state should open");
380        assert_eq!(clock.now(), Err(HlcError::ExceedsSafeInteger));
381        assert_eq!(
382            encode(clock.snapshot().wall_time_ms, clock.snapshot().counter),
383            Ok(MAX_ENCODED_TIMESTAMP)
384        );
385    }
386
387    #[test]
388    fn matches_shared_cross_runtime_vectors() {
389        let vectors: Vectors =
390            serde_json::from_str(include_str!("../../../../fixtures/hlc/vectors-v1.json"))
391                .expect("shared HLC vectors must be valid JSON");
392
393        for vector in vectors.encode {
394            assert_eq!(
395                encode(vector.wall_time_ms, vector.counter),
396                Ok(vector.timestamp)
397            );
398            assert_eq!(
399                decode(vector.timestamp),
400                Ok((vector.wall_time_ms, vector.counter))
401            );
402        }
403        for vector in vectors.compare {
404            let expected = match vector.ordering {
405                -1 => Ordering::Less,
406                0 => Ordering::Equal,
407                1 => Ordering::Greater,
408                value => panic!("invalid vector ordering {value}"),
409            };
410            assert_eq!(compare(vector.left, vector.right), Some(expected));
411        }
412        for vector in vectors.merge {
413            let physical_time_ms = vector.physical_time_ms;
414            let mut clock = HybridLogicalClock::open(
415                vector.device_id,
416                move || physical_time_ms,
417                Some(vector.initial),
418            )
419            .expect("vector clock state should open");
420            assert_eq!(
421                clock.observe(vector.remote_timestamp),
422                Ok(vector.expected_timestamp)
423            );
424            assert_eq!(clock.snapshot(), vector.expected_state);
425        }
426    }
427}
baukit_sync/hlc.rs

baukit_sync/
hlc.rs
