topology: introduce outbound members

3 files changed, 112 insertions, 1 deletions

diff --git a/src/simtest/timing.rs b/src/simtest/timing.rs
index 20ad15c..72c6c05 100644
--- a/src/simtest/timing.rs
+++ b/src/simtest/timing.rs
@@ -414,3 +414,77 @@ async fn forgotten_node_that_is_alive_cancels_its_record() {
     }
     sim.wait_converged(&[a, b, c], SOON).await;
 }
+
+/// A transport that only connects out: it publishes no address, as a libp2p node
+/// without a listen address (`topology::refresh`, `outbound`).
+struct OutboundOnly(std::sync::Arc<dyn crate::transport::Transport>);
+
+#[async_trait::async_trait]
+impl crate::transport::Transport for OutboundOnly {
+    fn local_id(&self) -> &str {
+        self.0.local_id()
+    }
+    fn is_peer_id(&self, id: &str) -> bool {
+        self.0.is_peer_id(id)
+    }
+    fn delivery_limit(&self) -> Duration {
+        self.0.delivery_limit()
+    }
+    fn local_addresses(&self) -> Vec<String> {
+        Vec::new()
+    }
+    async fn resolve_seed(&self, seed: &str) -> anyhow::Result<String> {
+        self.0.resolve_seed(seed).await
+    }
+    fn encode(
+        &self,
+        peer: &str,
+        message: &ParsedMessage,
+    ) -> anyhow::Result<crate::transport::EncodedMessage> {
+        self.0.encode(peer, message)
+    }
+    async fn send_encoded(&self, message: &crate::transport::EncodedMessage) -> anyhow::Result<()> {
+        self.0.send_encoded(message).await
+    }
+    async fn start_consumer(
+        &self,
+        handler: crate::transport::MessageHandler,
+    ) -> anyhow::Result<()> {
+        self.0.start_consumer(handler).await
+    }
+    async fn wait_for_exit(&self) -> std::io::Result<std::process::ExitStatus> {
+        self.0.wait_for_exit().await
+    }
+}
+
+/// A member with no address (it only connects out) is in nobody's graph but its own:
+/// those it picked keep it as a neighbour, so their changes reach it too, not only
+/// answers to its own.
+#[tokio::test(flavor = "multi_thread", worker_threads = 4)]
+async fn a_member_that_only_connects_out_gets_changes() {
+    let sim = Sim::new(7);
+    let a = sim.start("a", &[], Start::Genesis);
+    let b = sim.start_with("b", &["a"], Start::Join, |config| {
+        config.transport = std::sync::Arc::new(OutboundOnly(config.transport.clone()));
+    });
+    let c = sim.start("c", &["a"], Start::Join);
+    let (a, b, c) = (&a, &b, &c);
+    sim.wait_ready(&[a, b, c], SOON).await;
+    // Once b is listed (without an address), the member list no longer makes it a's
+    // or c's neighbour.
+    sim.wait_until("b is listed", SOON, move || async move {
+        read(a, "_continuum/members/b").await.is_some()
+            && read(c, "_continuum/members/b").await.is_some()
+    })
+    .await;
+    sim.wait_until("a and c keep b", SOON, move || async move {
+        has_neighbour(a, "b") && has_neighbour(c, "b")
+    })
+    .await;
+    write(a, "from/a", json!(1)).await;
+    write(c, "from/c", json!(3)).await;
+    sim.wait_until("b has a's and c's writes", SOON, move || async move {
+        read(b, "from/a").await == Some(json!(1)) && read(b, "from/c").await == Some(json!(3))
+    })
+    .await;
+}
diff --git a/src/swarm/peer/keepalive.rs b/src/swarm/peer/keepalive.rs
index ea7ade2..afc2b3d 100644
--- a/src/swarm/peer/keepalive.rs
+++ b/src/swarm/peer/keepalive.rs
@@ -171,9 +171,14 @@ pub(super) async fn note_status(state: &AppState, peer: &str, remote: &transport
         let mut peer_state = peer_state.lock().await;
         let became_ready = remote.ready && peer_state.reports_ready != Some(true);
         peer_state.reports_ready = Some(remote.ready);
-        peer_state.syncs_with_us = Some(remote.syncs_with_you);
+        let changed =
+            peer_state.syncs_with_us.replace(remote.syncs_with_you) != Some(remote.syncs_with_you);
         peer_state.reported_heads = remote.heads.heads().map(<[_]>::len);
         drop(peer_state);
+        if changed {
+            // A neighbour that only connects out stays one as long as it says so.
+            state.topology_changed.notify_one();
+        }
         // Off the consumer: telling which heads we lack takes the document's lock.
         {
             let (state, peer, remote) = (state.clone(), peer.to_owned(), remote.clone());
diff --git a/src/swarm/topology.rs b/src/swarm/topology.rs
index 1d98637..3731036 100644
--- a/src/swarm/topology.rs
+++ b/src/swarm/topology.rs
@@ -47,6 +47,10 @@ pub(crate) struct NeighbourView {
     /// Peers that proved they are of our swarm but aren't in our member list (yet): their
     /// registration hasn't reached us. They are neighbours too (`refresh`).
     unlisted: BTreeSet<String>,
+    /// Members the member list gives no address for on our transports, that reach us
+    /// anyway: they only connect out (no listen address). They are neighbours too while
+    /// they say they sync with us (`refresh`).
+    outbound: BTreeSet<String>,
     neighbours: BTreeSet<String>,
     /// Every reachable member is a neighbour.
     full_mesh: bool,
@@ -227,11 +231,36 @@ pub(crate) async fn refresh(state: &AppState) -> Option<NeighbourChange> {
             unlisted.insert(peer);
         }
     }
+    // A member that lists no address on our transports, but reached us anyway, only
+    // connects out: it isn't in the graph on our side (it's on none of our transports, as
+    // far as the list goes), while on its side we are, so it would sync with us while we
+    // never sent it our changes (only answers to its). It picks its neighbours; those it
+    // picked keep it, as it says (STATUS).
+    let mut outbound = BTreeSet::new();
+    for member in &current {
+        if member == state.local_id()
+            || unreachable.contains(member)
+            || transports
+                .get(member)
+                .is_some_and(|on| !on.is_disjoint(&local_transports))
+        {
+            continue;
+        }
+        if let Some(peer_state) = state.get_peer_state(member).await {
+            let peer_state = peer_state.lock().await;
+            if matches!(peer_state.standing, crate::state::Standing::Member { .. })
+                && peer_state.syncs_with_us != Some(false)
+            {
+                outbound.insert(member.clone());
+            }
+        }
+    }
     let previous = {
         let view = state.neighbours.read_or_recover();
         if view.as_ref().is_some_and(|view| {
             view.members == current
                 && view.unlisted == unlisted
+                && view.outbound == outbound
                 && view.k == k
                 && view.unreachable == unreachable
                 && view.stable == stable
@@ -256,6 +285,7 @@ pub(crate) async fn refresh(state: &AppState) -> Option<NeighbourChange> {
         &|m| unreachable.contains(m),
         &|m| stable.contains(m),
     );
+    computed.extend(outbound.iter().cloned());
     let listed_neighbours = computed.len();
     computed.extend(unlisted.iter().cloned());
     let change = NeighbourChange {
@@ -277,6 +307,7 @@ pub(crate) async fn refresh(state: &AppState) -> Option<NeighbourChange> {
         k,
         neighbours = computed.len(),
         unlisted = unlisted.len(),
+        outbound = outbound.len(),
         added = change.added.len(),
         removed = change.removed.len(),
         "Neighbours computed from the member list"
@@ -294,6 +325,7 @@ pub(crate) async fn refresh(state: &AppState) -> Option<NeighbourChange> {
         transports,
         local_transports,
         unlisted,
+        outbound,
         neighbours: computed,
     });
     Some(change)