diff options
Diffstat (limited to 'src')
| -rw-r--r-- | src/node.rs | 2 | ||||
| -rw-r--r-- | src/state/mod.rs | 4 | ||||
| -rw-r--r-- | src/swarm/skew.rs | 143 |
3 files changed, 137 insertions, 12 deletions
diff --git a/src/node.rs b/src/node.rs index ac4c252..42b3c91 100644 --- a/src/node.rs +++ b/src/node.rs @@ -117,7 +117,7 @@ impl Node { electorate_window: config.electorate_window, clock_skew_warn: config.clock_skew_warn, clock_ahead_ms: Default::default(), - clock_skew_warned: Default::default(), + clock_skew: Default::default(), blobs: Arc::new(swarm::blob::Blobs::new(config.blob_chunk_bytes)), links: Default::default(), protocol_writers: Arc::new(config.protocol_writers), diff --git a/src/state/mod.rs b/src/state/mod.rs index ec8823a..36dd4ad 100644 --- a/src/state/mod.rs +++ b/src/state/mod.rs @@ -119,8 +119,8 @@ pub(crate) struct AppState { /// How far this node's clock is ahead of its peers' (their median, `swarm::skew`), as /// last measured; 0 before. The propagation figures take it off. pub(crate) clock_ahead_ms: Arc<std::sync::atomic::AtomicI64>, - /// When the log last said this node's clock is off, while it is (`swarm::skew`). - pub(crate) clock_skew_warned: Arc<std::sync::Mutex<Option<tokio::time::Instant>>>, + /// What the clock checks remember between them (`swarm::skew`). + pub(crate) clock_skew: Arc<std::sync::Mutex<crate::swarm::skew::Watch>>, /// Chunked transfers served and fetched (`swarm::blob`). pub(crate) blobs: Arc<crate::swarm::blob::Blobs>, /// LINKS asked and answered, for a map of the swarm (`swarm::links`). diff --git a/src/swarm/skew.rs b/src/swarm/skew.rs index f5d9858..69c85f5 100644 --- a/src/swarm/skew.rs +++ b/src/swarm/skew.rs @@ -1,11 +1,15 @@ //! 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`). +//! sure, then the log says so (every `WARN_EVERY` while it lasts). A clock set anew +//! between two checks (by NTP, a resumed machine, by hand) by more than the limit shows +//! as a jump, this node's when most peers' clocks moved alike, else those peers'; the +//! log says so too. 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. +//! Two nodes alone can't tell whose clock is off, or jumped: each says it is its own. + +use std::collections::HashMap; use std::time::Duration; @@ -65,32 +69,122 @@ pub(crate) fn of(offsets: &[ClockOffset], limit: Duration) -> Skew { } } -/// Ours against the clocks of the peers that are active now. -pub(crate) async fn measure(state: &AppState) -> Skew { +/// The clocks of the peers that are active now, by peer. +async fn offsets(state: &AppState) -> Vec<(String, ClockOffset)> { 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); + offsets.push((peer, offset)); } } + offsets +} + +/// Ours against the clocks of the peers that are active now. +pub(crate) async fn measure(state: &AppState) -> Skew { + let offsets: Vec<ClockOffset> = offsets(state).await.into_iter().map(|(_, o)| o).collect(); of(&offsets, state.clock_skew_warn) } +/// What the checks remember between them (`AppState::clock_skew`). +#[derive(Debug, Default)] +pub(crate) struct Watch { + /// When the log last said this node's clock is off, while it is. + warned: Option<Instant>, + /// Each peer's clock at the last check. + last: HashMap<String, ClockOffset>, +} + +/// Clocks set anew between two checks, by more than the limit for sure. +#[derive(Debug, Default, PartialEq, Eq)] +struct Jumps { + /// Ours, by how much (forward if positive): most peers' clocks moved alike. + ours: Option<i64>, + /// Otherwise the peers' whose clocks moved, each by how much. + peers: Vec<(String, i64)>, +} + +/// Which clocks jumped from `last` to `now` (each peer's ahead of ours), by more than +/// `limit_ms` beyond both bounds. None with a limit of 0. +fn jumps( + last: &HashMap<String, ClockOffset>, + now: &[(String, ClockOffset)], + limit_ms: u64, +) -> Jumps { + if limit_ms == 0 { + return Jumps::default(); + } + let moves: Vec<(&String, i64, bool)> = now + .iter() + .filter_map(|(peer, offset)| { + let before = last.get(peer)?; + let moved = offset.offset_ms - before.offset_ms; + let bound = offset + .within_ms + .saturating_add(before.within_ms) + .saturating_add(limit_ms); + Some((peer, moved, moved.unsigned_abs() > bound)) + }) + .collect(); + let forward = moves + .iter() + .filter(|(_, moved, sure)| *sure && *moved > 0) + .count(); + let back = moves + .iter() + .filter(|(_, moved, sure)| *sure && *moved < 0) + .count(); + if forward * 2 > moves.len() || back * 2 > moves.len() { + // Ours moved the other way, by the median. + let mut ours: Vec<i64> = moves.iter().map(|(_, moved, _)| -moved).collect(); + ours.sort_unstable(); + return Jumps { + ours: ours.get(ours.len() / 2).copied(), + peers: Vec::new(), + }; + } + Jumps { + ours: None, + peers: moves + .into_iter() + .filter(|(_, _, sure)| *sure) + .map(|(peer, moved, _)| (peer.clone(), moved)) + .collect(), + } +} + /// 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; + let offsets = offsets(state).await; + let skew = of( + &offsets.iter().map(|(_, o)| *o).collect::<Vec<_>>(), + state.clock_skew_warn, + ); 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(); + let mut watch = state.clock_skew.lock_or_recover(); + let jumped = jumps(&watch.last, &offsets, skew.limit_ms); + if let Some(ms) = jumped.ours { + warn!( + jump_ms = ms, + peers = offsets.len(), + "This node's clock jumped (set anew)" + ); + } + for (peer, ms) in &jumped.peers { + info!(peer = %peer, jump_ms = ms, "A peer's clock jumped (set anew)"); + } + watch.last = offsets.into_iter().collect(); + let warned = &mut watch.warned; if skew.off { if warned.is_none_or(|at| now.saturating_duration_since(at) >= WARN_EVERY) { warn!( @@ -149,4 +243,35 @@ mod tests { let none = of(&[], limit); assert_eq!((none.peers, none.offset_ms, none.off), (0, None, false)); } + + /// Most peers' clocks moved alike: ours jumped; a few moved: theirs. + #[test] + fn jumps_are_ours_when_most_moved_alike() { + let peers = |offsets: &[i64]| -> Vec<(String, ClockOffset)> { + offsets + .iter() + .enumerate() + .map(|(i, &ms)| (format!("p{i}"), offset(ms, 50))) + .collect() + }; + let last: HashMap<_, _> = peers(&[0, 10, -20]).into_iter().collect(); + // Ours went 30 s forward: theirs are 30 s behind it now. + let ours = jumps(&last, &peers(&[-30_000, -29_990, -30_020]), 2_000); + assert_eq!(ours.ours, Some(30_000)); + assert!(ours.peers.is_empty()); + // One peer's went 10 s back. + let one = jumps(&last, &peers(&[-10_000, 10, -20]), 2_000); + assert_eq!(one.ours, None); + assert_eq!(one.peers, vec![("p0".to_owned(), -10_000)]); + // Within the bounds and the limit: nothing. + assert_eq!( + jumps(&last, &peers(&[1_000, 10, -20]), 2_000), + Jumps::default() + ); + // A limit of 0: never. + assert_eq!( + jumps(&last, &peers(&[-30_000, -30_000, -30_000]), 0), + Jumps::default() + ); + } } |