build topology per transport

1 files changed, 128 insertions, 4 deletions

diff --git a/src/swarm/topology.rs b/src/swarm/topology.rs
index 5150e38..2c6b867 100644
--- a/src/swarm/topology.rs
+++ b/src/swarm/topology.rs
@@ -16,6 +16,14 @@
 //! `k` neighbours. With at most `k + 1` reachable members the graph is the full mesh:
 //! small swarms behave exactly as before. A member joining, leaving or going offline
 //! changes only links near it.
+//!
+//! Members on several transports (Mycelium, libp2p) get such a graph per transport,
+//! among the members on it (as their addresses in the member list say), and a node's
+//! neighbours are the union of its neighbours on each transport it is on
+//! (`neighbours_across`). Each of those graphs is connected (its ring), and a member on
+//! two transports is in both: as long as members on several transports link the
+//! transports, the swarm stays connected, and such a member has about `k` neighbours per
+//! transport. With one transport, it is the one graph as before.
 
 use std::collections::{BTreeMap, BTreeSet};
 
@@ -33,6 +41,9 @@ pub(crate) struct NeighbourView {
     k: usize,
     unreachable: BTreeSet<String>,
     stable: BTreeSet<String>,
+    /// The transports (address schemes) of each member, and ours.
+    transports: BTreeMap<String, BTreeSet<String>>,
+    local_transports: BTreeSet<String>,
     neighbours: BTreeSet<String>,
     /// Every reachable member is a neighbour.
     full_mesh: bool,
@@ -132,7 +143,7 @@ pub(crate) async fn refresh(state: &AppState) -> Option<NeighbourChange> {
     if !state.is_ready().await {
         return None;
     }
-    let (current, stable) = {
+    let (current, stable, transports) = {
         let replica = state.doc.read().await;
         members::learn_addresses(state, &replica.doc);
         let current = members::members(&replica.doc);
@@ -141,8 +152,13 @@ pub(crate) async fn refresh(state: &AppState) -> Option<NeighbourChange> {
             .filter(|m| members::tier(&replica.doc, m) == members::Tier::Stable)
             .cloned()
             .collect();
-        (current, stable)
+        let transports: BTreeMap<String, BTreeSet<String>> = current
+            .iter()
+            .map(|m| (m.clone(), schemes(&members::addresses(&replica.doc, m))))
+            .collect();
+        (current, stable, transports)
     };
+    let local_transports: BTreeSet<String> = state.transport.transports().into_iter().collect();
     let k = state.neighbours_k;
     // Those that don't answer, and those on none of our transports.
     let mut unreachable = BTreeSet::new();
@@ -165,6 +181,8 @@ pub(crate) async fn refresh(state: &AppState) -> Option<NeighbourChange> {
                 && view.k == k
                 && view.unreachable == unreachable
                 && view.stable == stable
+                && view.transports == transports
+                && view.local_transports == local_transports
         }) {
             return None;
         }
@@ -176,9 +194,10 @@ pub(crate) async fn refresh(state: &AppState) -> Option<NeighbourChange> {
         Some(previous) => previous,
         None => state.peer_states.read().await.keys().cloned().collect(),
     };
-    let computed = neighbours(
-        &current,
+    let computed = neighbours_across(
+        &transports,
         state.local_id(),
+        &local_transports,
         k,
         &|m| unreachable.contains(m),
         &|m| stable.contains(m),
@@ -215,6 +234,8 @@ pub(crate) async fn refresh(state: &AppState) -> Option<NeighbourChange> {
         k,
         unreachable,
         stable,
+        transports,
+        local_transports,
         neighbours: computed,
     });
     Some(change)
@@ -230,6 +251,43 @@ const PAIR_LABEL: &[u8] = b"continuum/topology-pair/2";
 /// graph stays connected and symmetric.
 const STABLE_FANIN_FACTOR: usize = 4;
 
+/// The transports `addresses` are on: their schemes (`mycelium` of `mycelium:<key>`);
+/// the empty one for an address without (a transport whose identities are addresses).
+pub(crate) fn schemes(addresses: &[String]) -> BTreeSet<String> {
+    addresses
+        .iter()
+        .map(|address| {
+            address
+                .split_once(':')
+                .map_or("", |(scheme, _)| scheme)
+                .to_owned()
+        })
+        .collect()
+}
+
+/// The neighbours of `local` among `members` (by the transports each is on), on each of
+/// `local_transports`: on each, its neighbours among the members on it
+/// (`neighbours`), all together.
+pub(crate) fn neighbours_across(
+    members: &BTreeMap<String, BTreeSet<String>>,
+    local: &str,
+    local_transports: &BTreeSet<String>,
+    k: usize,
+    unreachable: &dyn Fn(&str) -> bool,
+    stable: &dyn Fn(&str) -> bool,
+) -> BTreeSet<String> {
+    let mut all = BTreeSet::new();
+    for transport in local_transports {
+        let on_it: Vec<String> = members
+            .iter()
+            .filter(|(member, transports)| *member != local && transports.contains(transport))
+            .map(|(member, _)| member.clone())
+            .collect();
+        all.extend(neighbours(&on_it, local, k, unreachable, stable));
+    }
+    all
+}
+
 /// The neighbours of `local` among `members` (which may or may not contain `local`),
 /// skipping members for which `unreachable` holds.
 pub(crate) fn neighbours(
@@ -642,6 +700,72 @@ mod tests {
         );
     }
 
+    /// Two transports with members on one or the other, and two members on both: every
+    /// link is over a shared transport, and the two members on both connect the rest.
+    #[test]
+    fn members_on_two_transports_bridge_the_others() {
+        let k = 8;
+        let ms = members(201);
+        let on = |names: &[&str]| -> BTreeSet<String> {
+            names.iter().map(|name| (*name).to_owned()).collect()
+        };
+        let transports: BTreeMap<String, BTreeSet<String>> = ms
+            .iter()
+            .enumerate()
+            .map(|(i, m)| {
+                let transports = match i {
+                    0 | 1 => on(&["a", "b"]),
+                    i if i % 2 == 0 => on(&["a"]),
+                    _ => on(&["b"]),
+                };
+                (m.clone(), transports)
+            })
+            .collect();
+        let no_stable = |_: &str| false;
+        let mut graph: BTreeMap<String, BTreeSet<String>> = BTreeMap::new();
+        for (m, own) in &transports {
+            for n in neighbours_across(&transports, m, own, k, &all_reachable, &no_stable) {
+                assert!(
+                    !transports[&n].is_disjoint(own),
+                    "{m} and {n} share no transport"
+                );
+                graph.entry(m.clone()).or_default().insert(n.clone());
+                graph.entry(n).or_default().insert(m.clone());
+            }
+        }
+        assert_eq!(graph.len(), ms.len());
+        assert!(connected(&graph));
+        let bridge = &ms[0];
+        assert!(graph[bridge].iter().any(|n| !transports[n].contains("b")));
+        assert!(graph[bridge].iter().any(|n| !transports[n].contains("a")));
+
+        // One transport: the one graph, as without transports.
+        let one: BTreeMap<String, BTreeSet<String>> =
+            ms.iter().map(|m| (m.clone(), on(&[""]))).collect();
+        for m in &ms[..5] {
+            assert_eq!(
+                neighbours_across(&one, m, &on(&[""]), k, &all_reachable, &no_stable),
+                neighbours(&ms, m, k, &all_reachable, &no_stable)
+            );
+        }
+    }
+
+    #[test]
+    fn schemes_of_addresses() {
+        let addresses =
+            |all: &[&str]| -> Vec<String> { all.iter().map(|a| (*a).to_owned()).collect() };
+        assert_eq!(
+            schemes(&addresses(&[
+                "mycelium:ab",
+                "libp2p:/ip4/1.2.3.4/tcp/1",
+                "libp2p:/ip6/::1/tcp/1"
+            ])),
+            BTreeSet::from(["libp2p".to_owned(), "mycelium".to_owned()])
+        );
+        assert_eq!(schemes(&addresses(&["a"])), BTreeSet::from([String::new()]));
+        assert!(schemes(&[]).is_empty());
+    }
+
     #[test]
     fn hub_degree_is_bounded_however_few_stable_members() {
         for (n, stable_count) in [(250, 10), (400, 5)] {