From 22e539d9285626ed217c661e6a5e5d2adb12150c Mon Sep 17 00:00:00 2001 From: tv Date: Mon, 2 Nov 2026 22:59:58 +0100 Subject: swarm: announce to whom writers can't reach --- docs/architecture.md | 3 ++- docs/peers.md | 5 ++++- src/swarm/relay.rs | 56 ++++++++++++++++++++++++++++++++++++++--------- src/swarm/topology/mod.rs | 47 ++++++++++++++++++++++++--------------- 4 files changed, 81 insertions(+), 30 deletions(-) diff --git a/docs/architecture.md b/docs/architecture.md index e0e2c94..07edbfb 100644 --- a/docs/architecture.md +++ b/docs/architecture.md @@ -225,7 +225,8 @@ catch-up source. The phases, what each serves and how the files come in: [`lifec the write lock, stamped with the clock) → `publish_change` (bumps the version, emits `DocumentChanged`) → the sync worker generates a message per neighbour → each neighbour applies it, relays its new heads to *its* neighbours, which pull what they -lack. In a full mesh (up to `k + 1` members) the writer reaches everyone directly. +lack. In a full mesh (up to `k + 1` members) the writer reaches everyone directly, but +those on none of its transports, to whom the nodes between announce. **An inbound message.** Consumer → `handle_message`: drop our own and stale-run replies, admit (swarm key), mark the peer seen, record its epoch (ignoring late ones), diff --git a/docs/peers.md b/docs/peers.md index 95b4dab..c63cec1 100644 --- a/docs/peers.md +++ b/docs/peers.md @@ -284,7 +284,10 @@ the current pull is answered, if they still have something it lacks. The regular STATUS (every third keepalive) triggers the same pull, so gaps close within a few keepalive intervals. A node that just caught up announces its heads once. In a full mesh nothing is announced: the -writer reaches everyone (while a node doesn't know its neighbours yet, it announces). +writer reaches everyone (while a node doesn't know its neighbours yet, it announces); +but to the neighbours the writer can't reach, those on none of its transports (by the +member list, the writer's neighbours as this node computes them): a node on Tor and +libp2p announces to one on Tor alone what came from one on libp2p alone. ## Keepalive and self-healing diff --git a/src/swarm/relay.rs b/src/swarm/relay.rs index 1736cec..5ab93f2 100644 --- a/src/swarm/relay.rs +++ b/src/swarm/relay.rs @@ -16,7 +16,10 @@ //! message would announce to every neighbour, and announcements would outnumber the //! data they announce (about 15 to 1 at 250 nodes). //! -//! In a full mesh the writer reaches everyone itself, so nothing is announced. +//! In a full mesh the writer reaches everyone itself, so nothing is announced, but to +//! the neighbours the writer can't reach: those on none of its transports (a member on +//! Tor alone, reached through one on Tor and libp2p, from a writer on libp2p alone). +//! Whom it reaches is what the member list makes its neighbours (`topology::neighbours_of`). use std::collections::{HashMap, HashSet}; use std::sync::atomic::Ordering; @@ -44,7 +47,11 @@ pub(crate) async fn announce(state: &AppState, from: &str) { // While the view is unknown everyone counts as a neighbour, but that isn't a full // mesh: the others may not hear from the writer. if !state.initial_sync_done.load(Ordering::Acquire) - || (topology::view_known(state) && topology::is_full_mesh(state)) + || (topology::view_known(state) + && topology::is_full_mesh(state) + && unreached(state, &HashSet::from([from.to_owned()])) + .await + .is_empty()) { return; } @@ -80,26 +87,55 @@ pub(crate) async fn announce(state: &AppState, from: &str) { /// round with a single sender skips it; with several, each lacks the others' changes, /// so every neighbour is told. async fn announce_now(state: &AppState, senders: &HashSet) { + let neighbours = unreached(state, senders).await; + if neighbours.is_empty() { + return; + } state.stats.announce_round(); + for peer in neighbours { + if let Err(error) = send_status(state, &peer).await { + debug!(peer = %peer, error = %format!("{error:#}"), "Announcing our heads failed"); + } + } +} + +/// The neighbours to tell of the changes `senders` sent us. The peer whose changes we +/// announce has them already, so with a single sender it is skipped; with several, each +/// lacks the others' changes. In a full mesh, also those every sender is a neighbour of: +/// the writers reached them themselves. +async fn unreached(state: &AppState, senders: &HashSet) -> Vec { let skip = match senders.len() { 1 => senders.iter().next().cloned(), _ => None, }; - let neighbours: Vec = state + let full_mesh = topology::view_known(state) && topology::is_full_mesh(state); + // Unknown (a sender not in the member list): nobody counts as reached by it. + let reached: Vec>> = if full_mesh { + senders + .iter() + .map(|sender| topology::neighbours_of(state, sender)) + .collect() + } else { + Vec::new() + }; + let reached_by_all = |peer: &str| { + full_mesh + && reached.iter().zip(senders).all(|(neighbours, sender)| { + sender == peer || neighbours.as_ref().is_some_and(|n| n.contains(peer)) + }) + }; + state .peer_states .read() .await .keys() .filter(|peer| { - skip.as_deref() != Some(peer.as_str()) && topology::is_neighbour(state, peer) + skip.as_deref() != Some(peer.as_str()) + && topology::is_neighbour(state, peer) + && !reached_by_all(peer) }) .cloned() - .collect(); - for peer in neighbours { - if let Err(error) = send_status(state, &peer).await { - debug!(peer = %peer, error = %format!("{error:#}"), "Announcing our heads failed"); - } - } + .collect() } /// Pulls in progress: one neighbour at a time, so each change arrives about once. The diff --git a/src/swarm/topology/mod.rs b/src/swarm/topology/mod.rs index 30fa56e..946fc33 100644 --- a/src/swarm/topology/mod.rs +++ b/src/swarm/topology/mod.rs @@ -133,31 +133,42 @@ pub(crate) fn is_neighbour(state: &AppState, peer: &str) -> bool { pub(crate) fn intended_graph(state: &AppState) -> Option>> { let view = state.neighbours.read_or_recover(); let view = view.as_ref()?; - let unreachable = |member: &str| view.unreachable.contains(member); - let stable = |member: &str| view.stable.contains(member); let graph = view .members .iter() - .map(|member| { - let neighbours = if member == state.local_id() { - view.neighbours.clone() - } else { - let own = view.transports.get(member).cloned().unwrap_or_default(); - neighbours_across( - &view.transports, - member, - &own, - view.k, - &unreachable, - &stable, - ) - }; - (member.clone(), neighbours) - }) + .map(|member| (member.clone(), view.neighbours_of(state.local_id(), member))) .collect(); Some(graph) } +/// `member`'s neighbours as the member list makes them, as this node sees it (as in +/// `intended_graph`); None until the view is known, or for one the list doesn't have. +pub(crate) fn neighbours_of(state: &AppState, member: &str) -> Option> { + let view = state.neighbours.read_or_recover(); + let view = view.as_ref()?; + view.members + .iter() + .any(|listed| listed == member) + .then(|| view.neighbours_of(state.local_id(), member)) +} + +impl NeighbourView { + fn neighbours_of(&self, local: &str, member: &str) -> BTreeSet { + if member == local { + return self.neighbours.clone(); + } + let own = self.transports.get(member).cloned().unwrap_or_default(); + neighbours_across( + &self.transports, + member, + &own, + self.k, + &|member| self.unreachable.contains(member), + &|member| self.stable.contains(member), + ) + } +} + /// How the neighbours changed. pub(crate) struct NeighbourChange { pub(crate) added: Vec, -- cgit v1.3.1