move keepalive and firewall to paths

1 files changed, 76 insertions, 32 deletions

diff --git a/src/swarm/rtt.rs b/src/swarm/rtt.rs
index 01d10ee..466fc98 100644
--- a/src/swarm/rtt.rs
+++ b/src/swarm/rtt.rs
@@ -1,9 +1,11 @@
-//! Round-trip times to peers, measured from traffic we send anyway: STATUS → STATUS_REPLY and
-//! HELLO → HELLO_REPLY. No probes of our own.
+//! Round-trip times to peers, measured from traffic we send anyway: the keepalive PING
+//! → PONG (`swarm::peer::keepalive`), and HELLO → HELLO_REPLY.
 //!
-//! 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.
+//! 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 the oldest request still unanswered (the round trip takes at least that long), so
@@ -18,11 +20,20 @@ use tokio::time::Instant;
 
 #[derive(Debug, Default)]
 pub(crate) struct RttTable {
-    statuses: HashMap<String, Outstanding>,
+    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,
+    /// When the oldest PING still unanswered went out (an earlier one, replaced).
+    waiting_since: Instant,
+}
+
 #[derive(Debug, Clone, Copy)]
 struct Sample {
     rtt: Duration,
@@ -36,13 +47,26 @@ struct Outstanding {
 }
 
 impl RttTable {
-    pub(crate) fn status_sent(&mut self, peer: &str, now: Instant) {
-        Self::sent(&mut self.statuses, peer, now);
+    pub(crate) fn ping_sent(&mut self, peer: &str, nonce: u64, now: Instant) {
+        let waiting_since = self
+            .pings
+            .get(peer)
+            .map_or(now, |earlier| earlier.waiting_since);
+        self.pings.insert(
+            peer.to_owned(),
+            Ping {
+                nonce,
+                sent: now,
+                waiting_since,
+            },
+        );
     }
 
-    pub(crate) fn status_reply_received(&mut self, peer: &str, now: Instant) {
-        if let Some(sample) = Self::answered(&mut self.statuses, peer, now) {
-            self.record(peer, sample, 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);
         }
     }
 
@@ -68,12 +92,15 @@ impl RttTable {
             .smoothed
             .get(peer)
             .filter(|sample| now.saturating_duration_since(sample.at) < fresh_for)?;
-        let waiting = [self.statuses.get(peer), self.hellos.get(peer)]
-            .into_iter()
-            .flatten()
-            .map(|outstanding| now.saturating_duration_since(outstanding.sent))
-            .max()
-            .unwrap_or_default();
+        let waiting = [
+            self.pings.get(peer).map(|ping| ping.waiting_since),
+            self.hellos.get(peer).map(|outstanding| outstanding.sent),
+        ]
+        .into_iter()
+        .flatten()
+        .map(|sent| now.saturating_duration_since(sent))
+        .max()
+        .unwrap_or_default();
         Some(sample.rtt.max(waiting))
     }
 
@@ -128,31 +155,47 @@ mod tests {
         let mut table = RttTable::default();
         let start = Instant::now();
         let at = |ms| start + Duration::from_millis(ms);
-        table.status_sent("a", start);
-        table.status_reply_received("a", at(100));
+        table.ping_sent("a", 1, start);
+        table.pong_received("a", 1, at(100));
         assert_eq!(
             table.effective("a", at(100), FRESH),
             Some(Duration::from_millis(100))
         );
 
-        table.status_sent("a", at(100));
-        table.status_reply_received("a", at(300));
+        table.ping_sent("a", 2, at(100));
+        table.pong_received("a", 2, at(300));
         assert_eq!(
             table.effective("a", at(300), FRESH),
             Some(Duration::from_millis(125))
         );
 
-        // Two STATUSes out: the STATUS_REPLY can't be attributed.
-        table.status_sent("a", at(300));
-        table.status_sent("a", at(400));
-        table.status_reply_received("a", at(450));
+        // 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));
         assert_eq!(
-            table.effective("a", at(450), FRESH),
+            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 STATUS_REPLY changes nothing.
-        table.status_reply_received("b", start);
+        // An unsolicited PONG changes nothing.
+        table.pong_received("b", 9, start);
         assert_eq!(table.effective("b", start, FRESH), None);
     }
 
@@ -160,12 +203,13 @@ mod tests {
     fn unanswered_requests_raise_it_and_old_samples_expire() {
         let mut table = RttTable::default();
         let start = Instant::now();
-        table.status_sent("a", start);
-        table.status_reply_received("a", start + Duration::from_millis(50));
+        table.ping_sent("a", 1, start);
+        table.pong_received("a", 1, start + Duration::from_millis(50));
 
-        // A STATUS unanswered for 40 s: at least 40 s.
+        // PINGs unanswered for 40 s: at least 40 s, from the oldest.
         let sent = start + Duration::from_secs(60);
-        table.status_sent("a", sent);
+        table.ping_sent("a", 2, sent);
+        table.ping_sent("a", 3, sent + Duration::from_secs(30));
         let later = sent + Duration::from_secs(40);
         assert_eq!(
             table.effective("a", later, FRESH),