links replies carry clock offsets

1 files changed, 24 insertions, 3 deletions

diff --git a/src/swarm/links.rs b/src/swarm/links.rs
index 82cd991..d5238f3 100644
--- a/src/swarm/links.rs
+++ b/src/swarm/links.rs
@@ -25,7 +25,7 @@ use tracing::debug;
 use crate::locks::MutexExt;
 use crate::state::AppState;
 use crate::swarm::{members, topology};
-use crate::transport::paths;
+use crate::transport::paths::{self, ClockOffset};
 use crate::transport::{
     Connection, Destination, Link, LinksReply, LinksRequest, OtherPath, ParsedMessage,
 };
@@ -128,6 +128,7 @@ pub(crate) async fn own_links(state: &AppState) -> Vec<Link> {
                 .rtt(&peer)
                 .map(|rtt| u32::try_from(rtt.as_micros()).unwrap_or(u32::MAX)),
             dormant,
+            clock: state.transport.clock_offset(&peer),
             peer,
         });
     }
@@ -354,9 +355,16 @@ pub(crate) async fn watch(state: AppState) {
 }
 
 /// Whether `new` says something `old` didn't: another peer, transport, relay or
-/// dormancy, or a round trip measured or moved by more than a fifth (and 1 ms). Round
-/// trips wobble; small moves aren't news.
+/// dormancy, a round trip measured or moved by more than a fifth (and 1 ms), or a clock
+/// measured or moved beyond both bounds. Round trips and clock samples wobble; small
+/// moves aren't news.
 fn changed(old: &[Link], new: &[Link]) -> bool {
+    let clock_moved = |a: Option<ClockOffset>, b: Option<ClockOffset>| match (a, b) {
+        (Some(a), Some(b)) => {
+            a.offset_ms.abs_diff(b.offset_ms) > a.within_ms.saturating_add(b.within_ms)
+        }
+        (a, b) => a.is_some() != b.is_some(),
+    };
     let moved = |a: Option<u32>, b: Option<u32>| match (a, b) {
         (Some(a), Some(b)) => {
             let by = a.abs_diff(b);
@@ -371,6 +379,7 @@ fn changed(old: &[Link], new: &[Link]) -> bool {
                 || a.relayed != b.relayed
                 || a.dormant != b.dormant
                 || moved(a.rtt_us, b.rtt_us)
+                || clock_moved(a.clock, b.clock)
                 || a.others.len() != b.others.len()
                 || a.others.iter().zip(&b.others).any(|(x, y)| {
                     x.transport != y.transport
@@ -392,6 +401,7 @@ mod tests {
             rtt_us,
             dormant: false,
             others: Vec::new(),
+            clock: None,
         }
     }
 
@@ -408,5 +418,16 @@ mod tests {
         assert!(changed(&old, &[dormant]));
         // Small round trips: a millisecond's wobble isn't news.
         assert!(!changed(&[link("a", Some(400))], &[link("a", Some(1_300))]));
+        // A clock: news when measured, and when it moved beyond both bounds.
+        let clocked = |offset_ms| Link {
+            clock: Some(ClockOffset {
+                offset_ms,
+                within_ms: 50,
+            }),
+            ..link("a", Some(10_000))
+        };
+        assert!(changed(&old, &[clocked(0)]));
+        assert!(!changed(&[clocked(0)], &[clocked(80)]));
+        assert!(changed(&[clocked(0)], &[clocked(5_000)]));
     }
 }