path probes follow keepalive

3 files changed, 41 insertions, 9 deletions

diff --git a/src/node.rs b/src/node.rs
index eca0e40..844a44b 100644
--- a/src/node.rs
+++ b/src/node.rs
@@ -81,6 +81,7 @@ impl Node {
                 Arc::clone(&traffic),
                 Some(Arc::clone(&firewall)),
                 Arc::clone(&config.clock),
+                config.timing.keepalive,
             )) as Arc<dyn Transport>,
             None => config.transport,
         };
diff --git a/src/transport/identity.rs b/src/transport/identity.rs
index 3ac06cb..8ef0535 100644
--- a/src/transport/identity.rs
+++ b/src/transport/identity.rs
@@ -246,13 +246,14 @@ impl Book {
 impl IdentityTransport {
     /// Over `transport`, whose addresses have `config.scheme`, and `config.more`.
     /// `firewall`: the debug firewall, if any, which paths applies by address; `clock`:
-    /// the node's wall clock (`Paths::new`).
+    /// the node's wall clock; `keepalive`: the node's keepalive interval (`Paths::new`).
     pub(crate) fn new(
         transport: Arc<dyn Transport>,
         config: IdentityConfig,
         traffic: Arc<Traffic>,
         firewall: Option<Arc<Firewall>>,
         clock: Arc<crate::clock::Clock>,
+        keepalive: Duration,
     ) -> Self {
         let node_id = config.key.id();
         let shared = Arc::new_cyclic(|shared: &std::sync::Weak<Shared>| {
@@ -265,6 +266,7 @@ impl IdentityTransport {
                     Some(Arc::clone(&traffic)),
                     firewall.is_some().then_some(gate),
                     clock,
+                    keepalive,
                 )),
                 firewall,
                 key: config.key,
diff --git a/src/transport/paths.rs b/src/transport/paths.rs
index 8149781..8bb8533 100644
--- a/src/transport/paths.rs
+++ b/src/transport/paths.rs
@@ -16,7 +16,8 @@
 //! next carries on within the same send.
 //!
 //! Paths measures the addresses of such nodes itself (and every path the node's
-//! keepalive PINGs go over, below): every `PROBE_EVERY`, each address it had a choice of
+//! 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
 //! path's round-trip time (smoothed); `UNANSWERED_DOWN` PINGs in a row unanswered, or a
@@ -90,8 +91,9 @@ pub(crate) fn scheme_of(address: &str) -> &str {
 /// 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.
+/// How often a path in use is pinged at most (every keepalive interval, if that is
+/// longer: `Table::probe_every`), 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;
@@ -170,15 +172,35 @@ impl PathState {
 }
 
 /// Every path's state, and which one is in use for each set to choose from.
-#[derive(Debug, Default)]
+#[derive(Debug)]
 struct Table {
     paths: HashMap<String, PathState>,
     /// By the set's addresses (sorted, joined): the one in use.
     chosen: HashMap<String, String>,
     counts: PathCounts,
+    /// How often a path in use is pinged: `PROBE_EVERY`, or the node's keepalive
+    /// interval if that is longer, so that the probes don't add to the keepalives a
+    /// long interval is chosen to spare. A busy pair, whose keepalives don't go, still
+    /// gets its paths measured this often.
+    probe_every: Duration,
+}
+
+impl Default for Table {
+    fn default() -> Self {
+        Self::new(PROBE_EVERY)
+    }
 }
 
 impl Table {
+    fn new(keepalive: Duration) -> Self {
+        Self {
+            paths: HashMap::new(),
+            chosen: HashMap::new(),
+            counts: PathCounts::default(),
+            probe_every: PROBE_EVERY.max(keepalive),
+        }
+    }
+
     /// `usable`, with each one's place in the configured order, best first (as
     /// `Paths::rank` says). With `sending`, for a message about to go: they count as
     /// used (`ACTIVE_FOR`), and the one in use may change for a faster one. Without, a
@@ -295,7 +317,7 @@ impl Table {
 
     /// The PINGs due now, with their nonces (from `nonce`): to every path chosen among
     /// others or sent over within `ACTIVE_FOR` that wasn't pinged (by us, or by a
-    /// keepalive) within `PROBE_EVERY` and has no PING out. A PING out for
+    /// keepalive) 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);
@@ -317,7 +339,7 @@ impl Table {
             let active = lately(path.used) || lately(path.carried);
             let rested = path
                 .last_probe
-                .is_none_or(|last| now.saturating_duration_since(last) >= PROBE_EVERY);
+                .is_none_or(|last| now.saturating_duration_since(last) >= self.probe_every);
             if active && rested && path.probe.is_none() {
                 let nonce = nonce();
                 path.probe = Some((nonce, now));
@@ -454,12 +476,14 @@ pub(crate) struct Paths {
 
 impl Paths {
     /// Over `transports`, by scheme, in order of preference; its own PINGs and PONGs
-    /// counted in `traffic`, the firewall through `gate`, PONGs stamped by `clock`.
+    /// counted in `traffic`, the firewall through `gate`, PONGs stamped by `clock`;
+    /// paths pinged every `keepalive` at least (`Table::probe_every`).
     pub(crate) fn new(
         transports: Vec<(&'static str, Arc<dyn Transport>)>,
         traffic: Option<Arc<Traffic>>,
         gate: Option<Weak<dyn Gate>>,
         clock: Arc<Clock>,
+        keepalive: Duration,
     ) -> Self {
         Self {
             links: transports
@@ -469,7 +493,7 @@ impl Paths {
             traffic,
             gate,
             clock,
-            table: Mutex::default(),
+            table: Mutex::new(Table::new(keepalive)),
         }
     }
 
@@ -1049,6 +1073,7 @@ mod tests {
                 None,
                 None,
                 Arc::default(),
+                Duration::ZERO,
             ))
         };
         (on_both("a"), on_both("b"))
@@ -1379,12 +1404,14 @@ mod tests {
             None,
             Some(Arc::downgrade(&gate)),
             Arc::default(),
+            Duration::ZERO,
         ));
         let a = Arc::new(Paths::new(
             vec![("sim", net.transport("a") as Arc<dyn Transport>)],
             None,
             None,
             Arc::default(),
+            Duration::ZERO,
         ));
         let received = |paths: &Arc<Paths>| {
             let (sender, received) = tokio::sync::mpsc::unbounded_channel();
@@ -1479,12 +1506,14 @@ mod tests {
             None,
             None,
             Arc::default(),
+            Duration::ZERO,
         );
         let b = Arc::new(Paths::new(
             vec![("sim", net.transport("b") as Arc<dyn Transport>)],
             None,
             None,
             Arc::default(),
+            Duration::ZERO,
         ));
         let (sender, mut received) = tokio::sync::mpsc::unbounded_channel();
         let handler: Arc<PathHandler> = Arc::new(move |from, _, _| {