measure paths

1 files changed, 517 insertions, 51 deletions

diff --git a/src/transport/paths.rs b/src/transport/paths.rs
index 3330b2d..30aab8d 100644
--- a/src/transport/paths.rs
+++ b/src/transport/paths.rs
@@ -11,10 +11,19 @@
 //! as configured (`--transports`).
 //!
 //! A message to a node with several addresses goes to the first of them that takes it
-//! (`send_first`), in the order `rank` gives: addresses whose last send failed, for
-//! `DOWN_FOR`, after the others, and among each the configured order. A send that fails
-//! is tried at once at the next address, the message encoded for its transport: when
-//! one transport fails, the next carries on within the same send.
+//! (`send_first`), in the order `rank` gives. A send that fails is tried at once at the
+//! next address, the message encoded for its transport: when one transport fails, the
+//! next carries on within the same send.
+//!
+//! Paths measures the addresses of such nodes itself: every `PROBE_EVERY`, each address
+//! it had a choice of within `ACTIVE_FOR` gets a PING with a random nonce, which the
+//! other side answers with a PONG at once, 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
+//! 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
+//! configured order decides. PING and PONG end here: the identity layer never sees them.
 
 use std::collections::HashMap;
 use std::sync::{Arc, Mutex};
@@ -22,12 +31,12 @@ use std::time::Duration;
 
 use anyhow::{Context, Result, anyhow};
 use tokio::time::Instant;
-use tracing::debug;
+use tracing::{debug, info};
 
 use crate::locks::MutexExt;
-
-use crate::transport::stats::TransportStats;
-use crate::transport::{Connection, EncodedMessage, ParsedMessage, Transport};
+use crate::transport::stats::{PathCounts, PathInfo, PathsStats, TransportStats};
+use crate::transport::traffic::Traffic;
+use crate::transport::{Connection, EncodedMessage, MessageKind, ParsedMessage, Transport};
 
 /// Called with the sender's address, the message and its `payload_len`.
 pub(crate) type PathHandler = dyn Fn(String, ParsedMessage, usize) -> std::pin::Pin<Box<dyn Future<Output = Result<()>> + Send>>
@@ -56,27 +65,259 @@ pub(crate) fn scheme_of(address: &str) -> &str {
     address.split_once(':').map_or("", |(scheme, _)| scheme)
 }
 
-/// An address whose send failed is ranked after the others for this long (unless a
+/// A path whose send failed is down for this long (unless it answers a PING, or a
 /// send there works again).
 const DOWN_FOR: Duration = Duration::from_secs(60);
+/// How often a path in use is pinged, how long an answer may take, and how many PINGs
+/// in a row unanswered make it down.
+const PROBE_EVERY: Duration = Duration::from_secs(30);
+const PROBE_TIMEOUT: Duration = Duration::from_secs(15);
+const UNANSWERED_DOWN: u32 = 3;
+/// How often the probe task looks for paths due.
+const PROBE_TICK: Duration = Duration::from_secs(5);
+/// Paths this node had a choice of within this long are probed; ones it didn't use for
+/// twice as long are forgotten.
+const ACTIVE_FOR: Duration = Duration::from_secs(300);
+/// The path in use is changed for one this much faster (a fraction of its round-trip
+/// time), and at least `SWITCH_MIN`.
+const SWITCH_GAIN: f64 = 0.25;
+const SWITCH_MIN: Duration = Duration::from_millis(5);
+/// Paths kept at most: they are of peers this node chose among, but bounded anyway.
+const MAX_PATHS: usize = 4096;
+
+/// What paths knows about one path (an address as paths keep it).
+#[derive(Debug, Default)]
+struct PathState {
+    /// When it was last among several to choose from (`rank`).
+    used: Option<Instant>,
+    /// Smoothed round-trip time of its PINGs.
+    rtt: Option<Duration>,
+    /// The PING out, by nonce, and when it went.
+    probe: Option<(u64, Instant)>,
+    last_probe: Option<Instant>,
+    /// PINGs in a row unanswered.
+    unanswered: u32,
+    /// When a send there last failed.
+    failed_at: Option<Instant>,
+}
+
+impl PathState {
+    fn is_down(&self, now: Instant) -> bool {
+        self.unanswered >= UNANSWERED_DOWN
+            || self
+                .failed_at
+                .is_some_and(|at| now.saturating_duration_since(at) < DOWN_FOR)
+    }
+}
+
+/// Every path's state, and which one is in use for each set to choose from.
+#[derive(Debug, Default)]
+struct Table {
+    paths: HashMap<String, PathState>,
+    /// By the set's addresses (sorted, joined): the one in use.
+    chosen: HashMap<String, String>,
+    counts: PathCounts,
+}
+
+impl Table {
+    /// `usable`, with each one's place in the configured order, best first (as
+    /// `Paths::rank` says), noting the choice.
+    fn rank(&mut self, mut usable: Vec<(String, usize)>, now: Instant) -> Vec<String> {
+        usable.sort_by_key(|(_, preference)| *preference);
+        usable.dedup_by(|a, b| a.0 == b.0);
+        if usable.len() < 2 {
+            return usable.into_iter().map(|(address, _)| address).collect();
+        }
+        if self.paths.len() >= MAX_PATHS {
+            self.forget_unused(now);
+        }
+        for (address, _) in &usable {
+            if self.paths.len() < MAX_PATHS || self.paths.contains_key(address) {
+                self.paths.entry(address.clone()).or_default().used = Some(now);
+            }
+        }
+        let rtt = |table: &Self, address: &str| table.paths.get(address).and_then(|path| path.rtt);
+        let (up, down): (Vec<_>, Vec<_>) = usable.into_iter().partition(|(address, _)| {
+            !self
+                .paths
+                .get(address)
+                .is_some_and(|path| path.is_down(now))
+        });
+        let mut set: Vec<&str> = up
+            .iter()
+            .chain(down.iter())
+            .map(|(address, _)| address.as_str())
+            .collect();
+        set.sort_unstable();
+        let key = set.join(" ");
+        let mut ranked: Vec<String> = Vec::new();
+        if !up.is_empty() {
+            let mut current = self
+                .chosen
+                .get(&key)
+                .filter(|chosen| up.iter().any(|(address, _)| address == *chosen))
+                .cloned()
+                .unwrap_or_else(|| up[0].0.clone());
+            let fastest = up
+                .iter()
+                .filter_map(|(address, _)| Some((address, rtt(self, address)?)))
+                .min_by_key(|(_, rtt)| *rtt);
+            if let (Some((fastest, fastest_rtt)), Some(current_rtt)) =
+                (fastest, rtt(self, &current))
+                && *fastest != current
+                && current_rtt.saturating_sub(fastest_rtt)
+                    >= SWITCH_MIN.max(current_rtt.mul_f64(SWITCH_GAIN))
+            {
+                info!(
+                    from = %current,
+                    to = %fastest,
+                    from_rtt = ?current_rtt,
+                    to_rtt = ?fastest_rtt,
+                    "Switching to a faster path"
+                );
+                self.counts.switches += 1;
+                current = fastest.clone();
+            }
+            if self.chosen.len() < MAX_PATHS || self.chosen.contains_key(&key) {
+                self.chosen.insert(key, current.clone());
+            }
+            // The rest: measured ones by round-trip time, then unmeasured ones in the
+            // configured order.
+            let mut rest: Vec<&(String, usize)> = up
+                .iter()
+                .filter(|(address, _)| *address != current)
+                .collect();
+            rest.sort_by_key(|(address, preference)| {
+                (
+                    rtt(self, address).is_none(),
+                    rtt(self, address),
+                    *preference,
+                )
+            });
+            ranked.push(current);
+            ranked.extend(rest.into_iter().map(|(address, _)| address.clone()));
+        }
+        ranked.extend(down.into_iter().map(|(address, _)| address));
+        ranked
+    }
+
+    /// A send to `address` failed.
+    fn failed(&mut self, address: &str, now: Instant) {
+        self.counts.send_failures += 1;
+        if let Some(path) = self.paths.get_mut(address) {
+            path.failed_at = Some(now);
+        }
+    }
+
+    /// A send to `address` worked.
+    fn sent(&mut self, address: &str) {
+        if let Some(path) = self.paths.get_mut(address) {
+            path.failed_at = None;
+        }
+    }
+
+    /// The PINGs due now, with their nonces (from `nonce`): to every path used within
+    /// `ACTIVE_FOR` that wasn't pinged within `PROBE_EVERY` and has no PING out. A PING
+    /// out for `PROBE_TIMEOUT` counts as unanswered.
+    fn due(&mut self, now: Instant, mut nonce: impl FnMut() -> u64) -> Vec<(String, u64)> {
+        self.forget_unused(now);
+        let mut due = Vec::new();
+        for (address, path) in &mut self.paths {
+            if let Some((_, sent)) = path.probe
+                && now.saturating_duration_since(sent) >= PROBE_TIMEOUT
+            {
+                path.probe = None;
+                path.unanswered += 1;
+                self.counts.probes_unanswered += 1;
+                if path.unanswered == UNANSWERED_DOWN {
+                    info!(address = %address, "A path stopped answering PINGs; using others");
+                }
+            }
+            let active = path
+                .used
+                .is_some_and(|used| now.saturating_duration_since(used) < ACTIVE_FOR);
+            let rested = path
+                .last_probe
+                .is_none_or(|last| now.saturating_duration_since(last) >= PROBE_EVERY);
+            if active && rested && 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
+    }
+
+    /// A PONG with `nonce` from `address`: the round trip, if it answers our PING out.
+    fn pong(&mut self, address: &str, nonce: u64, now: Instant) -> Option<Duration> {
+        let path = self.paths.get_mut(address)?;
+        let (expected, sent) = path.probe?;
+        if expected != nonce {
+            return None;
+        }
+        let sample = now.saturating_duration_since(sent);
+        path.probe = None;
+        if path.unanswered >= UNANSWERED_DOWN {
+            info!(address = %address, "A path answers PINGs again");
+        }
+        path.unanswered = 0;
+        path.failed_at = None;
+        path.rtt = Some(match path.rtt {
+            Some(rtt) => (rtt * 3 + sample) / 4,
+            None => sample,
+        });
+        self.counts.probes_answered += 1;
+        path.rtt
+    }
+
+    /// Forgets the paths not used for twice `ACTIVE_FOR`, and choices among them.
+    fn forget_unused(&mut self, now: Instant) {
+        self.paths.retain(|_, path| {
+            path.used
+                .is_some_and(|used| now.saturating_duration_since(used) < ACTIVE_FOR * 2)
+        });
+        let paths = &self.paths;
+        self.chosen
+            .retain(|key, _| key.split(' ').all(|address| paths.contains_key(address)));
+    }
+
+    fn info(&self, address: &str, now: Instant) -> PathInfo {
+        let path = self.paths.get(address);
+        PathInfo {
+            address: address.to_owned(),
+            rtt_ms: path
+                .and_then(|path| path.rtt)
+                .map(|rtt| rtt.as_secs_f64() * 1000.0),
+            down: path.is_some_and(|path| path.is_down(now)),
+        }
+    }
+}
 
 pub(crate) struct Paths {
     /// In order of preference.
     links: Vec<Link>,
-    /// Addresses whose last send failed, and when (`DOWN_FOR`). A std mutex: never held
-    /// across an await.
-    down: Mutex<HashMap<String, Instant>>,
+    /// PINGs and PONGs are counted here, under the address (`traffic`).
+    traffic: Option<Arc<Traffic>>,
+    /// A std mutex: never held across an await.
+    table: Mutex<Table>,
 }
 
 impl Paths {
-    /// Over `transports`, by scheme, in order of preference.
-    pub(crate) fn new(transports: Vec<(&'static str, Arc<dyn Transport>)>) -> Self {
+    /// Over `transports`, by scheme, in order of preference; PINGs and PONGs counted in
+    /// `traffic`.
+    pub(crate) fn new(
+        transports: Vec<(&'static str, Arc<dyn Transport>)>,
+        traffic: Option<Arc<Traffic>>,
+    ) -> Self {
         Self {
             links: transports
                 .into_iter()
                 .map(|(scheme, inner)| Link { scheme, inner })
                 .collect(),
-            down: Mutex::default(),
+            traffic,
+            table: Mutex::default(),
         }
     }
 
@@ -123,26 +364,25 @@ impl Paths {
             .unwrap_or(usize::MAX)
     }
 
-    /// The usable ones of `addresses` (as paths keep them), best first: those whose
-    /// last send failed (within `DOWN_FOR`) last, and by the configured order.
+    /// The usable ones of `addresses` (as paths keep them), best first: paths that are
+    /// up before those down, the one in use first, then by round-trip time, then in
+    /// the configured order (module docs). Several addresses are paths to measure.
     pub(crate) fn rank(&self, addresses: &[String]) -> Vec<String> {
-        let mut usable: Vec<(bool, usize, &String)> = addresses
+        let usable: Vec<(String, usize)> = addresses
             .iter()
             .filter(|address| self.usable(address))
-            .map(|address| (false, self.preference(address), address))
+            .map(|address| (address.clone(), self.preference(address)))
             .collect();
-        {
-            let now = Instant::now();
-            let mut down = self.down.lock_or_recover();
-            down.retain(|_, at| now.duration_since(*at) < DOWN_FOR);
-            for (failed, _, address) in &mut usable {
-                *failed = down.contains_key(*address);
-            }
-        }
-        usable.sort_by_key(|(failed, preference, _)| (*failed, *preference));
-        usable
-            .into_iter()
-            .map(|(_, _, address)| address.clone())
+        self.table.lock_or_recover().rank(usable, Instant::now())
+    }
+
+    /// What paths knows about each of `addresses`.
+    pub(crate) fn info(&self, addresses: &[String]) -> Vec<PathInfo> {
+        let now = Instant::now();
+        let table = self.table.lock_or_recover();
+        addresses
+            .iter()
+            .map(|address| table.info(address, now))
             .collect()
     }
 
@@ -277,8 +517,11 @@ impl Paths {
             };
             match self.send(&encoded).await {
                 Ok(()) => {
-                    self.down.lock_or_recover().remove(address);
+                    let mut table = self.table.lock_or_recover();
+                    table.sent(address);
                     if !failures.is_empty() {
+                        table.counts.failovers += 1;
+                        drop(table);
                         debug!(
                             peer = %message.peer,
                             address = %address,
@@ -289,9 +532,7 @@ impl Paths {
                     return Ok(address.clone());
                 }
                 Err(error) => {
-                    self.down
-                        .lock_or_recover()
-                        .insert(address.clone(), Instant::now());
+                    self.table.lock_or_recover().failed(address, Instant::now());
                     failures.push((address.clone(), error));
                 }
             }
@@ -321,35 +562,136 @@ impl Paths {
         link.inner.connection(&target)
     }
 
-    /// Every transport's figures together.
+    /// Paths' own figures, and every transport's.
     pub(crate) fn stats(&self) -> TransportStats {
         let mut stats = TransportStats::default();
         for link in &self.links {
             stats.merge(link.inner.stats());
         }
+        let now = Instant::now();
+        let table = self.table.lock_or_recover();
+        let mut paths: Vec<PathInfo> = table
+            .paths
+            .keys()
+            .map(|address| table.info(address, now))
+            .collect();
+        paths.sort_by(|a, b| a.address.cmp(&b.address));
+        stats.paths = Some(PathsStats {
+            paths,
+            counts: table.counts,
+        });
         stats
     }
 
+    /// Sends a PING or PONG with `nonce` to `address`, counted (`traffic`).
+    async fn send_probe(&self, address: &str, pong: bool, nonce: u64) -> Result<()> {
+        let message = if pong {
+            ParsedMessage::Pong(nonce)
+        } else {
+            ParsedMessage::Ping(nonce)
+        };
+        let encoded = self.encode(address, &message)?;
+        let sent = self.send(&encoded).await;
+        if let Some(traffic) = &self.traffic {
+            match &sent {
+                Ok(()) => traffic.sent(address, encoded.kind, encoded.payload_len),
+                Err(_) => traffic.refused(address, encoded.kind, encoded.payload_len),
+            }
+        }
+        sent
+    }
+
+    /// 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);
+        loop {
+            tick.tick().await;
+            let due = self
+                .table
+                .lock_or_recover()
+                .due(Instant::now(), random_nonce);
+            for (address, nonce) in due {
+                if let Err(error) = self.send_probe(&address, false, nonce).await {
+                    debug!(to = %address, error = %format!("{error:#}"), "Couldn't send a PING");
+                    self.table
+                        .lock_or_recover()
+                        .failed(&address, Instant::now());
+                }
+            }
+        }
+    }
+
+    /// A message from `address`: a PING is answered, a PONG noted; anything else goes to
+    /// `handler`.
+    async fn receive(
+        self: Arc<Self>,
+        address: String,
+        message: ParsedMessage,
+        payload_len: usize,
+        handler: &PathHandler,
+    ) -> Result<()> {
+        match message {
+            ParsedMessage::Ping(nonce) => {
+                if let Some(traffic) = &self.traffic {
+                    traffic.received(&address, MessageKind::Ping, payload_len);
+                }
+                tokio::spawn(async move {
+                    if let Err(error) = self.send_probe(&address, true, nonce).await {
+                        debug!(to = %address, error = %format!("{error:#}"), "Couldn't answer a PING");
+                    }
+                });
+                Ok(())
+            }
+            ParsedMessage::Pong(nonce) => {
+                if let Some(traffic) = &self.traffic {
+                    traffic.received(&address, MessageKind::Pong, payload_len);
+                }
+                self.table
+                    .lock_or_recover()
+                    .pong(&address, nonce, Instant::now());
+                Ok(())
+            }
+            message => handler(address, message, payload_len).await,
+        }
+    }
+
     /// Every transport's consumer, each message handed to `handler` under the sender's
-    /// address; returns when the first one does.
-    pub(crate) async fn start_consumer(&self, handler: Arc<PathHandler>) -> Result<()> {
+    /// address (PINGs and PONGs end here), and the probe task; returns when the first
+    /// consumer does.
+    pub(crate) async fn start_consumer(self: &Arc<Self>, handler: Arc<PathHandler>) -> Result<()> {
+        let probing = tokio::spawn(Arc::clone(self).probe());
         let mut consumers: Vec<std::pin::Pin<Box<dyn Future<Output = Result<()>> + Send>>> =
             Vec::new();
         for link in &self.links {
             let inner = Arc::clone(&link.inner);
             let scheme = link.scheme;
             let handler = Arc::clone(&handler);
+            let paths = Arc::clone(self);
             consumers.push(Box::pin(async move {
                 inner
                     .start_consumer(Box::new(move |sender, message, payload_len| {
-                        handler(format!("{scheme}:{sender}"), message, payload_len)
+                        let paths = Arc::clone(&paths);
+                        let handler = Arc::clone(&handler);
+                        Box::pin(async move {
+                            paths
+                                .receive(
+                                    format!("{scheme}:{sender}"),
+                                    message,
+                                    payload_len,
+                                    &*handler,
+                                )
+                                .await
+                        })
                     }))
                     .await
             }));
         }
-        first(consumers)
+        let result = first(consumers)
             .await
-            .unwrap_or_else(|| Err(anyhow!("no transports")))
+            .unwrap_or_else(|| Err(anyhow!("no transports")));
+        probing.abort();
+        result
     }
 
     /// When the first transport exits.
@@ -369,6 +711,18 @@ impl Paths {
     }
 }
 
+/// A PING's nonce: random, so a PONG can't be made up for a PING not seen.
+fn random_nonce() -> u64 {
+    let mut bytes = [0u8; 8];
+    if getrandom::fill(&mut bytes).is_err() {
+        // No randomness: a nonce still tells answers apart, if not unguessably.
+        return std::time::SystemTime::now()
+            .duration_since(std::time::UNIX_EPOCH)
+            .map_or(0, |since| since.as_nanos() as u64);
+    }
+    u64::from_le_bytes(bytes)
+}
+
 /// `address: error; …` for each failed send.
 fn describe(failures: &[(String, anyhow::Error)]) -> String {
     failures
@@ -401,13 +755,16 @@ mod tests {
     use crate::transport::{HeadsSummary, Status};
 
     /// "a" on two sim nets, as `sim` (preferred) and `other`; "b" on both too.
-    fn two_nets() -> (Paths, Paths) {
+    fn two_nets() -> (Arc<Paths>, Arc<Paths>) {
         let (first, second) = (SimNet::new(1), SimNet::new(2));
         let on_both = |id: &str| {
-            Paths::new(vec![
-                ("sim", first.transport(id) as Arc<dyn Transport>),
-                ("other", second.transport(id) as Arc<dyn Transport>),
-            ])
+            Arc::new(Paths::new(
+                vec![
+                    ("sim", first.transport(id) as Arc<dyn Transport>),
+                    ("other", second.transport(id) as Arc<dyn Transport>),
+                ],
+                None,
+            ))
         };
         (on_both("a"), on_both("b"))
     }
@@ -436,6 +793,109 @@ mod tests {
         assert!(a.resolve("nobody").await.is_err());
     }
 
+    fn paths(addresses: &[&str]) -> Vec<(String, usize)> {
+        addresses
+            .iter()
+            .enumerate()
+            .map(|(preference, address)| (address.to_string(), preference))
+            .collect()
+    }
+
+    /// Answers every PING due `rtt` later, as from the path it went to (`rtts`).
+    fn answer(table: &mut Table, now: Instant, rtts: &[(&str, u64)]) -> Instant {
+        let mut nonces = 0;
+        let due = table.due(now, || {
+            nonces += 1;
+            nonces
+        });
+        for (address, nonce) in due {
+            let rtt = rtts
+                .iter()
+                .find(|(to, _)| *to == address)
+                .map(|(_, ms)| *ms);
+            if let Some(ms) = rtt {
+                table.pong(&address, nonce, now + Duration::from_millis(ms));
+            }
+        }
+        now + PROBE_EVERY
+    }
+
+    #[test]
+    fn the_configured_order_until_measured_then_the_faster_path() {
+        let mut table = Table::default();
+        let mut now = Instant::now();
+        let set = paths(&["libp2p:x", "mycelium:x"]);
+        assert_eq!(table.rank(set.clone(), now), ["libp2p:x", "mycelium:x"]);
+        // libp2p a bit slower: not enough to switch.
+        for _ in 0..8 {
+            now = answer(&mut table, now, &[("libp2p:x", 110), ("mycelium:x", 100)]);
+        }
+        assert_eq!(table.rank(set.clone(), now), ["libp2p:x", "mycelium:x"]);
+        assert_eq!(table.counts.switches, 0);
+        // Much slower: switched, and it stays switched while that holds.
+        for _ in 0..8 {
+            now = answer(&mut table, now, &[("libp2p:x", 300), ("mycelium:x", 100)]);
+        }
+        assert_eq!(table.rank(set.clone(), now), ["mycelium:x", "libp2p:x"]);
+        assert_eq!(table.rank(set.clone(), now), ["mycelium:x", "libp2p:x"]);
+        assert_eq!(table.counts.switches, 1);
+        // Small differences under SWITCH_MIN never switch.
+        let lan = paths(&["libp2p:y", "mycelium:y"]);
+        table.rank(lan.clone(), now);
+        for _ in 0..8 {
+            now = answer(&mut table, now, &[("libp2p:y", 4), ("mycelium:y", 1)]);
+        }
+        assert_eq!(table.rank(lan, now), ["libp2p:y", "mycelium:y"]);
+    }
+
+    #[test]
+    fn unanswered_or_failing_paths_are_down() {
+        let mut table = Table::default();
+        let mut now = Instant::now();
+        let set = paths(&["libp2p:x", "mycelium:x"]);
+        table.rank(set.clone(), now);
+        // libp2p stops answering: down after UNANSWERED_DOWN PINGs, then last.
+        for _ in 0..=UNANSWERED_DOWN {
+            now = answer(&mut table, now, &[("mycelium:x", 100)]);
+            table.rank(set.clone(), now);
+        }
+        assert!(table.info("libp2p:x", now).down);
+        assert_eq!(table.rank(set.clone(), now), ["mycelium:x", "libp2p:x"]);
+        // It answers again: up, and measured.
+        now = answer(&mut table, now, &[("libp2p:x", 50), ("mycelium:x", 100)]);
+        assert!(!table.info("libp2p:x", now).down);
+
+        // A failed send: down for DOWN_FOR, unless a send there works.
+        table.failed("mycelium:x", now);
+        assert!(table.info("mycelium:x", now).down);
+        assert!(!table.info("mycelium:x", now + DOWN_FOR).down);
+        table.sent("mycelium:x");
+        assert!(!table.info("mycelium:x", now).down);
+    }
+
+    #[test]
+    fn only_paths_in_use_are_pinged() {
+        let mut table = Table::default();
+        let now = Instant::now();
+        let mut nonce = 0;
+        let mut next = || {
+            nonce += 1;
+            nonce
+        };
+        // One address: nothing to choose, nothing to measure.
+        table.rank(paths(&["libp2p:z"]), now);
+        assert!(table.due(now, &mut next).is_empty());
+        table.rank(paths(&["libp2p:x", "mycelium:x"]), now);
+        assert_eq!(table.due(now, &mut next).len(), 2);
+        // Not again within PROBE_EVERY; a stale answer is ignored.
+        assert!(table.due(now + PROBE_TIMEOUT, &mut next).is_empty());
+        assert_eq!(table.pong("libp2p:x", 99, now), None);
+        // Unused for ACTIVE_FOR: no more PINGs; for twice as long: forgotten.
+        assert!(table.due(now + ACTIVE_FOR, &mut next).is_empty());
+        table.due(now + ACTIVE_FOR * 2, &mut next);
+        assert!(table.paths.is_empty() && table.chosen.is_empty());
+    }
+
     /// A transport whose sends all fail.
     struct Broken;
 
@@ -482,11 +942,17 @@ mod tests {
     #[tokio::test]
     async fn a_failed_send_goes_to_the_next_address() {
         let net = SimNet::new(3);
-        let a = Paths::new(vec![
-            ("broken", Arc::new(Broken) as Arc<dyn Transport>),
-            ("sim", net.transport("a") as Arc<dyn Transport>),
-        ]);
-        let b = Paths::new(vec![("sim", net.transport("b") as Arc<dyn Transport>)]);
+        let a = Paths::new(
+            vec![
+                ("broken", Arc::new(Broken) as Arc<dyn Transport>),
+                ("sim", net.transport("a") as Arc<dyn Transport>),
+            ],
+            None,
+        );
+        let b = Arc::new(Paths::new(
+            vec![("sim", net.transport("b") as Arc<dyn Transport>)],
+            None,
+        ));
         let (sender, mut received) = tokio::sync::mpsc::unbounded_channel();
         let handler: Arc<PathHandler> = Arc::new(move |from, _, _| {
             let sender = sender.clone();