From df3f11a337955dd703199cc2b037d4bfb698ceba Mon Sep 17 00:00:00 2001 From: tv Date: Sat, 31 Oct 2026 06:53:00 +0100 Subject: add peer clock skew warning --- docs/openapi.yaml | 20 +++++++ src/api/stats.rs | 16 ++++- src/api/stats/prometheus.rs | 22 +++++++ src/config/mod.rs | 15 +++++ src/main.rs | 1 + src/node.rs | 3 + src/simtest/mod.rs | 1 + src/state/mod.rs | 5 ++ src/swarm/mod.rs | 1 + src/swarm/peer/worker.rs | 1 + src/swarm/skew.rs | 141 ++++++++++++++++++++++++++++++++++++++++++++ 11 files changed, 225 insertions(+), 1 deletion(-) create mode 100644 src/swarm/skew.rs diff --git a/docs/openapi.yaml b/docs/openapi.yaml index d1f5935..1b20ee9 100644 --- a/docs/openapi.yaml +++ b/docs/openapi.yaml @@ -1802,6 +1802,26 @@ components: type: [integer, 'null'] count: type: integer + clock: + type: object + description: This node's clock against its active peers' (`measurements.md`, Clocks). + required: [peers, offsetMs, behind, ahead, off] + properties: + peers: + type: integer + description: Active peers whose clocks were compared. + offsetMs: + type: [integer, 'null'] + description: How far ours is ahead of theirs (behind if negative), the median. + behind: + type: integer + description: Peers behind ours by more than --clock-skew-warn, for sure. + ahead: + type: integer + description: Peers ahead of ours by more than --clock-skew-warn, for sure. + off: + type: boolean + description: Ours differs from more than half of theirs that way. LockWaits: type: object diff --git a/src/api/stats.rs b/src/api/stats.rs index 13610dd..d274fd9 100644 --- a/src/api/stats.rs +++ b/src/api/stats.rs @@ -13,9 +13,10 @@ use std::collections::BTreeMap; use serde::Serialize; -use super::peers::{PeerListResponse, peer_list}; +use super::peers::{PeerListResponse, PeerLivenessStatus, peer_list}; use crate::document::replica::EpochRecord; use crate::state::{AppState, Incidents}; +use crate::swarm::skew::{self, Skew}; use crate::swarm::{members, stability, topology}; use crate::transport::stats::TransportStats; use crate::transport::traffic::PeerTraffic; @@ -44,6 +45,8 @@ pub(crate) struct StatsResponse { /// The peers listed in /peers together (a forgotten peer's are gone). incidents: Incidents, propagation: PropagationStats, + /// This node's clock against its active peers' (`swarm::skew`). + clock: Skew, /// This node's own files (`files`), by kind. disk: BTreeMap<&'static str, DiskStats>, /// How the transports find and reach peers (`Transport::stats`). @@ -271,6 +274,16 @@ pub(crate) async fn collect(state: &AppState, list: &PeerListResponse) -> StatsR let (connected_fraction, observed) = stability::measured(state); + let clock = skew::of( + &list + .peers + .values() + .filter(|peer| !matches!(peer.status, PeerLivenessStatus::Dormant)) + .filter_map(|peer| peer.clock) + .collect::>(), + state.clock_skew_warn, + ); + let mut incidents = Incidents::default(); let mut last_from_neighbour_secs: Option = None; let mut diverged_peers = 0; @@ -347,6 +360,7 @@ pub(crate) async fn collect(state: &AppState, list: &PeerListResponse) -> StatsR stale_hello_replies: node.stale_hello_replies, traffic: state.traffic.total(), incidents, + clock, propagation: PropagationStats { count: node.propagation.count, sum_ms: node.propagation.sum_ms, diff --git a/src/api/stats/prometheus.rs b/src/api/stats/prometheus.rs index 2f33bef..6774e74 100644 --- a/src/api/stats/prometheus.rs +++ b/src/api/stats/prometheus.rs @@ -763,6 +763,28 @@ pub(super) fn render(stats: &StatsResponse, list: &PeerListResponse) -> String { propagation.ahead as f64, ); + let clock = &stats.clock; + out.single( + "clock_peers_compared", + "gauge", + "Active peers whose clocks were compared with ours (from their PONGs).", + clock.peers as f64, + ); + if let Some(offset_ms) = clock.offset_ms { + out.single( + "clock_offset_seconds", + "gauge", + "How far this node's clock is ahead of its peers' (behind: negative), the median.", + offset_ms as f64 / 1000.0, + ); + } + out.single( + "clock_off", + "gauge", + "1 while this node's clock differs from most of its peers' by more than --clock-skew-warn.", + flag(clock.off), + ); + // Node totals, including peers no longer listed. let totals = traffic_series(&stats.traffic); out.family( diff --git a/src/config/mod.rs b/src/config/mod.rs index 37c3ce6..2dfefcf 100644 --- a/src/config/mod.rs +++ b/src/config/mod.rs @@ -249,6 +249,13 @@ pub(crate) const OPTIONS: &[OptionSpec] = &[ Some("7d"), "Members that renewed their alive stamp within this time vote on remakes (a majority of them must agree). Matters on the node that coordinates a remake.", ), + option( + "clock-skew-warn", + Kind::Duration, + "DURATION", + Some("2s"), + "Warn in the log when this node's clock differs from most of its peers' by more than this, for sure (beyond half the round trip it was measured over, from their PONGs); 0: never.", + ), option( "blob-chunk", Kind::Size, @@ -467,6 +474,11 @@ pub(crate) struct Settings { with = "crate::units::duration_option" )] pub(crate) electorate_window: Duration, + #[serde( + default = "default_clock_skew_warn", + with = "crate::units::duration_option" + )] + pub(crate) clock_skew_warn: Duration, /// In bytes. #[serde(default = "default_blob_chunk", with = "crate::units::size_option")] pub(crate) blob_chunk: u64, @@ -539,6 +551,9 @@ fn default_neighbours() -> usize { fn default_electorate_window() -> Duration { Duration::from_secs(7 * 24 * 3600) } +fn default_clock_skew_warn() -> Duration { + Duration::from_secs(2) +} fn default_blob_chunk() -> u64 { 64 * 1024 } diff --git a/src/main.rs b/src/main.rs index d282158..f5bd5d4 100644 --- a/src/main.rs +++ b/src/main.rs @@ -210,6 +210,7 @@ async fn main() -> Result<()> { neighbours_k: args.neighbours, alias: args.alias.clone(), electorate_window: args.electorate_window, + clock_skew_warn: args.clock_skew_warn, // At most 4MiB (`Settings::check`). blob_chunk_bytes: args.blob_chunk as u32, swarm_key, diff --git a/src/node.rs b/src/node.rs index 9f1bd6b..e70b205 100644 --- a/src/node.rs +++ b/src/node.rs @@ -45,6 +45,7 @@ pub(crate) struct NodeConfig { pub(crate) neighbours_k: usize, pub(crate) alias: Option, pub(crate) electorate_window: Duration, + pub(crate) clock_skew_warn: Duration, pub(crate) blob_chunk_bytes: u32, pub(crate) swarm_key: Option>, pub(crate) protocol_writers: HashSet, @@ -114,6 +115,8 @@ impl Node { neighbours_k: config.neighbours_k, alias: config.alias, electorate_window: config.electorate_window, + clock_skew_warn: config.clock_skew_warn, + clock_skew_warned: 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/simtest/mod.rs b/src/simtest/mod.rs index f191946..4dedc4b 100644 --- a/src/simtest/mod.rs +++ b/src/simtest/mod.rs @@ -113,6 +113,7 @@ impl Sim { neighbours_k: 8, alias: Some(id.to_owned()), electorate_window: Duration::from_secs(30 * 24 * 3600), + clock_skew_warn: Duration::from_secs(2), blob_chunk_bytes: 64 * 1024, swarm_key: None, protocol_writers: HashSet::new(), diff --git a/src/state/mod.rs b/src/state/mod.rs index a2d5d74..cf50b2a 100644 --- a/src/state/mod.rs +++ b/src/state/mod.rs @@ -113,6 +113,11 @@ pub(crate) struct AppState { pub(crate) alias: Option, /// Members alive within this window vote on remakes (`--electorate-window`). pub(crate) electorate_window: Duration, + /// This node's clock is off when it differs from most of its peers' by more than + /// this (`--clock-skew-warn`, `swarm::skew`); 0: never. + pub(crate) clock_skew_warn: Duration, + /// When the log last said this node's clock is off, while it is (`swarm::skew`). + pub(crate) clock_skew_warned: Arc>>, /// Chunked transfers served and fetched (`swarm::blob`). pub(crate) blobs: Arc, /// LINKS asked and answered, for a map of the swarm (`swarm::links`). diff --git a/src/swarm/mod.rs b/src/swarm/mod.rs index f36af9c..8259ed5 100644 --- a/src/swarm/mod.rs +++ b/src/swarm/mod.rs @@ -12,6 +12,7 @@ pub(crate) mod pair; pub(crate) mod peer; pub(crate) mod relay; pub(crate) mod rtt; +pub(crate) mod skew; pub(crate) mod source; pub(crate) mod stability; pub(crate) mod state_file; diff --git a/src/swarm/peer/worker.rs b/src/swarm/peer/worker.rs index 797f091..9b62d24 100644 --- a/src/swarm/peer/worker.rs +++ b/src/swarm/peer/worker.rs @@ -110,6 +110,7 @@ pub(crate) async fn run_peer_worker(state: AppState) { forget_strangers(&state).await; crate::swarm::members::ensure_registered(&state).await; crate::swarm::stability::tick(&state).await; + crate::swarm::skew::check(&state).await; crate::swarm::stats::measure_saved_size(&state); crate::swarm::source::warn_if_stalled(&state); crate::swarm::state_file::update(&state, false).await; 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, + /// 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 = 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()); + } +} -- cgit v1.3.1