record RTT sampling instant

4 files changed, 104 insertions, 30 deletions

diff --git a/src/api/peers.rs b/src/api/peers.rs
index 3893e0f..bafb5fa 100644
--- a/src/api/peers.rs
+++ b/src/api/peers.rs
@@ -62,8 +62,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,
-    /// Smoothed round-trip time in milliseconds, from PING/PONG and HELLO/HELLO_REPLY;
-    /// absent until measured.
+    /// Effective round-trip time in milliseconds (smoothed, raised by requests still
+    /// unanswered); absent without a recent sample.
     rtt_ms: Option<f64>,
 }
 
@@ -114,7 +114,7 @@ pub(crate) async fn list_peers(
         let neighbour = crate::swarm::topology::is_neighbour(&state, &public_key);
         let rtt_ms = state
             .lock_rtt()
-            .get(&public_key)
+            .effective(&public_key, now, state.timing.rtt_fresh_for())
             .map(|rtt| rtt.as_secs_f64() * 1000.0);
         let ps = peer_state_mutex.lock().await;
 
diff --git a/src/state.rs b/src/state.rs
index 6021baa..9d567a8 100644
--- a/src/state.rs
+++ b/src/state.rs
@@ -473,6 +473,12 @@ pub(crate) struct Timing {
 }
 
 impl Timing {
+    /// How long a round-trip sample stays meaningful (`swarm::rtt`): a few keepalive
+    /// intervals, since nodes move.
+    pub(crate) fn rtt_fresh_for(&self) -> Duration {
+        self.keepalive * 5
+    }
+
     // Recommended values
     //  Environment                                 | Keepalive | Notes
     //  --------------------------------------------+-----------+------------------------------------------------------------
diff --git a/src/swarm/rtt.rs b/src/swarm/rtt.rs
index 04d0722..ed6cc73 100644
--- a/src/swarm/rtt.rs
+++ b/src/swarm/rtt.rs
@@ -4,6 +4,12 @@
 //! A sample only counts when exactly one request 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 the oldest request still unanswered (the round trip takes at least that long), so
+//! a peer that stops answering drifts away by itself. 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.
 
 use std::collections::HashMap;
 use std::time::Duration;
@@ -14,7 +20,13 @@ use tokio::time::Instant;
 pub(crate) struct RttTable {
     pings: HashMap<String, Outstanding>,
     hellos: HashMap<String, Outstanding>,
-    smoothed: HashMap<String, Duration>,
+    smoothed: HashMap<String, Sample>,
+}
+
+#[derive(Debug, Clone, Copy)]
+struct Sample {
+    rtt: Duration,
+    at: Instant,
 }
 
 #[derive(Debug, Clone, Copy)]
@@ -30,7 +42,7 @@ impl RttTable {
 
     pub(crate) fn pong_received(&mut self, peer: &str, now: Instant) {
         if let Some(sample) = Self::answered(&mut self.pings, peer, now) {
-            self.record(peer, sample);
+            self.record(peer, sample, now);
         }
     }
 
@@ -40,20 +52,41 @@ impl RttTable {
 
     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);
+            self.record(peer, sample, now);
         }
     }
 
-    /// The smoothed round-trip time, once measured.
-    pub(crate) fn get(&self, peer: &str) -> Option<Duration> {
-        self.smoothed.get(peer).copied()
+    /// 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.
+    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.pings.get(peer), self.hellos.get(peer)]
+            .into_iter()
+            .flatten()
+            .map(|outstanding| now.saturating_duration_since(outstanding.sent))
+            .max()
+            .unwrap_or_default();
+        Some(sample.rtt.max(waiting))
     }
 
-    /// All measured peers.
-    pub(crate) fn all(&self) -> impl Iterator<Item = (&str, Duration)> {
-        self.smoothed
-            .iter()
-            .map(|(peer, rtt)| (peer.as_str(), *rtt))
+    /// The effective 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| {
+            self.effective(peer, now, fresh_for)
+                .map(|rtt| (peer.as_str(), rtt))
+        })
     }
 
     fn sent(map: &mut HashMap<String, Outstanding>, peer: &str, now: Instant) {
@@ -74,12 +107,13 @@ impl RttTable {
         (outstanding.count == 1).then(|| now.saturating_duration_since(outstanding.sent))
     }
 
-    fn record(&mut self, peer: &str, sample: Duration) {
-        let smoothed = match self.smoothed.get(peer) {
-            Some(old) => (*old * 3 + sample) / 4,
+    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(), smoothed);
+        self.smoothed
+            .insert(peer.to_owned(), Sample { rtt, at: now });
     }
 }
 
@@ -87,26 +121,59 @@ impl RttTable {
 mod tests {
     use super::*;
 
+    const FRESH: Duration = Duration::from_secs(300);
+
     #[test]
     fn samples_only_unambiguous_answers_and_smooths() {
         let mut table = RttTable::default();
         let start = Instant::now();
+        let at = |ms| start + Duration::from_millis(ms);
         table.ping_sent("a", start);
-        table.pong_received("a", start + Duration::from_millis(100));
-        assert_eq!(table.get("a"), Some(Duration::from_millis(100)));
+        table.pong_received("a", at(100));
+        assert_eq!(
+            table.effective("a", at(100), FRESH),
+            Some(Duration::from_millis(100))
+        );
 
-        table.ping_sent("a", start);
-        table.pong_received("a", start + Duration::from_millis(200));
-        assert_eq!(table.get("a"), Some(Duration::from_millis(125)));
+        table.ping_sent("a", at(100));
+        table.pong_received("a", at(300));
+        assert_eq!(
+            table.effective("a", at(300), FRESH),
+            Some(Duration::from_millis(125))
+        );
 
         // Two PINGs out: the PONG can't be attributed.
-        table.ping_sent("a", start);
-        table.ping_sent("a", start + Duration::from_secs(1));
-        table.pong_received("a", start + Duration::from_secs(1));
-        assert_eq!(table.get("a"), Some(Duration::from_millis(125)));
+        table.ping_sent("a", at(300));
+        table.ping_sent("a", at(400));
+        table.pong_received("a", at(450));
+        assert_eq!(
+            table.effective("a", at(450), FRESH),
+            Some(Duration::from_millis(125))
+        );
 
         // An unsolicited PONG changes nothing.
         table.pong_received("b", start);
-        assert_eq!(table.get("b"), None);
+        assert_eq!(table.effective("b", start, FRESH), None);
+    }
+
+    #[test]
+    fn unanswered_requests_raise_it_and_old_samples_expire() {
+        let mut table = RttTable::default();
+        let start = Instant::now();
+        table.ping_sent("a", start);
+        table.pong_received("a", start + Duration::from_millis(50));
+
+        // A PING unanswered for 40 s: at least 40 s.
+        let sent = start + Duration::from_secs(60);
+        table.ping_sent("a", sent);
+        let later = sent + Duration::from_secs(40);
+        assert_eq!(
+            table.effective("a", later, FRESH),
+            Some(Duration::from_secs(40))
+        );
+
+        // No sample 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/swarm/source.rs b/src/swarm/source.rs
index 292cfc8..36a333e 100644
--- a/src/swarm/source.rs
+++ b/src/swarm/source.rs
@@ -82,8 +82,9 @@ pub(crate) fn defer(state: &AppState, peer: &str, peer_state: &PeerState) -> Opt
     // that, whoever asks first takes it (the closer one may not be suitable after all).
     let rtt = {
         let rtts = state.lock_rtt();
-        let mine = rtts.get(peer);
-        let closer = rtts.all().any(|(other, rtt)| {
+        let fresh_for = state.timing.rtt_fresh_for();
+        let mine = rtts.effective(peer, now, fresh_for);
+        let closer = rtts.all_effective(now, fresh_for).any(|(other, rtt)| {
             other != peer
                 && quiet.as_deref() != Some(other)
                 && mine.is_none_or(|mine| rtt * 5 < mine * 4)