always announce new heads

3 files changed, 62 insertions, 45 deletions

diff --git a/src/simtest/relay.rs b/src/simtest/relay.rs
index 9b016b8..dadeed3 100644
--- a/src/simtest/relay.rs
+++ b/src/simtest/relay.rs
@@ -64,3 +64,56 @@ async fn changes_reach_members_the_writer_cant_reach_at_once() {
         .await;
     }
 }
+
+/// One transport, a full mesh by the member list, but the writer x reaches b alone (as
+/// one behind NAT may): what x writes reaches c through b, at once, though the member
+/// list makes c x's neighbour too.
+#[tokio::test(flavor = "multi_thread", worker_threads = 4)]
+async fn changes_reach_members_the_writer_fails_to_reach_at_once() {
+    let keepalive = Duration::from_secs(5);
+    let slow = move |config: &mut crate::node::NodeConfig| {
+        config.timing = Timing::new(keepalive, SYNC_REPLY_TIMEOUT, DELIVERY_LIMIT);
+    };
+    let sim = Sim::with_identity(63);
+    // Nothing between x and c, either way.
+    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_with("b", &["sim"], &[], Start::Genesis, slow);
+    let c = sim.start_on_with("c", &["sim"], &["b"], Start::Join, slow);
+    let x = sim.start_on_with("x", &["sim"], &["b"], Start::Join, slow);
+    sim.wait_ready(&[&b, &c, &x], SOON).await;
+    let (c_id, x_id) = (c.local_id().to_owned(), x.local_id().to_owned());
+    {
+        let (b, c_id, x_id) = (&b, c_id.clone(), x_id.clone());
+        sim.wait_until(
+            "b knows its neighbours, a full mesh, x and c among them",
+            SOON,
+            move || {
+                let (c_id, x_id) = (c_id.clone(), x_id.clone());
+                async move {
+                    topology::view_known(b)
+                        && topology::is_full_mesh(b)
+                        && topology::is_neighbour(b, &c_id)
+                        && topology::is_neighbour(b, &x_id)
+                }
+            },
+        )
+        .await;
+    }
+    sim.wait_converged(&[&b, &c, &x], SOON).await;
+
+    write(&x, "y", json!(1)).await;
+    {
+        let c = &c;
+        sim.wait_until(
+            "c has x's write within a second",
+            Duration::from_secs(1),
+            move || async move { read(c, "y").await == Some(json!(1)) },
+        )
+        .await;
+    }
+}
diff --git a/src/swarm/relay.rs b/src/swarm/relay.rs
index 5ab93f2..1393d50 100644
--- a/src/swarm/relay.rs
+++ b/src/swarm/relay.rs
@@ -16,10 +16,11 @@
 //! 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, 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`).
+//! In a full mesh too: the member list makes every member the writer's neighbour, but
+//! that says nothing about whether the writer reaches them. One behind NAT may reach a
+//! single one of them, one on another transport (on Tor alone) none, and no node can
+//! tell which from where it is. A neighbour that has the changes already does nothing
+//! with an announcement but answer it.
 
 use std::collections::{HashMap, HashSet};
 use std::sync::atomic::Ordering;
@@ -44,15 +45,7 @@ const ANNOUNCE_DELAY: Duration = Duration::from_millis(200);
 /// After changes from `from` were applied and our heads changed: tell our other
 /// neighbours, in the next announcement round.
 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)
-            && unreached(state, &HashSet::from([from.to_owned()]))
-                .await
-                .is_empty())
-    {
+    if !state.initial_sync_done.load(Ordering::Acquire) {
         return;
     }
     let schedule = {
@@ -101,38 +94,19 @@ async fn announce_now(state: &AppState, senders: &HashSet<String>) {
 
 /// 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.
+/// lacks the others' changes.
 async fn unreached(state: &AppState, senders: &HashSet<String>) -> Vec<String> {
     let skip = match senders.len() {
         1 => senders.iter().next().cloned(),
         _ => None,
     };
-    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)
-                && !reached_by_all(peer)
+            skip.as_deref() != Some(peer.as_str()) && topology::is_neighbour(state, peer)
         })
         .cloned()
         .collect()
diff --git a/src/swarm/topology/mod.rs b/src/swarm/topology/mod.rs
index df75f46..e3479e3 100644
--- a/src/swarm/topology/mod.rs
+++ b/src/swarm/topology/mod.rs
@@ -67,6 +67,7 @@ pub(crate) fn view_known(state: &AppState) -> bool {
 }
 
 /// Whether every reachable member is our neighbour (also while the view is unknown).
+#[cfg(test)]
 pub(crate) fn is_full_mesh(state: &AppState) -> bool {
     state
         .neighbours
@@ -141,17 +142,6 @@ pub(crate) fn intended_graph(state: &AppState) -> Option<BTreeMap<String, BTreeS
     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))
-}
-
 /// The transports `member` is on, by the member list (the schemes of its addresses);
 /// ours as we are on them. None known: empty.
 pub(crate) fn transports_of(state: &AppState, member: &str) -> BTreeSet<String> {