From 88160e8aa273090c55103e7a92eeae9a8bdac9ed Mon Sep 17 00:00:00 2001 From: tv Date: Sat, 31 Oct 2026 06:51:17 +0100 Subject: calculate peer clock drift --- src/transport/identity.rs | 9 +++++ src/transport/mod.rs | 7 ++++ src/transport/paths.rs | 95 ++++++++++++++++++++++++++++++++++++++++++++--- src/transport/traffic.rs | 4 ++ 4 files changed, 110 insertions(+), 5 deletions(-) (limited to 'src/transport') diff --git a/src/transport/identity.rs b/src/transport/identity.rs index eb7bb93..a748505 100644 --- a/src/transport/identity.rs +++ b/src/transport/identity.rs @@ -603,6 +603,15 @@ impl Transport for IdentityTransport { self.shared.paths.rtt(&address) } + /// The narrowest of its proven paths'. + fn clock_offset(&self, peer: &str) -> Option { + self.shared + .proven_at(peer, false) + .iter() + .filter_map(|address| self.shared.paths.clock_offset(address)) + .min_by_key(|offset| offset.within_ms) + } + /// The identity layer's, and each transport's. fn stats(&self) -> TransportStats { let mut stats = self.shared.paths.stats(); diff --git a/src/transport/mod.rs b/src/transport/mod.rs index 9a82cbe..1ce4cd1 100644 --- a/src/transport/mod.rs +++ b/src/transport/mod.rs @@ -880,6 +880,13 @@ pub(crate) trait Transport: Send + Sync { None } + /// How far `peer`'s clock is ahead of ours, as its PONGs say (`paths::ClockOffset`): + /// the narrowest bound of its paths. By default unknown. + fn clock_offset(&self, peer: &str) -> Option { + let _ = peer; + None + } + /// Whether the transport answers PINGs with PONGs itself (the paths layer, /// `transport::paths`); if not (a bare transport, in tests), the node does. By /// default not. diff --git a/src/transport/paths.rs b/src/transport/paths.rs index fcfadaa..6512759 100644 --- a/src/transport/paths.rs +++ b/src/transport/paths.rs @@ -47,7 +47,7 @@ use anyhow::{Context, Result, anyhow}; use tokio::time::Instant; use tracing::{debug, info}; -use crate::clock::Clock; +use crate::clock::{Clock, UnixMillis}; use crate::locks::MutexExt; use crate::transport::stats::{PathCounts, PathInfo, PathsStats, TransportStats}; use crate::transport::traffic::Traffic; @@ -106,6 +106,35 @@ const SWITCH_GAIN: f64 = 0.25; const SWITCH_MIN: Duration = Duration::from_millis(5); /// Paths kept at most: they are of peers this node chose among, but bounded anyway. const MAX_PATHS: usize = 4096; +/// A path's clock sample with the narrowest bound is kept this long at most; then the +/// next one takes its place, so a clock that was set anew shows within this time. +const CLOCK_KEPT: Duration = Duration::from_secs(300); + +/// How far a peer's clock is ahead of ours (behind if negative), from a PING and its +/// PONG: `offset_ms`, give or take `within_ms`. The PONG says its sender's time when it +/// answered, which was after our PING went and before the PONG came: within half the +/// round trip of its middle, however differently long the two ways took. +#[derive(Clone, Copy, Debug, PartialEq, Eq, serde::Serialize)] +#[serde(rename_all = "camelCase")] +pub(crate) struct ClockOffset { + pub(crate) offset_ms: i64, + pub(crate) within_ms: u64, +} + +impl ClockOffset { + /// From a PONG stamped `at` by its sender, which came at `heard` by our clock, `rtt` + /// after its PING went. + pub(crate) fn of(at: UnixMillis, heard: UnixMillis, rtt: Duration) -> Self { + let half_us = rtt.as_micros() / 2; + let middle_us = i128::from(heard.as_u64()) * 1000 - half_us as i128; + let offset_us = i128::from(at.as_u64()) * 1000 - middle_us; + Self { + offset_ms: i64::try_from(offset_us / 1000).unwrap_or(i64::MAX), + // And a millisecond for the stamps' resolution. + within_ms: u64::try_from(half_us.div_ceil(1000)).unwrap_or(u64::MAX) + 1, + } + } +} /// What paths knows about one path (an address as paths keep it). #[derive(Debug, Default)] @@ -126,6 +155,9 @@ struct PathState { unanswered: u32, /// When a send there last failed. failed_at: Option, + /// The peer's clock as its PONGs on this path say (the narrowest bound lately), and + /// when that sample came. + clock: Option<(ClockOffset, Instant)>, } impl PathState { @@ -324,7 +356,6 @@ impl Table { } } - /// A PONG with `nonce` from `address`: the round trip, if it answers our PING out. /// The transport of `address` now goes to `location` (a libp2p peer over another /// connection: through a relay before, directly now): round trips measured before /// were of another way, and would weigh on the next ones for minutes (`pong` @@ -341,6 +372,7 @@ impl Table { } } + /// A PONG with `nonce` from `address`: its round trip, if it answers our PING out. fn pong(&mut self, address: &str, nonce: u64, now: Instant) -> Option { let path = self.paths.get_mut(address)?; let (expected, sent) = path.probe?; @@ -359,7 +391,22 @@ impl Table { None => sample, }); self.counts.probes_answered += 1; - path.rtt + Some(sample) + } + + /// A clock sample from a PONG over `address`: kept if its bound is as narrow as the + /// one kept, or that one is older than `CLOCK_KEPT`. The narrowest bound is the + /// round trip least delayed by queues and relays. + fn clock(&mut self, address: &str, offset: ClockOffset, now: Instant) { + let Some(path) = self.paths.get_mut(address) else { + return; + }; + let replaces = path.clock.is_none_or(|(kept, at)| { + offset.within_ms <= kept.within_ms || now.saturating_duration_since(at) >= CLOCK_KEPT + }); + if replaces { + path.clock = Some((offset, now)); + } } /// Forgets the paths neither used nor pinged for twice `ACTIVE_FOR`, and choices @@ -490,6 +537,12 @@ impl Paths { self.table.lock_or_recover().paths.get(address)?.rtt } + /// The peer's clock at `address`, as its PONGs there say (`ClockOffset`). + pub(crate) fn clock_offset(&self, address: &str) -> Option { + let table = self.table.lock_or_recover(); + table.paths.get(address)?.clock.map(|(offset, _)| offset) + } + /// What paths knows about each of `addresses`. pub(crate) fn info(&self, addresses: &[String]) -> Vec { let now = Instant::now(); @@ -847,11 +900,15 @@ impl Paths { } }); } - ParsedMessage::Pong { nonce, .. } => { + ParsedMessage::Pong { nonce, at } => { + let heard = self.clock.now_millis(); let location = self.location(&address); + let now = Instant::now(); let mut table = self.table.lock_or_recover(); table.located(&address, &location); - table.pong(&address, nonce, Instant::now()); + if let Some(rtt) = table.pong(&address, nonce, now) { + table.clock(&address, ClockOffset::of(at, heard, rtt), now); + } } _ => {} } @@ -1141,6 +1198,34 @@ mod tests { assert!(table.paths.is_empty() && table.chosen.is_empty()); } + /// A PONG's time against the middle of its round trip: the peer's clock, give or + /// take half the round trip; the narrowest bound kept, until it is old. + #[test] + fn clocks_compare_by_the_middle_of_the_round_trip() { + let rtt = Duration::from_millis(200); + let heard = UnixMillis(10_000); + let ahead = ClockOffset::of(UnixMillis(14_900), heard, rtt); + assert_eq!((ahead.offset_ms, ahead.within_ms), (5_000, 101)); + let behind = ClockOffset::of(UnixMillis(6_900), heard, rtt); + assert_eq!(behind.offset_ms, -3_000); + + let mut table = Table::default(); + let now = Instant::now(); + table.sent("libp2p:z", now); + table.clock("libp2p:z", ahead, now); + let wider = ClockOffset::of(UnixMillis(15_000), heard, rtt * 4); + table.clock("libp2p:z", wider, now); + assert_eq!( + table.paths["libp2p:z"].clock.map(|(kept, _)| kept), + Some(ahead) + ); + table.clock("libp2p:z", wider, now + CLOCK_KEPT); + assert_eq!( + table.paths["libp2p:z"].clock.map(|(kept, _)| kept), + Some(wider) + ); + } + /// A peer's only path is measured too once messages go over it, unless its /// keepalives just did. #[test] diff --git a/src/transport/traffic.rs b/src/transport/traffic.rs index f8ca586..4f56d21 100644 --- a/src/transport/traffic.rs +++ b/src/transport/traffic.rs @@ -207,6 +207,10 @@ impl Transport for CountingTransport { self.inner.rtt(peer) } + fn clock_offset(&self, peer: &str) -> Option { + self.inner.clock_offset(peer) + } + async fn resolve_seed(&self, seed: &str) -> Result { self.inner.resolve_seed(seed).await } -- cgit v1.3.1