From 32c788b4d6245abebafb250b005a211fcbeac32c Mon Sep 17 00:00:00 2001 From: tv Date: Sat, 31 Oct 2026 06:50:07 +0100 Subject: pong carries time --- src/transport/paths.rs | 19 ++++++++++++++++--- 1 file changed, 16 insertions(+), 3 deletions(-) (limited to 'src/transport/paths.rs') 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>, /// The debug firewall, if any. gate: Option>, + /// The node's wall clock, which its PONGs say the time by. + clock: Arc, /// A std mutex: never held across an await. table: Mutex, } 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)>, traffic: Option>, gate: Option>, + clock: Arc, ) -> 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)], None, Some(Arc::downgrade(&gate)), + Arc::default(), )); let a = Arc::new(Paths::new( vec![("sim", net.transport("a") as Arc)], None, None, + Arc::default(), )); let received = |paths: &Arc| { 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)], None, None, + Arc::default(), )); let (sender, mut received) = tokio::sync::mpsc::unbounded_channel(); let handler: Arc = Arc::new(move |from, _, _| { -- cgit v1.3.1