1 files changed, 134 insertions, 9 deletions

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()
+        );
+    }
 }