1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
|
//! Whether this node's clock agrees with its peers' (`Transport::clock_offset`, from
//! their PONGs). Each peer's clock is known within half a round trip; this node's own
//! is off when it differs from most of its peers' by more than `--clock-skew-warn` for
//! sure, then the log says so (every `WARN_EVERY` while it lasts). Clocks are the
//! operating system's to set (NTP); the propagation figures only take this node's own
//! offset off (`AppState::clock_ahead_ms`).
//!
//! Two nodes alone can't tell whose clock is off: each warns.
use std::time::Duration;
use serde::Serialize;
use tokio::time::Instant;
use tracing::{info, warn};
use crate::locks::MutexExt;
use crate::state::AppState;
use crate::transport::paths::ClockOffset;
/// While this node's clock is off, the log says so this often.
const WARN_EVERY: Duration = Duration::from_secs(600);
/// How this node's clock compares with its peers'.
#[derive(Clone, Copy, Debug, Default, PartialEq, Eq, Serialize)]
#[serde(rename_all = "camelCase")]
pub(crate) struct Skew {
/// Peers whose clocks were compared (active ones with a PONG lately).
pub(crate) peers: usize,
/// How far this node's clock is ahead of theirs (behind if negative): the median.
pub(crate) offset_ms: Option<i64>,
/// Peers whose clocks are behind ours by more than the limit for sure, and ahead.
pub(crate) behind: usize,
pub(crate) ahead: usize,
/// Whether ours is off: more than half of them differ from it beyond the limit,
/// the same way. Never with a limit of 0.
pub(crate) off: bool,
/// The limit (`--clock-skew-warn`).
pub(crate) limit_ms: u64,
}
/// Ours against the peers' clocks `offsets` (each ahead of ours), with `limit`.
pub(crate) fn of(offsets: &[ClockOffset], limit: Duration) -> Skew {
let limit_ms = u64::try_from(limit.as_millis()).unwrap_or(u64::MAX);
let beyond = |offset_ms: i64, within_ms: u64| {
offset_ms > 0 && offset_ms.unsigned_abs() > within_ms.saturating_add(limit_ms)
};
let ahead = offsets
.iter()
.filter(|o| beyond(o.offset_ms, o.within_ms))
.count();
let behind = offsets
.iter()
.filter(|o| beyond(-o.offset_ms, o.within_ms))
.count();
let mut ours: Vec<i64> = offsets.iter().map(|o| -o.offset_ms).collect();
ours.sort_unstable();
let peers = offsets.len();
Skew {
peers,
offset_ms: ours.get(peers / 2).copied(),
behind,
ahead,
off: limit_ms > 0 && (ahead * 2 > peers || behind * 2 > peers),
limit_ms,
}
}
/// Ours against the clocks of the peers that are active now.
pub(crate) async fn measure(state: &AppState) -> Skew {
let mut offsets = Vec::new();
for (peer, peer_state) in state.peers_snapshot().await {
if peer_state.lock().await.liveness.is_dormant() {
continue;
}
if let Some(offset) = state.transport.clock_offset(&peer) {
offsets.push(offset);
}
}
of(&offsets, state.clock_skew_warn)
}
/// Notes how far this node's clock is ahead of its peers' (`AppState::clock_ahead_ms`),
/// and says in the log when it is off, and when it agrees again. From the peer worker,
/// every few seconds.
pub(crate) async fn check(state: &AppState) {
let skew = measure(state).await;
if let Some(offset_ms) = skew.offset_ms {
state
.clock_ahead_ms
.store(offset_ms, std::sync::atomic::Ordering::Relaxed);
}
let now = Instant::now();
let mut warned = state.clock_skew_warned.lock_or_recover();
if skew.off {
if warned.is_none_or(|at| now.saturating_duration_since(at) >= WARN_EVERY) {
warn!(
offset_ms = skew.offset_ms,
peers = skew.peers,
behind = skew.behind,
ahead = skew.ahead,
limit_ms = state.clock_skew_warn.as_millis() as u64,
"This node's clock differs from most of its peers'; check its time synchronisation (NTP)"
);
*warned = Some(now);
}
} else if warned.take().is_some() {
info!(
offset_ms = skew.offset_ms,
peers = skew.peers,
"This node's clock agrees with its peers' again"
);
}
}
#[cfg(test)]
mod tests {
use super::*;
fn offset(offset_ms: i64, within_ms: u64) -> ClockOffset {
ClockOffset {
offset_ms,
within_ms,
}
}
/// Off only beyond the limit and the bound, and only against most peers.
#[test]
fn off_against_most_peers_for_sure() {
let limit = Duration::from_secs(2);
// Every peer 5 s ahead: ours is 5 s behind.
let behind = of(
&[offset(5_000, 100), offset(5_100, 50), offset(4_900, 300)],
limit,
);
assert_eq!(behind.offset_ms, Some(-5_000));
assert_eq!((behind.ahead, behind.behind, behind.off), (3, 0, true));
// One peer off: that peer's clock, not ours.
let one = of(
&[offset(5_000, 100), offset(10, 50), offset(-20, 30)],
limit,
);
assert_eq!((one.ahead, one.off), (1, false));
// Within the bound: not for sure.
let unsure = of(&[offset(2_500, 600), offset(2_500, 600)], limit);
assert!(!unsure.off);
// A limit of 0: never.
assert!(!of(&[offset(60_000, 10)], Duration::ZERO).off);
// No peers: nothing known.
let none = of(&[], limit);
assert_eq!((none.peers, none.offset_ms, none.off), (0, None, false));
}
}
|
