get swarm RTT from paths

9 files changed, 117 insertions, 137 deletions

diff --git a/docs/openapi.yaml b/docs/openapi.yaml
index 5fda020..be3005b 100644
--- a/docs/openapi.yaml
+++ b/docs/openapi.yaml
@@ -1186,7 +1186,7 @@ components:
           description: We keep it alive and sync with it on our own; other peers we only answer.
         rttMs:
           type: [number, 'null']
-          description: Effective round-trip time (smoothed, raised by an unanswered HELLO); null without a recent sample.
+          description: Round-trip time of the path messages to the peer go over (before that is measured, of HELLOs); null without a recent value.
         traffic:
           $ref: '#/components/schemas/Traffic'
         incidents:
diff --git a/src/api/peers.rs b/src/api/peers.rs
index f2281c2..a653f7d 100644
--- a/src/api/peers.rs
+++ b/src/api/peers.rs
@@ -69,8 +69,8 @@ pub(crate) struct PeerStatus {
     heads: Option<usize>,
     /// We keep it alive and sync with it on our own; other peers we only answer.
     neighbour: bool,
-    /// Effective round-trip time in milliseconds (smoothed, raised by a HELLO still
-    /// unanswered); absent without a recent sample.
+    /// Round-trip time in milliseconds of the path messages to it go over (`swarm::rtt`);
+    /// absent without a recent value.
     rtt_ms: Option<f64>,
     /// Messages and payload bytes exchanged with it, by kind, since we started (or
     /// since it was last listed).
diff --git a/src/api/stats/prometheus.rs b/src/api/stats/prometheus.rs
index 911228c..559a23e 100644
--- a/src/api/stats/prometheus.rs
+++ b/src/api/stats/prometheus.rs
@@ -808,7 +808,7 @@ pub(super) async fn render(state: &AppState, stats: &StatsResponse) -> String {
     out.family(
         "peer_rtt_seconds",
         "gauge",
-        "Effective round-trip time (smoothed, raised by an unanswered HELLO).",
+        "Round-trip time of the path messages to the peer go over (before that, of HELLOs).",
     );
     for peer in &peers {
         if let Some(rtt) = peer.rtt_secs {
diff --git a/src/swarm/peer/keepalive.rs b/src/swarm/peer/keepalive.rs
index efb8a65..c8a93eb 100644
--- a/src/swarm/peer/keepalive.rs
+++ b/src/swarm/peer/keepalive.rs
@@ -2,7 +2,7 @@
 //!
 //! A quiet pair's keepalive (`PeerState::take_keepalive_due`) is a PING, answered with a
 //! PONG by the peer's paths layer (`transport::paths`), over the same path: it tells both
-//! sides the other is alive, and gives the round trip (`swarm::rtt`). Every `Timing::status_refresh` it is a
+//! sides the other is alive, and measures the path (`swarm::rtt`). Every `Timing::status_refresh` it is a
 //! STATUS instead, answered with a STATUS_REPLY: each side's state (heads, readiness,
 //! epoch), which also go out as they change (relay announcements, the epoch after a
 //! switch, readiness: `announce`). When a STATUS shows our documents differ while sync
@@ -33,18 +33,17 @@ pub(crate) async fn send_status(state: &AppState, peer: &str) -> Result<()> {
         .await
 }
 
-/// Sends `peer` a PING, noting it for its round trip.
+/// Sends `peer` a PING (the paths layer measures the path with it).
 pub(crate) async fn send_ping(state: &AppState, peer: &str) -> Result<()> {
     let nonce = crate::transport::paths::random_nonce();
-    state.lock_rtt().ping_sent(peer, nonce, Instant::now());
     state
         .transport
         .send_message(peer, &ParsedMessage::Ping(nonce))
         .await
 }
 
-/// A PING or PONG from `peer`: it is alive, and a PONG may answer our PING (its round
-/// trip). A transport without the paths layer (`Transport::answers_pings`: in tests)
+/// A PING or PONG from `peer`: it is alive, and after a PONG the paths layer has a
+/// new round-trip time to it (`swarm::rtt`). A transport without the paths layer (`Transport::answers_pings`: in tests)
 /// leaves answering a PING to us. Only peers we know already: a PING makes none.
 pub(super) async fn note_alive(state: &AppState, peer: &str, message: &ParsedMessage) {
     let now = Instant::now();
@@ -55,7 +54,11 @@ pub(super) async fn note_alive(state: &AppState, peer: &str, message: &ParsedMes
                 debug!(peer = %peer, error = %format!("{error:#}"), "PONG failed");
             }
         }
-        ParsedMessage::Pong(nonce) => state.lock_rtt().pong_received(peer, *nonce, now),
+        ParsedMessage::Pong(_) => {
+            if let Some(rtt) = state.transport.rtt(peer) {
+                state.lock_rtt().measured(peer, rtt, now);
+            }
+        }
         _ => {}
     }
     let Some(peer_state) = state.get_peer_state(peer).await else {
diff --git a/src/swarm/rtt.rs b/src/swarm/rtt.rs
index 04d7c8d..cf91ac2 100644
--- a/src/swarm/rtt.rs
+++ b/src/swarm/rtt.rs
@@ -1,19 +1,16 @@
-//! Round-trip times to peers, measured from traffic we send anyway: the keepalive PING
-//! → PONG (`swarm::peer::keepalive`), and HELLO → HELLO_REPLY.
+//! Round-trip times to peers. Mostly the paths layer's (`transport::paths`): the
+//! smoothed round-trip time of the path a peer's messages go over, measured by PINGs
+//! (keepalives, and the paths layer's own), taken whenever a PONG from the peer arrives
+//! (`measured`). Before a peer's path is measured (while joining: no keepalives yet),
+//! HELLO → HELLO_REPLY gives one, smoothed (exponential moving average, weight 1/4);
+//! a HELLO_REPLY only counts when exactly one HELLO was outstanding: with two, we can't
+//! tell which one the answer belongs to (an answer to the older one would look too
+//! fast).
 //!
-//! A PONG answers the PING with its nonce, so it is attributed exactly; a PONG to an
-//! older PING (one a newer one replaced) is no sample. A HELLO_REPLY only counts when
-//! exactly one HELLO was outstanding: with two, we can't tell which one the answer
-//! belongs to (an answer to the older one would look too fast). Samples are smoothed
-//! (exponential moving average, weight 1/4) to ride out jitter.
-//!
-//! What we use is the *effective* round-trip time: the smoothed value, raised to the age
-//! of a HELLO still unanswered (the round trip takes at least that long), so a peer that
-//! stops answering its handshake drifts away by itself. An unanswered PING isn't: a lost
-//! PING or PONG is no sample, not a slow one (the next keepalive measures again, and a
-//! peer that is gone goes dormant). Without a recent sample (nodes move; an old value
-//! says little) it is unknown. Silence alone doesn't count: we don't ask non-neighbours
-//! anything, so their quiet is ours.
+//! Without a recent value (nodes move; an old value says little) it is unknown. A lost
+//! PING or HELLO is no sample: a peer that stops answering goes dormant, and is given up
+//! as a catch-up source, by itself. Silence alone doesn't count: we don't ask
+//! non-neighbours anything, so their quiet is ours.
 
 use std::collections::HashMap;
 use std::time::Duration;
@@ -22,16 +19,11 @@ use tokio::time::Instant;
 
 #[derive(Debug, Default)]
 pub(crate) struct RttTable {
-    pings: HashMap<String, Ping>,
     hellos: HashMap<String, Outstanding>,
-    smoothed: HashMap<String, Sample>,
-}
-
-/// The PING out to a peer.
-#[derive(Debug, Clone, Copy)]
-struct Ping {
-    nonce: u64,
-    sent: Instant,
+    /// From the paths layer, as last measured.
+    from_paths: HashMap<String, Sample>,
+    /// From HELLOs, smoothed.
+    from_hellos: HashMap<String, Sample>,
 }
 
 #[derive(Debug, Clone, Copy)]
@@ -47,88 +39,73 @@ struct Outstanding {
 }
 
 impl RttTable {
-    /// A PING to `peer`: it replaces one still unanswered.
-    pub(crate) fn ping_sent(&mut self, peer: &str, nonce: u64, now: Instant) {
-        self.pings
-            .insert(peer.to_owned(), Ping { nonce, sent: now });
-    }
-
-    pub(crate) fn pong_received(&mut self, peer: &str, nonce: u64, now: Instant) {
-        if self.pings.get(peer).is_some_and(|ping| ping.nonce == nonce)
-            && let Some(ping) = self.pings.remove(peer)
-        {
-            self.record(peer, now.saturating_duration_since(ping.sent), now);
-        }
+    /// The paths layer's round-trip time to `peer` (`Transport::rtt`), as a PONG from
+    /// it arrived.
+    pub(crate) fn measured(&mut self, peer: &str, rtt: Duration, now: Instant) {
+        self.from_paths
+            .insert(peer.to_owned(), Sample { rtt, at: now });
     }
 
     pub(crate) fn hello_sent(&mut self, peer: &str, now: Instant) {
-        Self::sent(&mut self.hellos, peer, now);
+        self.hellos
+            .entry(peer.to_owned())
+            .and_modify(|outstanding| outstanding.count += 1)
+            .or_insert(Outstanding {
+                sent: now,
+                count: 1,
+            });
     }
 
     pub(crate) fn hello_reply_received(&mut self, peer: &str, now: Instant) {
-        if let Some(sample) = Self::answered(&mut self.hellos, peer, now) {
-            self.record(peer, sample, now);
+        let Some(outstanding) = self.hellos.remove(peer) else {
+            return;
+        };
+        if outstanding.count != 1 {
+            return;
         }
+        let sample = now.saturating_duration_since(outstanding.sent);
+        let rtt = match self.from_hellos.get(peer) {
+            Some(old) => (old.rtt * 3 + sample) / 4,
+            None => sample,
+        };
+        self.from_hellos
+            .insert(peer.to_owned(), Sample { rtt, at: now });
     }
 
-    /// The effective round-trip time: the smoothed value if sampled within `fresh_for`,
-    /// raised to the age of the oldest unanswered request; None without a recent sample.
+    /// The round-trip time: the paths layer's if measured within `fresh_for`, else
+    /// HELLOs' if sampled within it; None without either.
     pub(crate) fn effective(
         &self,
         peer: &str,
         now: Instant,
         fresh_for: Duration,
     ) -> Option<Duration> {
-        let sample = self
-            .smoothed
-            .get(peer)
-            .filter(|sample| now.saturating_duration_since(sample.at) < fresh_for)?;
-        let waiting = self
-            .hellos
+        let fresh = |sample: &&Sample| now.saturating_duration_since(sample.at) < fresh_for;
+        self.from_paths
             .get(peer)
-            .map(|outstanding| now.saturating_duration_since(outstanding.sent))
-            .unwrap_or_default();
-        Some(sample.rtt.max(waiting))
+            .filter(fresh)
+            .or_else(|| self.from_hellos.get(peer).filter(fresh))
+            .map(|sample| sample.rtt)
     }
 
-    /// The effective round-trip time of every peer that has one.
+    /// The round-trip time of every peer that has one.
     pub(crate) fn all_effective(
         &self,
         now: Instant,
         fresh_for: Duration,
     ) -> impl Iterator<Item = (&str, Duration)> {
-        self.smoothed.keys().filter_map(move |peer| {
+        let mut peers: Vec<&String> = self
+            .from_paths
+            .keys()
+            .chain(self.from_hellos.keys())
+            .collect();
+        peers.sort_unstable();
+        peers.dedup();
+        peers.into_iter().filter_map(move |peer| {
             self.effective(peer, now, fresh_for)
                 .map(|rtt| (peer.as_str(), rtt))
         })
     }
-
-    fn sent(map: &mut HashMap<String, Outstanding>, peer: &str, now: Instant) {
-        map.entry(peer.to_owned())
-            .and_modify(|outstanding| outstanding.count += 1)
-            .or_insert(Outstanding {
-                sent: now,
-                count: 1,
-            });
-    }
-
-    fn answered(
-        map: &mut HashMap<String, Outstanding>,
-        peer: &str,
-        now: Instant,
-    ) -> Option<Duration> {
-        let outstanding = map.remove(peer)?;
-        (outstanding.count == 1).then(|| now.saturating_duration_since(outstanding.sent))
-    }
-
-    fn record(&mut self, peer: &str, sample: Duration, now: Instant) {
-        let rtt = match self.smoothed.get(peer) {
-            Some(old) => (old.rtt * 3 + sample) / 4,
-            None => sample,
-        };
-        self.smoothed
-            .insert(peer.to_owned(), Sample { rtt, at: now });
-    }
 }
 
 #[cfg(test)]
@@ -138,79 +115,57 @@ mod tests {
     const FRESH: Duration = Duration::from_secs(300);
 
     #[test]
-    fn samples_only_unambiguous_answers_and_smooths() {
+    fn hellos_sample_only_unambiguous_answers_and_smooth() {
         let mut table = RttTable::default();
         let start = Instant::now();
         let at = |ms| start + Duration::from_millis(ms);
-        table.ping_sent("a", 1, start);
-        table.pong_received("a", 1, at(100));
+        table.hello_sent("a", start);
+        table.hello_reply_received("a", at(100));
         assert_eq!(
             table.effective("a", at(100), FRESH),
             Some(Duration::from_millis(100))
         );
-
-        table.ping_sent("a", 2, at(100));
-        table.pong_received("a", 2, at(300));
+        table.hello_sent("a", at(100));
+        table.hello_reply_received("a", at(300));
         assert_eq!(
             table.effective("a", at(300), FRESH),
             Some(Duration::from_millis(125))
         );
-
-        // A newer PING replaced it: a PONG to the older one is no sample.
-        table.ping_sent("a", 3, at(300));
-        table.ping_sent("a", 4, at(400));
-        table.pong_received("a", 3, at(410));
+        // Two HELLOs out: the HELLO_REPLY can't be attributed.
+        table.hello_sent("a", at(300));
+        table.hello_sent("a", at(400));
+        table.hello_reply_received("a", at(410));
         assert_eq!(
             table.effective("a", at(410), FRESH),
             Some(Duration::from_millis(125))
         );
-        table.pong_received("a", 4, at(500));
-        assert_eq!(
-            table.effective("a", at(500), FRESH),
-            Some(Duration::from_micros(118_750)) // (125 * 3 + 100) / 4
-        );
-
-        // Two HELLOs out: the HELLO_REPLY can't be attributed.
-        table.hello_sent("c", start);
-        table.hello_reply_received("c", at(100));
-        table.hello_sent("c", at(100));
-        table.hello_sent("c", at(200));
-        table.hello_reply_received("c", at(210));
-        assert_eq!(
-            table.effective("c", at(210), FRESH),
-            Some(Duration::from_millis(100))
-        );
-
-        // An unsolicited PONG changes nothing.
-        table.pong_received("b", 9, start);
+        // An unsolicited HELLO_REPLY changes nothing.
+        table.hello_reply_received("b", start);
         assert_eq!(table.effective("b", start, FRESH), None);
     }
 
     #[test]
-    fn unanswered_hellos_raise_it_and_old_samples_expire() {
+    fn the_path_wins_nothing_waits_and_old_values_expire() {
         let mut table = RttTable::default();
         let start = Instant::now();
-        table.ping_sent("a", 1, start);
-        table.pong_received("a", 1, start + Duration::from_millis(50));
-
-        // PINGs unanswered (lost): no sample, nothing raised.
-        let sent = start + Duration::from_secs(60);
-        table.ping_sent("a", 2, sent);
-        table.ping_sent("a", 3, sent + Duration::from_secs(30));
-        let later = sent + Duration::from_secs(40);
+        table.hello_sent("a", start);
+        table.hello_reply_received("a", start + Duration::from_millis(80));
+        table.measured("a", Duration::from_millis(20), start);
         assert_eq!(
-            table.effective("a", later, FRESH),
-            Some(Duration::from_millis(50))
+            table.effective("a", start, FRESH),
+            Some(Duration::from_millis(20))
         );
-
-        // A HELLO unanswered for 40 s: at least 40 s.
-        table.hello_sent("a", sent);
+        // A HELLO unanswered for long raises nothing.
+        table.hello_sent("a", start);
         assert_eq!(
-            table.effective("a", later, FRESH),
-            Some(Duration::from_secs(40))
+            table.effective("a", start + Duration::from_secs(40), FRESH),
+            Some(Duration::from_millis(20))
         );
-
-        // No sample within the freshness window: unknown.
+        assert_eq!(
+            table.all_effective(start, FRESH).collect::<Vec<_>>(),
+            [("a", Duration::from_millis(20))]
+        );
+        // No value within the freshness window: unknown.
         let stale = start + FRESH + Duration::from_secs(1);
         assert_eq!(table.effective("a", stale, FRESH), None);
     }
diff --git a/src/transport/identity.rs b/src/transport/identity.rs
index 153dacc..0c4cda4 100644
--- a/src/transport/identity.rs
+++ b/src/transport/identity.rs
@@ -496,6 +496,12 @@ impl Transport for IdentityTransport {
         true
     }
 
+    /// The best path's, where messages to `peer` go.
+    fn rtt(&self, peer: &str) -> Option<Duration> {
+        let address = self.shared.proven_at(peer).into_iter().next()?;
+        self.shared.paths.rtt(&address)
+    }
+
     /// 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 f60a9f6..80b71cd 100644
--- a/src/transport/mod.rs
+++ b/src/transport/mod.rs
@@ -649,6 +649,13 @@ pub(crate) trait Transport: Send + Sync {
         None
     }
 
+    /// The smoothed round-trip time of the path messages to `peer` go over, if
+    /// measured (`transport::paths`). By default none.
+    fn rtt(&self, peer: &str) -> Option<std::time::Duration> {
+        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 643a238..a9bac49 100644
--- a/src/transport/paths.rs
+++ b/src/transport/paths.rs
@@ -420,6 +420,11 @@ impl Paths {
         self.table.lock_or_recover().rank(usable, Instant::now())
     }
 
+    /// The smoothed round-trip time of the path to `address`, if measured.
+    pub(crate) fn rtt(&self, address: &str) -> Option<Duration> {
+        self.table.lock_or_recover().paths.get(address)?.rtt
+    }
+
     /// What paths knows about each of `addresses`.
     pub(crate) fn info(&self, addresses: &[String]) -> Vec<PathInfo> {
         let now = Instant::now();
diff --git a/src/transport/traffic.rs b/src/transport/traffic.rs
index b070023..64595cb 100644
--- a/src/transport/traffic.rs
+++ b/src/transport/traffic.rs
@@ -192,6 +192,10 @@ impl Transport for CountingTransport {
         self.inner.answers_pings()
     }
 
+    fn rtt(&self, peer: &str) -> Option<std::time::Duration> {
+        self.inner.rtt(peer)
+    }
+
     async fn resolve_seed(&self, seed: &str) -> Result<String> {
         self.inner.resolve_seed(seed).await
     }