suspect paths

5 files changed, 143 insertions, 18 deletions

diff --git a/docs/peers.md b/docs/peers.md
index e098b01..c41d431 100644
--- a/docs/peers.md
+++ b/docs/peers.md
@@ -19,7 +19,10 @@ go over it (a busy pair sends no keepalives). The
 configured order decides until both are measured; then the node changes to another
 path when it is a quarter faster (and at least 5 ms), so it doesn't flap. A send that
 fails goes to the next path at once, and a path whose sends fail, or whose PINGs go
-unanswered three times in a row, is down: used only when no other works. A node on
+unanswered three times in a row, is down: used only when no other works. A path that
+takes messages and drops them shows sooner: when what went to the peer goes unanswered
+(the PING that asks again, below, or an overdue sync answer), its other paths come
+first until that one answers a PING again, and all of them are pinged at once. A node on
 several transports bridges nodes that share none ([`transports.md`](transports.md), Several transports):
 the topology links it on each of its transports (below).
 
diff --git a/src/swarm/peer/worker.rs b/src/swarm/peer/worker.rs
index 399741c..b5c533a 100644
--- a/src/swarm/peer/worker.rs
+++ b/src/swarm/peer/worker.rs
@@ -111,6 +111,11 @@ pub(crate) async fn run_peer_worker(state: AppState) {
                 (became_dormant, keepalive, unanswered, ask_again)
             };
 
+            // Unanswered: the path in use may drop what it takes. Its others first, and
+            // all of them pinged now (`Transport::suspect`).
+            if unanswered.is_some() || ask_again {
+                state.transport.suspect(peer);
+            }
             if let Some(changes) = unanswered {
                 debug!(peer = %peer, changes, "No answer to our sync message; sending again");
                 crate::swarm::source::reply_timed_out(&state, peer);
diff --git a/src/transport/identity.rs b/src/transport/identity.rs
index 1df84ec..0b36967 100644
--- a/src/transport/identity.rs
+++ b/src/transport/identity.rs
@@ -610,6 +610,12 @@ impl Transport for IdentityTransport {
         self.shared.paths.rtt(&address)
     }
 
+    /// Its proven paths, the one in use first.
+    fn suspect(&self, peer: &str) {
+        let addresses = self.shared.proven_at(peer, false);
+        self.shared.paths.suspect(&addresses);
+    }
+
     /// The narrowest of its proven paths'.
     fn clock_offset(&self, peer: &str) -> Option<crate::transport::paths::ClockOffset> {
         self.shared
diff --git a/src/transport/mod.rs b/src/transport/mod.rs
index eeb9d34..474e386 100644
--- a/src/transport/mod.rs
+++ b/src/transport/mod.rs
@@ -1004,6 +1004,13 @@ pub(crate) trait Transport: Send + Sync {
         None
     }
 
+    /// What went to `peer` went unanswered for a while (`swarm::reach`): the path in use
+    /// may take messages and drop them. The paths layer tries the peer's other paths
+    /// first and pings them all now (`paths::Paths::suspect`). By default nothing.
+    fn suspect(&self, peer: &str) {
+        let _ = peer;
+    }
+
     /// 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 6f0a892..8c03b1f 100644
--- a/src/transport/paths.rs
+++ b/src/transport/paths.rs
@@ -21,7 +21,9 @@
 //! 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
 //! 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. `rank` puts paths that are up
+//! 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
+//! pinged at once. `rank` puts paths that are up
 //! first, the one in use first among them, and the rest by round-trip time; it changes
 //! the one in use when another is faster by `SWITCH_GAIN` (a quarter) and at least
 //! `SWITCH_MIN` (both measured), so it doesn't flap. Until both are measured, the
@@ -160,6 +162,9 @@ struct PathState {
     /// The peer's clock as its PONGs on this path say (the narrowest bound lately), and
     /// when that sample came.
     clock: Option<(ClockOffset, Instant)>,
+    /// What went over it went unanswered (`Paths::suspect`): the peer's other paths, if
+    /// up, come first until it answers a PING again.
+    suspect: bool,
 }
 
 impl PathState {
@@ -233,6 +238,13 @@ impl Table {
                 .get(address)
                 .is_some_and(|path| path.is_down(now))
         });
+        // Suspect paths after the others that are up, as long as any isn't.
+        let suspect =
+            |table: &Self, address: &str| table.paths.get(address).is_some_and(|path| path.suspect);
+        let (fine, doubted): (Vec<_>, Vec<_>) = up
+            .into_iter()
+            .partition(|(address, _)| !suspect(self, address));
+        let up: Vec<(String, usize)> = fine.into_iter().chain(doubted).collect();
         let mut set: Vec<&str> = up
             .iter()
             .chain(down.iter())
@@ -246,10 +258,12 @@ impl Table {
                 .chosen
                 .get(&key)
                 .filter(|chosen| up.iter().any(|(address, _)| address == *chosen))
+                .filter(|chosen| !suspect(self, chosen) || suspect(self, &up[0].0))
                 .cloned()
                 .unwrap_or_else(|| up[0].0.clone());
             let fastest = up
                 .iter()
+                .filter(|(address, _)| !suspect(self, address) || suspect(self, &current))
                 .filter_map(|(address, _)| Some((address, rtt(self, address)?)))
                 .min_by_key(|(_, rtt)| *rtt);
             if let (Some((fastest, fastest_rtt)), Some(current_rtt)) =
@@ -351,6 +365,40 @@ impl Table {
         due
     }
 
+    /// What went to the peer at `addresses` went unanswered: the first, the one in use,
+    /// is suspect (`PathState::suspect`), and each without a PING out gets one now.
+    /// Returns those PINGs, with their nonces (from `nonce`).
+    fn suspect(
+        &mut self,
+        addresses: &[String],
+        now: Instant,
+        mut nonce: impl FnMut() -> u64,
+    ) -> Vec<(String, u64)> {
+        if let Some(first) = addresses.first()
+            && addresses.len() > 1
+            && let Some(path) = self.paths.get_mut(first)
+            && !path.suspect
+        {
+            info!(address = %first, "Answers over a path are overdue; trying the peer's others first");
+            path.suspect = true;
+        }
+        let mut due = Vec::new();
+        for address in addresses {
+            if self.paths.len() >= MAX_PATHS && !self.paths.contains_key(address) {
+                continue;
+            }
+            let path = self.paths.entry(address.clone()).or_default();
+            if path.probe.is_none() {
+                let nonce = nonce();
+                path.probe = Some((nonce, now));
+                path.last_probe = Some(now);
+                self.counts.probes_sent += 1;
+                due.push((address.clone(), nonce));
+            }
+        }
+        due
+    }
+
     /// The node sent a PING with `nonce` to `address` (its keepalive): the probe of that
     /// path, which its PONG measures, even if it is a peer's only path; `due` then
     /// doesn't ping it again within `probe_every`.
@@ -408,6 +456,7 @@ impl Table {
         }
         path.unanswered = 0;
         path.failed_at = None;
+        path.suspect = false;
         path.rtt = Some(match path.rtt {
             Some(rtt) => (rtt * 3 + sample) / 4,
             None => sample,
@@ -913,9 +962,7 @@ impl Paths {
         sent
     }
 
-    /// Pings the paths due (`Table::due`), every `PROBE_TICK`, until dropped. Each PING
-    /// goes on its own, timed from when it goes: a send that takes long (a dial) neither
-    /// holds up the others nor counts in their round trips.
+    /// Pings the paths due (`Table::due`), every `PROBE_TICK`, until dropped.
     async fn probe(self: Arc<Self>) {
         let mut tick = tokio::time::interval(PROBE_TICK);
         tick.set_missed_tick_behavior(tokio::time::MissedTickBehavior::Delay);
@@ -925,25 +972,45 @@ impl Paths {
                 .table
                 .lock_or_recover()
                 .due(Instant::now(), random_nonce);
-            for (address, nonce) in due {
-                let paths = Arc::clone(&self);
-                tokio::spawn(async move {
+            Arc::clone(&self).ping(due);
+        }
+    }
+
+    /// Sends these PINGs. Each goes on its own, timed from when it goes: a send that
+    /// takes long (a dial) neither holds up the others nor counts in their round trips.
+    fn ping(self: Arc<Self>, due: Vec<(String, u64)>) {
+        for (address, nonce) in due {
+            let paths = Arc::clone(&self);
+            tokio::spawn(async move {
+                paths
+                    .table
+                    .lock_or_recover()
+                    .sending(&address, nonce, Instant::now());
+                if let Err(error) = paths.send_probe(&address, false, nonce).await {
+                    debug!(to = %address, error = %format!("{error:#}"), "Couldn't send a PING");
                     paths
                         .table
                         .lock_or_recover()
-                        .sending(&address, nonce, Instant::now());
-                    if let Err(error) = paths.send_probe(&address, false, nonce).await {
-                        debug!(to = %address, error = %format!("{error:#}"), "Couldn't send a PING");
-                        paths
-                            .table
-                            .lock_or_recover()
-                            .failed(&address, Instant::now());
-                    }
-                });
-            }
+                        .failed(&address, Instant::now());
+                }
+            });
         }
     }
 
+    /// What went to a peer at `addresses` (best first, as `rank` gives them) went
+    /// unanswered (`swarm::reach`): the path in use is suspect, and the peer's other
+    /// paths come first while they are up, until it answers a PING again; every path to
+    /// it gets a PING now, rather than at its next probe (up to `PROBE_EVERY`). A path
+    /// that takes messages and drops them (a transport that only queues them locally)
+    /// is otherwise down only after three unanswered probes, a minute and a half.
+    pub(crate) fn suspect(self: &Arc<Self>, addresses: &[String]) {
+        let due = self
+            .table
+            .lock_or_recover()
+            .suspect(addresses, Instant::now(), random_nonce);
+        Arc::clone(self).ping(due);
+    }
+
     /// A message from `address`, handed on to `handler`: a PING answered and a PONG
     /// noted first; dropped if the firewall blocks it.
     async fn receive(
@@ -1193,6 +1260,43 @@ mod tests {
         assert_eq!(table.rank(lan, now, true), ["libp2p:y", "mycelium:y"]);
     }
 
+    /// A path whose messages went unanswered gives way to the peer's other paths, each
+    /// pinged at once; answering a PING, it is trusted again, but the choice stays
+    /// where it went until it is clearly faster.
+    #[test]
+    fn suspect_paths_give_way() {
+        let mut table = Table::default();
+        let now = Instant::now();
+        let set = paths(&["libp2p:x", "mycelium:x"]);
+        assert_eq!(table.rank(set.clone(), now, true)[0], "libp2p:x");
+        let mut nonces = 0;
+        let pinged = table.suspect(
+            &["libp2p:x".to_string(), "mycelium:x".to_string()],
+            now,
+            || {
+                nonces += 1;
+                nonces
+            },
+        );
+        assert_eq!(pinged.len(), 2, "every path pinged");
+        assert_eq!(
+            table.rank(set.clone(), now, true),
+            ["mycelium:x", "libp2p:x"]
+        );
+        // It answers: no longer suspect, and nothing faster than the other.
+        for (address, nonce) in pinged {
+            table.pong(&address, nonce, now + Duration::from_millis(100));
+        }
+        assert_eq!(
+            table.rank(set.clone(), now, true),
+            ["mycelium:x", "libp2p:x"]
+        );
+        // A peer's only path: nothing to give way to, but pinged.
+        let only = table.suspect(&["mycelium:y".to_string()], now, || 9);
+        assert_eq!(only.len(), 1);
+        assert!(!table.paths["mycelium:y"].suspect);
+    }
+
     #[test]
     fn unanswered_or_failing_paths_are_down() {
         let mut table = Table::default();