From 5581992d5603322d7f02bd76625334be3f8a2b18 Mon Sep 17 00:00:00 2001 From: tv Date: Tue, 3 Nov 2026 04:44:49 +0100 Subject: measure 2-hop neighbourhood --- src/api/peers.rs | 17 ++++++ src/node.rs | 1 + src/simtest/identity.rs | 1 + src/simtest/mod.rs | 1 + src/simtest/reach.rs | 52 ++++++++++++++++++ src/simtest/timing.rs | 1 + src/state/mod.rs | 2 + src/state/peer.rs | 5 ++ src/swarm/mod.rs | 1 + src/swarm/peer/keepalive.rs | 13 +++-- src/swarm/peer/worker.rs | 1 + src/swarm/reach.rs | 119 ++++++++++++++++++++++++++++++++++++++++++ src/swarm/relay.rs | 6 +-- src/swarm/topology/graph.rs | 2 +- src/swarm/topology/mod.rs | 2 +- src/transport/libp2p/tests.rs | 1 + src/transport/mod.rs | 4 ++ src/transport/mycelium/mod.rs | 1 + src/transport/paths.rs | 1 + src/transport/sim.rs | 1 + src/transport/wire.rs | 8 ++- 21 files changed, 230 insertions(+), 10 deletions(-) create mode 100644 src/simtest/reach.rs create mode 100644 src/swarm/reach.rs (limited to 'src') diff --git a/src/api/peers.rs b/src/api/peers.rs index 537ad98..cd70f7d 100644 --- a/src/api/peers.rs +++ b/src/api/peers.rs @@ -60,6 +60,10 @@ pub(crate) struct PeerStatus { pub(crate) heads: Option, /// We keep it alive and sync with it on our own; other peers we only answer. pub(crate) neighbour: bool, + /// The members it said it reaches (`swarm::reach`), by node ID, while that holds; + /// absent before it said so, or once it is stale. Keys of members not in our member + /// list are left out. + pub(crate) reaches: Option>, /// Messages and payload bytes exchanged with it, by kind, since we started (or /// since it was last listed). pub(crate) traffic: PeerTraffic, @@ -92,6 +96,8 @@ pub(crate) struct PeerListResponse { pub(crate) local_heads: usize, /// Where this node can be reached (`Transport::local_addresses`). pub(crate) local_addresses: Vec, + /// The members we reach (`swarm::reach`), as we last told our neighbours. + pub(crate) local_reaches: Vec, /// By node ID. pub(crate) peers: BTreeMap, } @@ -117,6 +123,15 @@ pub(crate) async fn peer_list(state: &AppState) -> PeerListResponse { let now = Instant::now(); let local_epoch = state.local_epoch(); let mut aliases = crate::swarm::members::aliases(&state.doc.read().await.doc); + let names = crate::swarm::reach::names(&state).await; + let named = |keys: Vec| -> Vec { + let mut named: Vec = keys + .iter() + .filter_map(|key| names.get(key).cloned()) + .collect(); + named.sort(); + named + }; let mut peers = BTreeMap::new(); for (peer, peer_state_mutex) in snapshot { @@ -172,6 +187,7 @@ pub(crate) async fn peer_list(state: &AppState) -> PeerListResponse { same_epoch: ps.epoch.map(|epoch| epoch == local_epoch), heads: ps.reported_heads, neighbour, + reaches: crate::swarm::reach::of(&state, ps.reaches.as_ref()).map(&named), traffic, incidents: ps.incidents, sync_throughput, @@ -185,6 +201,7 @@ pub(crate) async fn peer_list(state: &AppState) -> PeerListResponse { local_alias: state.alias.clone(), local_addresses: state.transport.local_addresses(), local_epoch: local_epoch.into(), + local_reaches: named(crate::swarm::reach::published(&state)), local_heads, peers, } diff --git a/src/node.rs b/src/node.rs index 8b18cd4..b9ff348 100644 --- a/src/node.rs +++ b/src/node.rs @@ -141,6 +141,7 @@ impl Node { peers_due: Default::default(), pulling: Default::default(), announce_pending: Default::default(), + reach: Default::default(), rtt: Default::default(), throughput: Default::default(), stability: Default::default(), diff --git a/src/simtest/identity.rs b/src/simtest/identity.rs index 7379048..154c72e 100644 --- a/src/simtest/identity.rs +++ b/src/simtest/identity.rs @@ -112,6 +112,7 @@ fn status(node: &AppState) -> ParsedMessage { leaving: false, announcing: false, eager: false, + reaches: None, }) } diff --git a/src/simtest/mod.rs b/src/simtest/mod.rs index dc23440..e9d8c15 100644 --- a/src/simtest/mod.rs +++ b/src/simtest/mod.rs @@ -13,6 +13,7 @@ mod bench; mod deltas; mod identity; mod links; +mod reach; mod relay; mod remakes; mod soak; diff --git a/src/simtest/reach.rs b/src/simtest/reach.rs new file mode 100644 index 0000000..b38db87 --- /dev/null +++ b/src/simtest/reach.rs @@ -0,0 +1,52 @@ +//! Whom neighbours reach (`swarm::reach`): told in STATUSes, as measured. + +use std::time::Duration; + +use super::Sim; +use crate::state::AppState; +use crate::swarm::bootstrap::Start; +use crate::swarm::{reach, topology}; + +const SOON: Duration = Duration::from_secs(30); + +/// Whether `node` reaches `member`, as `node` measured and told it. +fn reaches(node: &AppState, member: &AppState) -> bool { + reach::published(node).contains(&topology::pair_key(member.local_id())) +} + +/// Whether `node` heard from `neighbour` that it reaches `member`. +async fn heard(node: &AppState, neighbour: &AppState, member: &AppState) -> bool { + let Some(peer_state) = node.get_peer_state(neighbour.local_id()).await else { + return false; + }; + let said = peer_state.lock().await.reaches.clone(); + reach::of(node, said.as_ref()) + .is_some_and(|keys| keys.contains(&topology::pair_key(member.local_id()))) +} + +/// x, b and c on one net, all members of one full mesh, but nothing passes between x +/// and c: b reaches both and says so; c learns from b that b reaches x, which c itself +/// doesn't. +#[tokio::test(flavor = "multi_thread", worker_threads = 4)] +async fn neighbours_tell_whom_they_reach() { + let sim = Sim::with_identity(66); + let cut = crate::transport::sim::LinkParams { + loss: 1.0, + ..Default::default() + }; + sim.net.set_link("x", "c", cut.clone()); + sim.net.set_link("c", "x", cut); + let b = sim.start_on("b", &["sim"], &[], Start::Genesis); + let c = sim.start_on("c", &["sim"], &["b"], Start::Join); + let x = sim.start_on("x", &["sim"], &["b"], Start::Join); + sim.wait_ready(&[&b, &c, &x], SOON).await; + let (b, c, x) = (&b, &c, &x); + sim.wait_until( + "c heard from b that b reaches x, and c doesn't reach x", + SOON, + move || async move { + reaches(b, x) && reaches(b, c) && !reaches(c, x) && heard(c, b, x).await + }, + ) + .await; +} diff --git a/src/simtest/timing.rs b/src/simtest/timing.rs index ea59689..d06e992 100644 --- a/src/simtest/timing.rs +++ b/src/simtest/timing.rs @@ -337,6 +337,7 @@ async fn late_epoch_does_not_flip_a_peers_epoch() { leaving: false, announcing: false, eager: false, + reaches: None, }) }; let a = &a; diff --git a/src/state/mod.rs b/src/state/mod.rs index fef6448..92f7810 100644 --- a/src/state/mod.rs +++ b/src/state/mod.rs @@ -152,6 +152,8 @@ pub(crate) struct AppState { /// An announcement round is scheduled, for changes from these peers /// (`swarm::relay`). pub(crate) announce_pending: Arc>>>, + /// Whom we reach, as told our neighbours (`swarm::reach`). + pub(crate) reach: Arc>, /// Our neighbours, once known (`swarm::topology`). pub(crate) neighbours: Arc>>, /// Something the neighbours are computed from may have changed: the member list or a diff --git a/src/state/peer.rs b/src/state/peer.rs index c6558d7..1e09079 100644 --- a/src/state/peer.rs +++ b/src/state/peer.rs @@ -151,6 +151,8 @@ pub(crate) struct PeerState { /// says so). Both lazy (announced only) until a pull makes ours eager. pub(crate) eager_to: bool, pub(crate) eager_from: bool, + /// Whom it said it reaches (`swarm::reach`), by pair key, and when. + pub(crate) reaches: Option<(Instant, Vec)>, /// It said farewell (`say_farewell`): what else that run of it still sends while it /// stops doesn't make it active again; a message from another run does (`returned`). @@ -244,6 +246,7 @@ impl PeerState { last_heads: None, eager_to: false, eager_from: false, + reaches: None, left: false, hello_attempts: 0, unreachable_since: None, @@ -262,6 +265,8 @@ impl PeerState { // The link starts lazy again: a pull makes it eager when it proves itself. self.eager_to = false; self.eager_from = false; + // Said over the session that ended. + self.reaches = None; } /// An overdue sync answer from an established, active peer (see diff --git a/src/swarm/mod.rs b/src/swarm/mod.rs index 8259ed5..322170b 100644 --- a/src/swarm/mod.rs +++ b/src/swarm/mod.rs @@ -10,6 +10,7 @@ pub(crate) mod links; pub(crate) mod members; pub(crate) mod pair; pub(crate) mod peer; +pub(crate) mod reach; pub(crate) mod relay; pub(crate) mod rtt; pub(crate) mod skew; diff --git a/src/swarm/peer/keepalive.rs b/src/swarm/peer/keepalive.rs index 37fcba8..c04b1ad 100644 --- a/src/swarm/peer/keepalive.rs +++ b/src/swarm/peer/keepalive.rs @@ -34,11 +34,14 @@ pub(crate) async fn send_status(state: &AppState, peer: &str) -> Result<()> { /// Sends `peer` a STATUS announcing our new heads (`swarm::relay`): no keepalive, so it /// leaves the keepalives' schedule as it is, and it is answered only if the peer's heads -/// differ (`Status::announcing`). -pub(crate) async fn announce_heads(state: &AppState, peer: &str) -> Result<()> { +/// differ (`Status::announcing`). Whom we reach only `with_reaches` (`swarm::reach`): +/// announcements of heads go often. +pub(crate) async fn announce_heads(state: &AppState, peer: &str, with_reaches: bool) -> Result<()> { + let local = local_status(state, peer).await; let status = transport::Status { announcing: true, - ..local_status(state, peer).await + reaches: local.reaches.filter(|_| with_reaches), + ..local }; state .transport @@ -192,6 +195,7 @@ pub(crate) async fn local_status(state: &AppState, peer: &str) -> transport::Sta leaving: state.is_closing(), announcing: false, eager, + reaches: Some(crate::swarm::reach::published(state)), } } @@ -229,6 +233,9 @@ pub(super) async fn note_status(state: &AppState, peer: &str, remote: &transport peer_state.reported_heads = remote.heads.heads().map(<[_]>::len); peer_state.last_heads = Some((remote.epoch, remote.heads.clone())); peer_state.eager_to = remote.eager; + if let Some(reaches) = &remote.reaches { + peer_state.reaches = Some((Instant::now(), reaches.clone())); + } drop(peer_state); if left { info!(peer = %peer, "Peer said farewell; dormant until it is heard from again"); diff --git a/src/swarm/peer/worker.rs b/src/swarm/peer/worker.rs index a803922..45e8d8f 100644 --- a/src/swarm/peer/worker.rs +++ b/src/swarm/peer/worker.rs @@ -164,6 +164,7 @@ pub(crate) async fn run_peer_worker(state: AppState) { forget_strangers(&state).await; crate::swarm::members::ensure_registered(&state).await; crate::swarm::stability::tick(&state).await; + crate::swarm::reach::check(&state).await; crate::swarm::skew::check(&state).await; crate::swarm::stats::measure_saved_size(&state); crate::swarm::source::warn_if_stalled(&state); diff --git a/src/swarm/reach.rs b/src/swarm/reach.rs new file mode 100644 index 0000000..17b189d --- /dev/null +++ b/src/swarm/reach.rs @@ -0,0 +1,119 @@ +//! Whom each neighbour reaches: a node's 2-hop neighbourhood, as measured, not as the +//! member list says. +//! +//! A node knows its own links, and from them whom it reaches: the members it has an +//! active session with (handshake done, neither dormant nor unreachable). It tells its +//! neighbours so in its STATUS (`Status::reaches`, each member by its pair key, +//! `topology::pair_key`, 8 bytes instead of a node ID): in every keepalive STATUS and +//! STATUS_REPLY, and soon (the peer worker looks every 5 s) to every neighbour when it +//! changed, +//! but not in announcements of heads, which go often. So every node knows whom its +//! neighbours reach, as of their latest STATUS: for as long as that is fresh +//! (`held_for`), then not, until the next. Nothing goes into the document, and nobody +//! knows more than two hops out. +//! +//! What it is for: passing changes on around links that don't work, and to exactly the +//! members the sender doesn't reach; a member a neighbour reaches is one worth trying, +//! and a way to it. Not for choosing multipoint relays (as OLSR does): over the member +//! list's graph, which is close to random, neighbourhoods hardly overlap, and nearly +//! every neighbour would be one (13.7 of 13.7 at 250 members and more), saving next to +//! nothing but in a full mesh. + +use std::collections::HashMap; +use std::time::Duration; + +use tokio::time::Instant; +use tracing::debug; + +use crate::locks::MutexExt; +use crate::state::{AppState, HelloState, Standing}; +use crate::swarm::peer::announce_heads; +use crate::swarm::{members, topology}; + +/// Whom we reach, as last told our neighbours. +#[derive(Default)] +pub(crate) struct Reach { + published: Option>, +} + +/// The pair keys of the members we reach now, sorted: those with an active session. +async fn measure(state: &AppState) -> Vec { + let mut reached = Vec::new(); + for (peer, peer_state) in state.peers_snapshot().await { + let peer_state = peer_state.lock().await; + if peer_state.hello_state == HelloState::Established + && !peer_state.liveness.is_dormant() + && !peer_state.is_unreachable() + && matches!(peer_state.standing, Standing::Member { .. }) + { + reached.push(topology::pair_key(&peer)); + } + } + reached.sort_unstable(); + reached +} + +/// Whom we reach, as we tell it (`Status::reaches`); empty until first measured. +pub(crate) fn published(state: &AppState) -> Vec { + state + .reach + .lock_or_recover() + .published + .clone() + .unwrap_or_default() +} + +/// With the peer worker's periodic pass (every 5 s): whom we reach now; if it changed, +/// our neighbours are told at once (a STATUS each, unanswered if the heads are the +/// same). A change that is undone within a pass isn't told. +pub(crate) async fn check(state: &AppState) { + let now = measure(state).await; + { + let mut reach = state.reach.lock_or_recover(); + if reach.published.as_ref() == Some(&now) { + return; + } + reach.published = Some(now.clone()); + } + debug!( + reached = now.len(), + "Whom we reach changed; telling our neighbours" + ); + for (peer, peer_state) in state.peers_snapshot().await { + let established = peer_state.lock().await.hello_state == HelloState::Established; + if !established || !topology::is_neighbour(state, &peer) { + continue; + } + if let Err(error) = tell(state, &peer).await { + debug!(peer = %peer, error = %format!("{error:#}"), "Telling whom we reach failed"); + } + } +} + +/// A STATUS to `peer` with whom we reach: as an announcement (unanswered if the heads +/// are the same), but with `reaches`. +async fn tell(state: &AppState, peer: &str) -> anyhow::Result<()> { + announce_heads(state, peer, true).await +} + +/// How long what a neighbour said it reaches holds: past two of its regular STATUSes, +/// each of which says it again. +pub(crate) fn held_for(state: &AppState) -> Duration { + state.timing.status_refresh * 2 + state.timing.keepalive +} + +/// The members `peer` reaches, as it last said, while that holds (`held_for`); by +/// pair key. +pub(crate) fn of(state: &AppState, said: Option<&(Instant, Vec)>) -> Option> { + said.filter(|(at, _)| at.elapsed() < held_for(state)) + .map(|(_, reaches)| reaches.clone()) +} + +/// The members of our member list by pair key, to name what neighbours say they reach. +pub(crate) async fn names(state: &AppState) -> HashMap { + let replica = state.doc.read().await; + members::members(&replica.doc) + .into_iter() + .map(|member| (topology::pair_key(&member), member)) + .collect() +} diff --git a/src/swarm/relay.rs b/src/swarm/relay.rs index e11bd69..69d73d3 100644 --- a/src/swarm/relay.rs +++ b/src/swarm/relay.rs @@ -122,7 +122,7 @@ async fn announce_now(state: &AppState, senders: &HashSet) { } state.stats.announce_round(); for peer in neighbours { - if let Err(error) = announce_heads(state, &peer).await { + if let Err(error) = announce_heads(state, &peer, false).await { debug!(peer = %peer, error = %format!("{error:#}"), "Announcing our heads failed"); } } @@ -268,7 +268,7 @@ async fn graft(state: &AppState, peer: &str) { } state.stats.link_grafted(); debug!(peer = %peer, "Link eager: we want its changes as it applies them"); - if let Err(error) = announce_heads(state, peer).await { + if let Err(error) = announce_heads(state, peer, false).await { debug!(peer = %peer, error = %format!("{error:#}"), "Telling a neighbour the link is eager failed"); } } @@ -281,7 +281,7 @@ pub(crate) async fn prune(state: &AppState, peer: &str) { } state.stats.link_pruned(); debug!(peer = %peer, "Link lazy: its changes reach us first another way"); - if let Err(error) = announce_heads(state, peer).await { + if let Err(error) = announce_heads(state, peer, false).await { debug!(peer = %peer, error = %format!("{error:#}"), "Telling a neighbour the link is lazy failed"); } } diff --git a/src/swarm/topology/graph.rs b/src/swarm/topology/graph.rs index fb33170..5401506 100644 --- a/src/swarm/topology/graph.rs +++ b/src/swarm/topology/graph.rs @@ -203,7 +203,7 @@ pub(crate) fn pair_score(a: &str, b: &str) -> u64 { } /// What a member contributes to its pair scores: computed once per member. -pub(super) fn pair_key(member: &str) -> u64 { +pub(crate) fn pair_key(member: &str) -> u64 { let digest = hash(PAIR_LABEL, &[member]); u64::from_be_bytes(digest[..8].try_into().expect("8 bytes")) } diff --git a/src/swarm/topology/mod.rs b/src/swarm/topology/mod.rs index e5dacdd..1c1a6c4 100644 --- a/src/swarm/topology/mod.rs +++ b/src/swarm/topology/mod.rs @@ -35,8 +35,8 @@ use crate::swarm::members; mod graph; -pub(crate) use graph::pair_score; use graph::*; +pub(crate) use graph::{pair_key, pair_score}; /// This node's current neighbours, and the member list they were computed from. #[derive(Debug, Default)] diff --git a/src/transport/libp2p/tests.rs b/src/transport/libp2p/tests.rs index 8f77bf7..7bedeaa 100644 --- a/src/transport/libp2p/tests.rs +++ b/src/transport/libp2p/tests.rs @@ -61,6 +61,7 @@ fn status() -> ParsedMessage { leaving: false, announcing: false, eager: false, + reaches: None, }) } diff --git a/src/transport/mod.rs b/src/transport/mod.rs index 22782ef..9ae8ff2 100644 --- a/src/transport/mod.rs +++ b/src/transport/mod.rs @@ -210,6 +210,10 @@ pub(crate) struct Status { /// sent as sync messages (an eager link, `swarm::relay`), not only announced. Every /// STATUS says so, or not, again: one lost changes nothing for long. pub(crate) eager: bool, + /// Whom the sender reaches (`swarm::reach`): the pair keys (`topology::pair_key`) + /// of the members it has an active session with, sorted. In keepalive STATUSes and + /// their replies, and when it changed; not in announcements of heads (None). + pub(crate) reaches: Option>, } /// Cheap to clone: large parts are shared (`bytes::Bytes`). diff --git a/src/transport/mycelium/mod.rs b/src/transport/mycelium/mod.rs index 0562e72..76ef8cd 100644 --- a/src/transport/mycelium/mod.rs +++ b/src/transport/mycelium/mod.rs @@ -719,6 +719,7 @@ mod tests { leaving: false, announcing: false, eager: false, + reaches: None, }; let payload = wire::encode(&ParsedMessage::StatusReply(status.clone())).unwrap(); // Mycelium's message data: a length byte, the topic, then our payload. diff --git a/src/transport/paths.rs b/src/transport/paths.rs index 1cd21e8..3e595ab 100644 --- a/src/transport/paths.rs +++ b/src/transport/paths.rs @@ -1515,6 +1515,7 @@ mod tests { leaving: false, announcing: false, eager: false, + reaches: None, }) } diff --git a/src/transport/sim.rs b/src/transport/sim.rs index af262aa..c9e45fb 100644 --- a/src/transport/sim.rs +++ b/src/transport/sim.rs @@ -389,6 +389,7 @@ mod tests { leaving: false, announcing: false, eager: false, + reaches: None, }) } diff --git a/src/transport/wire.rs b/src/transport/wire.rs index fe690a2..a6ca286 100644 --- a/src/transport/wire.rs +++ b/src/transport/wire.rs @@ -14,7 +14,7 @@ use super::{EncodedMessage, MessageKind, ParsedMessage}; /// The wire format's version, the first byte of every payload. Bump it with every change /// to a message type or a codec (`golden_bytes` fails then). -pub(crate) const VERSION: u8 = 18; +pub(crate) const VERSION: u8 = 21; /// What the second byte of a payload says. fn tag(kind: MessageKind) -> u8 { @@ -371,6 +371,7 @@ pub(crate) fn samples() -> Vec { leaving: false, announcing: false, eager: false, + reaches: None, }; let proof = IdentityProof { key: [8; 32], @@ -555,6 +556,7 @@ mod tests { leaving: false, announcing: false, eager: false, + reaches: None, }; let decoded = decode(&encode(&ParsedMessage::StatusReply(status.clone())).unwrap()).unwrap(); @@ -580,6 +582,7 @@ mod tests { leaving: false, announcing: false, eager: false, + reaches: None, }; assert!(decode(&encode(&ParsedMessage::Status(status)).unwrap()).is_err()); } @@ -679,10 +682,11 @@ mod tests { leaving: false, announcing: true, eager: false, + reaches: None, }); assert_eq!( hex(&encode(&status).unwrap()), - format!("{v}03989290c400{epoch_hex}01c3c2c2c3c2") + format!("{v}03999290c400{epoch_hex}01c3c2c2c3c2c0") ); let who = ParsedMessage::Who(IdentityProof { -- cgit v1.3.1