1 files changed, 141 insertions, 0 deletions
diff --git a/src/swarm/skew.rs b/src/swarm/skew.rs
new file mode 100644
index 0000000..1b98ff5
--- /dev/null
+++ b/src/swarm/skew.rs
@@ -0,0 +1,141 @@
+//! 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). A diagnostic only:
+//! setting clocks is the operating system's (NTP).
+//!
+//! 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,
+}
+
+/// 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),
+    }
+}
+
+/// 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)
+}
+
+/// Says in the log when this node's clock 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 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.
+        assert_eq!(of(&[], limit), Skew::default());
+    }
+}