hop delays

1 files changed, 10 insertions, 1 deletions

diff --git a/src/swarm/sync.rs b/src/swarm/sync.rs
index 4bb6ba5..a9d98b0 100644
--- a/src/swarm/sync.rs
+++ b/src/swarm/sync.rs
@@ -140,6 +140,7 @@ async fn deliver_sync_to_peer(state: &AppState, peer: &str) -> Result<()> {
         let delta_threshold = state.blobs.delta_threshold_for(rate, failed);
 
         let stats = Arc::clone(&state.stats);
+        let clock = Arc::clone(&state.clock);
         // The session the message comes from, for its send's outcome.
         let (session, generated) = tokio::task::spawn_blocking(
             move || -> (u64, Result<Option<transport::EncodedMessage>>) {
@@ -205,6 +206,7 @@ async fn deliver_sync_to_peer(state: &AppState, peer: &str) -> Result<()> {
                     ready,
                     writer,
                     delivered,
+                    sent: clock.now_millis(),
                 };
                 let encoded = transport
                     .encode(&peer_owned, &ParsedMessage::Sync(envelope))
@@ -318,11 +320,18 @@ pub(crate) async fn handle_sync(
     debug!(peer = %peer, "Processing sync envelope");
     // This message is counted already (`transport::traffic`).
     let arrived = (Instant::now(), SyncBytes::of(&state.traffic, peer));
+    // How long it took to come, where we know the peer's clock (`swarm::hop`).
+    let hop_delay = state.transport.clock_offset(peer).map(|offset| {
+        crate::swarm::hop::delay_ms(envelope.sent, state.clock.now_millis(), offset.offset_ms)
+    });
 
     let peer_state_mutex = state.get_or_create_peer_state(peer).await;
 
     let needs_hello = {
-        let peer_state = peer_state_mutex.lock().await;
+        let mut peer_state = peer_state_mutex.lock().await;
+        if let Some(delay_ms) = hop_delay {
+            peer_state.hop_delays.record(delay_ms, arrived.0);
+        }
 
         match &peer_state.standing {
             Standing::Stranger | Standing::Admitted => {