swarm: announce to whom writers can't reach

2 files changed, 75 insertions, 28 deletions

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<String>) {
+    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<String>) -> Vec<String> {
     let skip = match senders.len() {
         1 => senders.iter().next().cloned(),
         _ => None,
     };
-    let neighbours: Vec<String> = 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<Option<std::collections::BTreeSet<String>>> = 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<BTreeMap<String, BTreeSet<String>>> {
     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<BTreeSet<String>> {
+    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<String> {
+        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<String>,