calculate peer clock drift

1 files changed, 90 insertions, 5 deletions

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<Instant>,
+    /// 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<Duration> {
         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<ClockOffset> {
+        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<PathInfo> {
         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]