hlc.rs 100.0%

31 of 31 lines run, 6 of 8 functions · run #17 on main · run missed

//! The hybrid logical clock on every event (spec §9.1), with Oxygen's drift
//! bound (oxygen spec §6.4): observing a remote time never moves this clock
//! more than five minutes ahead of the local wall clock.

use std::sync::Mutex;

/// How far ahead of the local clock an observed time may move ours.
pub const MAX_DRIFT_MS: u64 = 5 * 60 * 1000;

/// A hybrid logical clock reading: wall milliseconds and a counter.
#[derive(Clone, Copy, Debug, Default, PartialEq, Eq, PartialOrd, Ord)]
pub struct Hlc {
    pub wall_ms: u64,
    pub logical: u32,
}

/// The clock: monotonic across calls, whatever the wall clock does.
#[derive(Debug, Default)]
pub struct Clock {
    last: Mutex<Hlc>,
}

impl Clock {
    /// A clock that continues from `last` (the ledger's newest reading).
    pub fn starting_at(last: Hlc) -> Clock {
        Clock { last: Mutex::new(last) }
    }

    /// The next reading for a local event, at wall time `now_ms`.
    pub fn tick(&self, now_ms: u64) -> Hlc {
        let mut last = self.last.lock().unwrap();
        *last = if now_ms > last.wall_ms {
            Hlc { wall_ms: now_ms, logical: 0 }
        } else {
            Hlc { wall_ms: last.wall_ms, logical: last.logical.saturating_add(1) }
        };
        *last
    }

    /// Take a remote reading into account, as on receiving another
    /// Spectra's event; the remote reading is kept as it is, only this
    /// clock's own advance is bounded.
    pub fn observe(&self, remote: Hlc, now_ms: u64) {
        let mut last = self.last.lock().unwrap();
        let bound = now_ms.saturating_add(MAX_DRIFT_MS);
        let remote = if remote.wall_ms > bound { Hlc { wall_ms: bound, logical: 0 } } else { remote };
        if remote > *last {
            *last = remote;
        }
    }
}

#[cfg(test)]
mod tests {
    use super::*;

    #[test]
    fn ticks_are_monotonic_when_the_wall_clock_stalls_or_steps_back() {
        let c = Clock::default();
        let a = c.tick(1_000);
        let b = c.tick(1_000);
        let d = c.tick(500);
        assert!(a < b && b < d);
        assert_eq!(c.tick(2_000), Hlc { wall_ms: 2_000, logical: 0 });
    }

    #[test]
    fn a_far_future_remote_time_moves_the_clock_by_the_drift_bound_at_most() {
        let c = Clock::default();
        c.observe(Hlc { wall_ms: 10_000_000_000, logical: 7 }, 1_000);
        assert_eq!(c.tick(1_000).wall_ms, 1_000 + MAX_DRIFT_MS);
    }
}

1		//! The hybrid logical clock on every event (spec §9.1), with Oxygen's drift
2		//! bound (oxygen spec §6.4): observing a remote time never moves this clock
3		//! more than five minutes ahead of the local wall clock.
4
5		use std::sync::Mutex;
6
7		/// How far ahead of the local clock an observed time may move ours.
8		pub const MAX_DRIFT_MS: u64 = 5 * 60 * 1000;
9
10		/// A hybrid logical clock reading: wall milliseconds and a counter.
11		#[derive(Clone, Copy, Debug, Default, PartialEq, Eq, PartialOrd, Ord)]
12		pub struct Hlc {
13		pub wall_ms: u64,
14		pub logical: u32,
15		}
16
17		/// The clock: monotonic across calls, whatever the wall clock does.
18		#[derive(Debug, Default)]
19		pub struct Clock {
20		last: Mutex<Hlc>,
21		}
22
23		impl Clock {
24		/// A clock that continues from `last` (the ledger's newest reading).
25	148×	pub fn starting_at(last: Hlc) -> Clock {
26	148×	Clock { last: Mutex::new(last) }
27	148×	}
28
29		/// The next reading for a local event, at wall time `now_ms`.
30	12147×	pub fn tick(&self, now_ms: u64) -> Hlc {
31	12147×	let mut last = self.last.lock().unwrap();
32	12147×	*last = if now_ms > last.wall_ms {
33	2182×	Hlc { wall_ms: now_ms, logical: 0 }
34		} else {
35	9965×	Hlc { wall_ms: last.wall_ms, logical: last.logical.saturating_add(1) }
36		};
37	12147×	*last
38	12147×	}
39
40		/// Take a remote reading into account, as on receiving another
41		/// Spectra's event; the remote reading is kept as it is, only this
42		/// clock's own advance is bounded.
43	1×	pub fn observe(&self, remote: Hlc, now_ms: u64) {
44	1×	let mut last = self.last.lock().unwrap();
45	1×	let bound = now_ms.saturating_add(MAX_DRIFT_MS);
46	1×	let remote = if remote.wall_ms > bound { Hlc { wall_ms: bound, logical: 0 } } else { remote };
47	1×	if remote > *last {
48	1×	*last = remote;
49	1×	}
50	1×	}
51		}
52
53		#[cfg(test)]
54		mod tests {
55		use super::*;
56
57		#[test]
58	1×	fn ticks_are_monotonic_when_the_wall_clock_stalls_or_steps_back() {
59	1×	let c = Clock::default();
60	1×	let a = c.tick(1_000);
61	1×	let b = c.tick(1_000);
62	1×	let d = c.tick(500);
63	1×	assert!(a < b && b < d);
64	1×	assert_eq!(c.tick(2_000), Hlc { wall_ms: 2_000, logical: 0 });
65	1×	}
66
67		#[test]
68	1×	fn a_far_future_remote_time_moves_the_clock_by_the_drift_bound_at_most() {
69	1×	let c = Clock::default();
70	1×	c.observe(Hlc { wall_ms: 10_000_000_000, logical: 7 }, 1_000);
71	1×	assert_eq!(c.tick(1_000).wall_ms, 1_000 + MAX_DRIFT_MS);
72	1×	}
73		}

`/`	Search
`g` `c`	Code
`g` `h`	History
`g` `b`	Bookmarks
`g` `t`	Trust
`y`	Permalink to this state (files and trees)
`?`	This sheet

Coverage / crates/spectra-core/src/hlc.rs 100.0%

Keyboard