topology neighbours and optional graph

2 files changed, 120 insertions, 58 deletions

diff --git a/src/swarm/topology/graph.rs b/src/swarm/topology/graph.rs
index 92240e9..d8b2921 100644
--- a/src/swarm/topology/graph.rs
+++ b/src/swarm/topology/graph.rs
@@ -34,8 +34,8 @@ pub(crate) fn schemes(addresses: &[String]) -> BTreeSet<String> {
 /// 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(
-    topology: &dyn Topology,
+pub(crate) fn neighbours_across<T: Topology + ?Sized>(
+    topology: &T,
     members: &BTreeMap<String, BTreeSet<String>>,
     local: &str,
     local_transports: &BTreeSet<String>,
@@ -56,9 +56,10 @@ pub(crate) fn neighbours_across(
 }
 
 /// The neighbours of `local` among `members` (which may or may not contain `local`),
-/// skipping members for which `unreachable` holds.
-pub(crate) fn neighbours(
-    topology: &dyn Topology,
+/// skipping members for which `unreachable` holds; none by a topology without a shared
+/// graph.
+pub(crate) fn neighbours<T: Topology + ?Sized>(
+    topology: &T,
     members: &[String],
     local: &str,
     k: usize,
@@ -69,7 +70,7 @@ pub(crate) fn neighbours(
     all.push(local.to_owned());
     topology
         .graph(&all, k, &|m: &str| m != local && unreachable(m), stable)
-        .remove(local)
+        .and_then(|mut graph| graph.remove(local))
         .unwrap_or_default()
 }
 
@@ -88,8 +89,8 @@ impl Topology for RingAndPairs {
         k: usize,
         unreachable: &dyn Fn(&str) -> bool,
         stable: &dyn Fn(&str) -> bool,
-    ) -> BTreeMap<String, BTreeSet<String>> {
-        graph(members, k, unreachable, stable)
+    ) -> Option<BTreeMap<String, BTreeSet<String>>> {
+        Some(graph(members, k, unreachable, stable))
     }
 }
 
diff --git a/src/swarm/topology/mod.rs b/src/swarm/topology/mod.rs
index b5da4d8..d9f361a 100644
--- a/src/swarm/topology/mod.rs
+++ b/src/swarm/topology/mod.rs
@@ -41,26 +41,73 @@ pub(crate) use graph::{RingAndPairs, pair_key, pair_score};
 
 /// How members' neighbours are chosen: the graph above (`RingAndPairs`) is one way;
 /// others can be tried against it in the benchmark (`docs/future-work.md`, Swappable
-/// topology and dissemination). One per swarm, as every node must compute the same
-/// graph: the writers' trees are built on it (`member_list_graph`, `swarm::tree`).
+/// topology and dissemination). One per swarm: a node's neighbours must count it as
+/// theirs.
 ///
-/// A topology each node keeps from its own messages instead (HyParView's views) has no
-/// graph every node shares; it takes more than this (its state, its messages, and a
-/// dissemination that needs no shared graph), and comes with one.
+/// A topology either computes a graph every node shares (`graph`: the same inputs give
+/// every node the same), or keeps its own state from its own messages (HyParView's
+/// views), and has none. What needs the shared graph does nothing without one: writers'
+/// trees (`member_list_graph`, `swarm::tree`), the links members record where it is
+/// wrong (`working_graph`, `swarm::working_links`), indirect probes through a member's
+/// neighbours, the dashboard's map of the intended links (`intended_graph`). A
+/// dissemination that needs none goes with such a topology (PlumTree).
+#[async_trait::async_trait]
 pub(crate) trait Topology: Send + Sync + 'static {
     /// Its name (in logs, and once chosen by the swarm, in the document).
     fn name(&self) -> &'static str;
 
     /// Every reachable member's neighbours among `members` (those on one transport),
     /// each keeping about `k`, skipping those `unreachable`, preferring those `stable`
-    /// as it does. Pure: the same inputs give every node the same graph.
+    /// as it does. Pure: the same inputs give every node the same graph. None for a
+    /// topology without a shared graph.
     fn graph(
         &self,
         members: &[String],
         k: usize,
         unreachable: &dyn Fn(&str) -> bool,
         stable: &dyn Fn(&str) -> bool,
-    ) -> BTreeMap<String, BTreeSet<String>>;
+    ) -> Option<BTreeMap<String, BTreeSet<String>>>;
+
+    /// Our neighbours, from what they are chosen from (`Inputs`), each time those may
+    /// have changed (`refresh`). By default our part of the shared graph, on each of our
+    /// transports (`neighbours_across`); a topology with state of its own keeps them
+    /// there.
+    fn neighbours(&self, local: &str, inputs: &Inputs<'_>) -> BTreeSet<String> {
+        neighbours_across(
+            self,
+            inputs.transports,
+            local,
+            inputs.local_transports,
+            inputs.k,
+            inputs.unreachable,
+            inputs.stable,
+        )
+    }
+
+    /// With every recomputation (`refresh`, at least every 5 s), before `neighbours`:
+    /// what a topology with state of its own does about it (asking members to be
+    /// neighbours, giving up on those that don't answer). By default nothing.
+    async fn maintain(&self, state: &AppState, inputs: &Inputs<'_>) {
+        let _ = (state, inputs);
+    }
+
+    /// Changes whenever the topology's own state changed what `neighbours` gives: the
+    /// neighbours are recomputed then, even though nothing they are chosen from did. By
+    /// default always the same.
+    fn version(&self) -> u64 {
+        0
+    }
+}
+
+/// What a node's neighbours are chosen from (`Topology::neighbours`): the members, by
+/// the transports each is on (as the member list gives their addresses), ours, the
+/// neighbours each keeps, whom we judge unreachable, who is stable.
+pub(crate) struct Inputs<'a> {
+    pub(crate) transports: &'a BTreeMap<String, BTreeSet<String>>,
+    pub(crate) local_transports: &'a BTreeSet<String>,
+    pub(crate) k: usize,
+    pub(crate) unreachable: &'a (dyn Fn(&str) -> bool + Sync),
+    pub(crate) stable: &'a (dyn Fn(&str) -> bool + Sync),
 }
 
 /// This node's current neighbours, and the member list they were computed from.
@@ -81,6 +128,8 @@ pub(crate) struct NeighbourView {
     /// they say they sync with us (`refresh`).
     outbound: BTreeSet<String>,
     neighbours: BTreeSet<String>,
+    /// The topology's version they were computed with (`Topology::version`).
+    version: u64,
     /// Every member not judged unreachable is a neighbour (the tests wait for it,
     /// `is_full_mesh`). Not the swarm being a full mesh: members on a transport we are
     /// not on, or that we never reached, don't count against it.
@@ -88,66 +137,71 @@ pub(crate) struct NeighbourView {
     full_mesh: bool,
     /// The graph the member list gives, as if every member reached every other
     /// (`member_list_graph`): computed when first asked for, kept while the member list
-    /// stays the same.
-    whole: std::sync::OnceLock<Arc<BTreeMap<String, BTreeSet<String>>>>,
+    /// stays the same. None for a topology without a shared graph.
+    whole: std::sync::OnceLock<Option<Arc<BTreeMap<String, BTreeSet<String>>>>>,
     /// Where members recorded their links differ from that graph (`swarm::working_links`),
     /// those with a record.
     recorded: BTreeMap<String, members::RecordedLinks>,
     /// That graph with the links members recorded (`working_graph`): computed when first
     /// asked for, kept while the member list and the records stay the same.
-    working: std::sync::OnceLock<Arc<BTreeMap<String, BTreeSet<String>>>>,
+    working: std::sync::OnceLock<Option<Arc<BTreeMap<String, BTreeSet<String>>>>>,
 }
 
 /// Every member's neighbours as the member list gives them, as if every member reached
 /// every other: the same on every node that knows the same member list, whatever each
 /// reaches (`swarm::tree` builds writers' trees on it). On each transport among the
-/// members on it, all together. None until the view is known.
+/// members on it, all together. None until the view is known, and for a topology
+/// without a shared graph (`Topology::graph`).
 pub(crate) fn member_list_graph(
     state: &AppState,
 ) -> Option<Arc<BTreeMap<String, BTreeSet<String>>>> {
     let view = state.neighbours.read_or_recover();
     let view = view.as_ref()?;
-    Some(Arc::clone(view.whole.get_or_init(|| {
-        let mut whole: BTreeMap<String, BTreeSet<String>> = view
-            .members
-            .iter()
-            .map(|member| (member.clone(), BTreeSet::new()))
-            .collect();
-        let transports: BTreeSet<&String> = view.transports.values().flatten().collect();
-        for transport in transports {
-            let on_it: Vec<String> = view
-                .transports
+    view.whole
+        .get_or_init(|| {
+            let mut whole: BTreeMap<String, BTreeSet<String>> = view
+                .members
                 .iter()
-                .filter(|(_, on)| on.contains(transport))
-                .map(|(member, _)| member.clone())
+                .map(|member| (member.clone(), BTreeSet::new()))
                 .collect();
-            let graph = state
-                .topology
-                .graph(&on_it, view.k, &|_| false, &|m| view.stable.contains(m));
-            for (member, neighbours) in graph {
-                whole.entry(member).or_default().extend(neighbours);
+            let transports: BTreeSet<&String> = view.transports.values().flatten().collect();
+            for transport in transports {
+                let on_it: Vec<String> = view
+                    .transports
+                    .iter()
+                    .filter(|(_, on)| on.contains(transport))
+                    .map(|(member, _)| member.clone())
+                    .collect();
+                let graph = state
+                    .topology
+                    .graph(&on_it, view.k, &|_| false, &|m| view.stable.contains(m))?;
+                for (member, neighbours) in graph {
+                    whole.entry(member).or_default().extend(neighbours);
+                }
             }
-        }
-        Arc::new(whole)
-    })))
+            Some(Arc::new(whole))
+        })
+        .clone()
 }
 
 /// The member list's graph (`member_list_graph`) with the links members recorded
 /// (`swarm::working_links`): without those either end recorded as unreached, with those
 /// one recorded as reached (`with_recorded`). The same on every node that has the same
 /// document; what the writers' trees are built on (`swarm::tree`). None until the view
-/// is known.
+/// is known, and for a topology without a shared graph.
 pub(crate) fn working_graph(state: &AppState) -> Option<Arc<BTreeMap<String, BTreeSet<String>>>> {
     let whole = member_list_graph(state)?;
     let view = state.neighbours.read_or_recover();
     let view = view.as_ref()?;
-    Some(Arc::clone(view.working.get_or_init(|| {
-        if view.recorded.is_empty() {
-            whole
-        } else {
-            Arc::new(with_recorded(&whole, &view.recorded))
-        }
-    })))
+    view.working
+        .get_or_init(|| {
+            Some(if view.recorded.is_empty() {
+                whole
+            } else {
+                Arc::new(with_recorded(&whole, &view.recorded))
+            })
+        })
+        .clone()
 }
 
 /// `graph` with the links members recorded (`recorded`): those one recorded as reached
@@ -303,9 +357,11 @@ pub(crate) fn is_neighbour(state: &AppState, peer: &str) -> bool {
 /// they are (with members that only connect out, or aren't listed yet). For a map of
 /// the swarm (the dashboard); None until the view is known. A link exists if either end
 /// chose it: draw one if either lists the other. About `members²` work: compute it when
-/// the member list may have changed, not on every look.
+/// the member list may have changed, not on every look. None too for a topology without
+/// a shared graph.
 #[cfg_attr(not(feature = "dashboard"), allow(dead_code))]
 pub(crate) fn intended_graph(state: &AppState) -> Option<BTreeMap<String, BTreeSet<String>>> {
+    member_list_graph(state)?;
     let view = state.neighbours.read_or_recover();
     let view = view.as_ref()?;
     let graph = view
@@ -581,10 +637,22 @@ pub(crate) async fn refresh(state: &AppState) -> Option<NeighbourChange> {
             }
         }
     }
+    let unreachable_of = |m: &str| unreachable.contains(m);
+    let stable_of = |m: &str| stable.contains(m);
+    let inputs = Inputs {
+        transports: &transports,
+        local_transports: &local_transports,
+        k,
+        unreachable: &unreachable_of,
+        stable: &stable_of,
+    };
+    state.topology.maintain(state, &inputs).await;
+    let version = state.topology.version();
     let previous = {
         let view = state.neighbours.read_or_recover();
         if view.as_ref().is_some_and(|view| {
-            view.members == current
+            view.version == version
+                && view.members == current
                 && view.unlisted == unlisted
                 && view.outbound == outbound
                 && view.k == k
@@ -626,15 +694,7 @@ pub(crate) async fn refresh(state: &AppState) -> Option<NeighbourChange> {
         Some(previous) => previous,
         None => state.peer_states.read().await.keys().cloned().collect(),
     };
-    let mut computed = neighbours_across(
-        &*state.topology,
-        &transports,
-        state.local_id(),
-        &local_transports,
-        k,
-        &|m| unreachable.contains(m),
-        &|m| stable.contains(m),
-    );
+    let mut computed = state.topology.neighbours(state.local_id(), &inputs);
     computed.extend(outbound.iter().cloned());
     #[cfg(test)]
     let listed_neighbours = computed.len();
@@ -680,6 +740,7 @@ pub(crate) async fn refresh(state: &AppState) -> Option<NeighbourChange> {
         unlisted,
         outbound,
         neighbours: computed,
+        version,
         whole,
         recorded,
         working,