paths: PINGs of a path take turns with the peer's

The node's keepalives take turns between the two sides (`PeerState::keepalive_due_at`),
but the paths layer pinged every path in use every `probe_every` (the keepalive interval,
30 s at least) whatever the peer did: a PING from the peer over the path didn't count.
So both sides pinged each path every interval, often within a second of each other
(seen on the dashboard: a PING answered, and our own PING right after it).

Now a PING from the peer over a path is noted (`PathState::peer_probe`), and a path is
pinged again (`rested`) an interval after the peer's PING, if that came last; after
ours, an interval and a half while the peer pings too (its PING within two intervals
before ours), so its turn comes first; else an interval. One exchange per interval
goes over a path, as the measurements' "two keepalives per neighbour per interval"
says, and each side measures every other interval; a shorter `--keepalive` measures
more often. A path whose way out broke while the peer's PINGs still come in is found by
our own PINGs half as often; a failed send or unanswered traffic (`suspect`) still
pings at once.

1 files changed, 82 insertions, 4 deletions

diff --git a/src/transport/paths.rs b/src/transport/paths.rs
index 844a897..a830143 100644
--- a/src/transport/paths.rs
+++ b/src/transport/paths.rs
@@ -19,7 +19,11 @@
 //! keepalive PINGs go over, below): every `PROBE_EVERY`, or every keepalive interval if
 //! that is longer, each address it had a choice of
 //! within `ACTIVE_FOR` gets a PING with a random nonce, which the
-//! other side's paths answers at once with a PONG, over the same path. Answers give each
+//! other side's paths answers at once with a PONG, over the same path. The two sides
+//! take turns on a path, as the node's keepalives do: a PING from the peer there makes
+//! the next ours, an interval after it; after ours, while the peer pings too, we wait
+//! half an interval more, in case its PING doesn't come. So one exchange per interval
+//! goes over a path, and each side measures every other interval. Answers give each
 //! path's round-trip time (smoothed); `UNANSWERED_DOWN` PINGs in a row unanswered, or a
 //! send that failed within `DOWN_FOR`, make a path down. What went to a peer going
 //! unanswered makes the path in use suspect (`Paths::suspect`): the others first, all
@@ -154,7 +158,10 @@ struct PathState {
     measured_at: Option<String>,
     /// The PING out, by nonce, and when it went.
     probe: Option<(u64, Instant)>,
+    /// When we last pinged it (a probe, or the node's keepalive).
     last_probe: Option<Instant>,
+    /// When the peer last pinged us over it: the next turn is ours (`Table::due`).
+    peer_probe: Option<Instant>,
     /// PINGs in a row unanswered.
     unanswered: u32,
     /// When a send there last failed.
@@ -167,6 +174,23 @@ struct PathState {
     suspect: bool,
 }
 
+/// Whether `path` may be pinged again at `now`, pinged every `every` by the two sides in
+/// turns: an interval after the peer's PING, if it came last; after ours, an interval and
+/// a half while the peer pings too (its PING within two intervals before ours), so its
+/// turn comes first, else an interval.
+fn rested(path: &PathState, now: Instant, every: Duration) -> bool {
+    let since = |at: Instant| now.saturating_duration_since(at);
+    match (path.last_probe, path.peer_probe) {
+        (None, Some(theirs)) => since(theirs) >= every,
+        (None, None) => true,
+        (Some(ours), Some(theirs)) if theirs > ours => since(theirs) >= every,
+        (Some(ours), Some(theirs)) if ours.saturating_duration_since(theirs) < every * 2 => {
+            since(ours) >= every + every / 2
+        }
+        (Some(ours), _) => since(ours) >= every,
+    }
+}
+
 impl PathState {
     fn is_down(&self, now: Instant) -> bool {
         self.unanswered >= UNANSWERED_DOWN
@@ -351,9 +375,7 @@ impl Table {
                 at.is_some_and(|at| now.saturating_duration_since(at) < ACTIVE_FOR)
             };
             let active = lately(path.used) || lately(path.carried);
-            let rested = path
-                .last_probe
-                .is_none_or(|last| now.saturating_duration_since(last) >= self.probe_every);
+            let rested = rested(path, now, self.probe_every);
             if active && rested && path.probe.is_none() {
                 let nonce = nonce();
                 path.probe = Some((nonce, now));
@@ -415,6 +437,14 @@ impl Table {
         self.counts.probes_sent += 1;
     }
 
+    /// The peer pinged us over `address`: the path's next PING is ours, an interval on
+    /// (`rested`). Only a path known already: a PING makes none.
+    fn peer_pinged(&mut self, address: &str, now: Instant) {
+        if let Some(path) = self.paths.get_mut(address) {
+            path.peer_probe = Some(now);
+        }
+    }
+
     /// The PING with `nonce` to `address` (from `due`) goes now: its round trip counts
     /// from here, not from when it was found due.
     fn sending(&mut self, address: &str, nonce: u64, now: Instant) {
@@ -1030,6 +1060,9 @@ impl Paths {
         debug!(from = %address, bytes = payload_len, message = ?message, "Received message");
         match message {
             ParsedMessage::Ping(nonce) => {
+                self.table
+                    .lock_or_recover()
+                    .peer_pinged(&address, Instant::now());
                 let paths = Arc::clone(&self);
                 let to = address.clone();
                 tokio::spawn(async move {
@@ -1508,6 +1541,51 @@ mod tests {
         assert!(!table.paths.contains_key("libp2p:z"));
     }
 
+    /// The two sides take turns pinging a path: the peer's PING makes the next ours, an
+    /// interval after it; after ours we wait half an interval more while the peer pings
+    /// too, and an interval where it doesn't.
+    #[test]
+    fn pings_take_turns_with_the_peers() {
+        let mut table = Table::default();
+        let now = Instant::now();
+        let every = table.probe_every;
+        let mut nonce = 0;
+        let mut next = || {
+            nonce += 1;
+            nonce
+        };
+        table.rank(paths(&["libp2p:x", "mycelium:x"]), now, true);
+        assert_eq!(table.due(now, &mut next).len(), 2);
+        table.paths.values_mut().for_each(|path| path.probe = None);
+        // The peer pings libp2p:x a little later: its next PING is ours, an interval on.
+        let theirs = now + Duration::from_secs(1);
+        table.peer_pinged("libp2p:x", theirs);
+        table.rank(paths(&["libp2p:x", "mycelium:x"]), now + every, true);
+        let due = |table: &mut Table, at, next: &mut dyn FnMut() -> u64| -> Vec<String> {
+            let due = table.due(at, next);
+            table.paths.values_mut().for_each(|path| path.probe = None);
+            due.into_iter().map(|(address, _)| address).collect()
+        };
+        assert_eq!(due(&mut table, now + every, &mut next), ["mycelium:x"]);
+        assert_eq!(due(&mut table, theirs + every, &mut next), ["libp2p:x"]);
+        // After ours, while the peer pings: an interval and a half.
+        let ours = theirs + every;
+        table.rank(paths(&["libp2p:x", "mycelium:x"]), ours + every, true);
+        assert!(!due(&mut table, ours + every, &mut next).contains(&"libp2p:x".to_owned()));
+        table.peer_pinged("libp2p:x", ours + every + Duration::from_secs(2));
+        assert!(
+            !due(&mut table, ours + every + every / 2, &mut next).contains(&"libp2p:x".to_owned())
+        );
+        let x = || "libp2p:x".to_owned();
+        let ours = ours + every * 2 + Duration::from_secs(2);
+        assert!(due(&mut table, ours, &mut next).contains(&x()));
+        // A peer that stopped pinging: half an interval more once, then an interval.
+        assert!(!due(&mut table, ours + every, &mut next).contains(&x()));
+        let ours = ours + every + every / 2;
+        assert!(due(&mut table, ours, &mut next).contains(&x()));
+        assert!(due(&mut table, ours + every, &mut next).contains(&x()));
+    }
+
     /// Blocks one address, both ways.
     struct Blocking(&'static str);