aboutsummaryrefslogtreecommitdiffstats
diff options
context:
space:
mode:
authortv <tv@krebsco.de>2026-11-01 21:00:18 +0100
committertv <tv@krebsco.de>2026-11-01 21:00:18 +0100
commitda18be3c9bbfdd5b3b4ff19f48863a7badf6074d (patch)
tree64e3db8fd39a59bbce8dd75b7381e48a9aff27c6
parent83b2f9efbbcedea3297fea6754ea4a0e7ac6b3e0 (diff)
path probes follow keepalive
-rw-r--r--src/node.rs1
-rw-r--r--src/transport/identity.rs4
-rw-r--r--src/transport/paths.rs45
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, _, _| {