6 files changed, 148 insertions, 5 deletions

diff --git a/src/api/peers.rs b/src/api/peers.rs
index 020221b..3fe4c39 100644
--- a/src/api/peers.rs
+++ b/src/api/peers.rs
@@ -18,6 +18,7 @@ use crate::locks::MutexExt;
 use crate::state::{AppState, HelloState, Incidents, Liveness};
 use crate::swarm::peer::forgotten_path;
 use crate::transport::Destination;
+use crate::transport::paths::ClockOffset;
 use crate::transport::traffic::PeerTraffic;
 
 #[derive(Debug, Serialize)]
@@ -66,6 +67,9 @@ pub(crate) struct PeerStatus {
     /// Smoothed sync throughput (`swarm::throughput`); absent before the first large
     /// exchange.
     pub(crate) sync_throughput: Option<ThroughputInfo>,
+    /// How far its clock is ahead of ours, give or take (`Transport::clock_offset`);
+    /// absent before a PONG from it.
+    pub(crate) clock: Option<ClockOffset>,
 }
 
 #[derive(Debug, Serialize)]
@@ -141,6 +145,7 @@ pub(crate) async fn peer_list(state: &AppState) -> PeerListResponse {
         let alias = aliases.remove(&peer);
         let addresses = state.transport.addresses_of(&peer);
         let destination = state.transport.destination(&peer);
+        let clock = state.transport.clock_offset(&peer);
         peers.insert(
             peer,
             PeerStatus {
@@ -168,6 +173,7 @@ pub(crate) async fn peer_list(state: &AppState) -> PeerListResponse {
                 traffic,
                 incidents: ps.incidents,
                 sync_throughput,
+                clock,
             },
         );
     }
diff --git a/src/api/stats/prometheus.rs b/src/api/stats/prometheus.rs
index 8f7538f..2f33bef 100644
--- a/src/api/stats/prometheus.rs
+++ b/src/api/stats/prometheus.rs
@@ -9,6 +9,7 @@ use crate::api::media_type;
 use crate::api::peers::{PeerListResponse, PeerLivenessStatus};
 use crate::state::Incidents;
 use crate::transport::Destination;
+use crate::transport::paths::ClockOffset;
 use crate::transport::stats::{Reachability, TransportStats};
 use crate::transport::traffic::{ByKind, Counter, PeerTraffic};
 
@@ -151,6 +152,7 @@ struct Peer {
     incidents: Incidents,
     throughput: Option<f64>,
     destination: Option<Destination>,
+    clock: Option<ClockOffset>,
 }
 
 /// The rows of `/peers` (`peers::peer_list`), as the series need them.
@@ -171,6 +173,7 @@ fn peers(list: &PeerListResponse) -> Vec<Peer> {
             incidents: peer.incidents,
             throughput: peer.sync_throughput.as_ref().map(|t| t.bytes_per_sec),
             destination: peer.destination.clone(),
+            clock: peer.clock,
         })
         .collect()
 }
@@ -928,6 +931,35 @@ pub(super) fn render(stats: &StatsResponse, list: &PeerListResponse) -> String {
     }
 
     out.family(
+        "peer_clock_offset_seconds",
+        "gauge",
+        "How far the peer's clock is ahead of ours (behind: negative), from its PONGs.",
+    );
+    for peer in &peers {
+        if let Some(clock) = peer.clock {
+            out.sample(
+                "peer_clock_offset_seconds",
+                &[("peer", &peer.key)],
+                clock.offset_ms as f64 / 1000.0,
+            );
+        }
+    }
+    out.family(
+        "peer_clock_offset_within_seconds",
+        "gauge",
+        "Give or take this much: half the round trip it was measured over.",
+    );
+    for peer in &peers {
+        if let Some(clock) = peer.clock {
+            out.sample(
+                "peer_clock_offset_within_seconds",
+                &[("peer", &peer.key)],
+                clock.within_ms as f64 / 1000.0,
+            );
+        }
+    }
+
+    out.family(
         "peer_messages_total",
         "counter",
         "Messages by direction and kind.",
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<crate::transport::paths::ClockOffset> {
+        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<paths::ClockOffset> {
+        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<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]
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<crate::transport::paths::ClockOffset> {
+        self.inner.clock_offset(peer)
+    }
+
     async fn resolve_seed(&self, seed: &str) -> Result<String> {
         self.inner.resolve_seed(seed).await
     }