pong carries time

1 files changed, 16 insertions, 3 deletions

diff --git a/src/transport/paths.rs b/src/transport/paths.rs
index 2e7ef76..fcfadaa 100644
--- a/src/transport/paths.rs
+++ b/src/transport/paths.rs
@@ -47,6 +47,7 @@ use anyhow::{Context, Result, anyhow};
 use tokio::time::Instant;
 use tracing::{debug, info};
 
+use crate::clock::Clock;
 use crate::locks::MutexExt;
 use crate::transport::stats::{PathCounts, PathInfo, PathsStats, TransportStats};
 use crate::transport::traffic::Traffic;
@@ -394,17 +395,20 @@ pub(crate) struct Paths {
     traffic: Option<Arc<Traffic>>,
     /// The debug firewall, if any.
     gate: Option<Weak<dyn Gate>>,
+    /// The node's wall clock, which its PONGs say the time by.
+    clock: Arc<Clock>,
     /// A std mutex: never held across an await.
     table: Mutex<Table>,
 }
 
 impl Paths {
     /// Over `transports`, by scheme, in order of preference; its own PINGs and PONGs
-    /// counted in `traffic`, the firewall through `gate`.
+    /// counted in `traffic`, the firewall through `gate`, PONGs stamped by `clock`.
     pub(crate) fn new(
         transports: Vec<(&'static str, Arc<dyn Transport>)>,
         traffic: Option<Arc<Traffic>>,
         gate: Option<Weak<dyn Gate>>,
+        clock: Arc<Clock>,
     ) -> Self {
         Self {
             links: transports
@@ -413,6 +417,7 @@ impl Paths {
                 .collect(),
             traffic,
             gate,
+            clock,
             table: Mutex::default(),
         }
     }
@@ -766,7 +771,10 @@ impl Paths {
     /// (`traffic`).
     async fn send_probe(&self, address: &str, pong: bool, nonce: u64) -> Result<()> {
         let message = if pong {
-            ParsedMessage::Pong(nonce)
+            ParsedMessage::Pong {
+                nonce,
+                at: self.clock.now_millis(),
+            }
         } else {
             ParsedMessage::Ping(nonce)
         };
@@ -839,7 +847,7 @@ impl Paths {
                     }
                 });
             }
-            ParsedMessage::Pong(nonce) => {
+            ParsedMessage::Pong { nonce, .. } => {
                 let location = self.location(&address);
                 let mut table = self.table.lock_or_recover();
                 table.located(&address, &location);
@@ -959,6 +967,7 @@ mod tests {
                 ],
                 None,
                 None,
+                Arc::default(),
             ))
         };
         (on_both("a"), on_both("b"))
@@ -1252,11 +1261,13 @@ mod tests {
             vec![("sim", net.transport("b") as Arc<dyn Transport>)],
             None,
             Some(Arc::downgrade(&gate)),
+            Arc::default(),
         ));
         let a = Arc::new(Paths::new(
             vec![("sim", net.transport("a") as Arc<dyn Transport>)],
             None,
             None,
+            Arc::default(),
         ));
         let received = |paths: &Arc<Paths>| {
             let (sender, received) = tokio::sync::mpsc::unbounded_channel();
@@ -1350,11 +1361,13 @@ mod tests {
             ],
             None,
             None,
+            Arc::default(),
         );
         let b = Arc::new(Paths::new(
             vec![("sim", net.transport("b") as Arc<dyn Transport>)],
             None,
             None,
+            Arc::default(),
         ));
         let (sender, mut received) = tokio::sync::mpsc::unbounded_channel();
         let handler: Arc<PathHandler> = Arc::new(move |from, _, _| {