HyParView (slim)

1 files changed, 152 insertions, 9 deletions

diff --git a/src/swarm/topology/mod.rs b/src/swarm/topology/mod.rs
index d9f361a..14e9de6 100644
--- a/src/swarm/topology/mod.rs
+++ b/src/swarm/topology/mod.rs
@@ -33,6 +33,7 @@ use tracing::info;
 use crate::locks::RwLockExt;
 use crate::state::{AppState, PeerState};
 use crate::swarm::members;
+use crate::transport::ParsedMessage;
 
 mod graph;
 
@@ -91,12 +92,149 @@ pub(crate) trait Topology: Send + Sync + 'static {
         let _ = (state, inputs);
     }
 
+    /// A message of the topology's own from `peer` (HyParView's NEIGHBOUR,
+    /// NEIGHBOUR_REPLY, DISCONNECT). By default ignored.
+    async fn received(&self, state: &AppState, peer: &str, message: &ParsedMessage) {
+        let _ = (state, peer, message);
+    }
+
     /// 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
     }
+
+    /// Taking over from another topology, whose neighbours were `neighbours` (the swarm
+    /// switched, `swarm_ways`): a topology with state of its own starts from them. By
+    /// default nothing.
+    fn seed(&self, neighbours: &BTreeSet<String>) {
+        let _ = neighbours;
+    }
+
+    /// The neighbours each agreed to, for a topology that keeps such (HyParView's active
+    /// view), for the tests and the benchmark to look at. By default none.
+    #[cfg(test)]
+    fn active(&self) -> Option<BTreeSet<String>> {
+        None
+    }
+}
+
+/// The topology and the dissemination in use (`AppState::ways`), and which they are.
+pub(crate) struct Ways {
+    pub(crate) kinds: (Kind, crate::swarm::dissemination::Kind),
+    pub(crate) topology: Arc<dyn Topology>,
+    pub(crate) dissemination: Arc<dyn crate::swarm::dissemination::Dissemination>,
+}
+
+impl Ways {
+    pub(crate) fn new(topology: Kind, dissemination: crate::swarm::dissemination::Kind) -> Self {
+        Self {
+            kinds: (topology, dissemination),
+            topology: topology.make(),
+            dissemination: dissemination.make(),
+        }
+    }
+}
+
+/// Where the swarm records its topology and its dissemination (`swarm_ways`); the
+/// genesis node writes its `--topology` and `--dissemination` there
+/// (`swarm::bootstrap`).
+pub(crate) fn topology_path() -> Vec<String> {
+    crate::document::meta::path(&[crate::document::meta::TOPOLOGY])
+}
+
+pub(crate) fn dissemination_path() -> Vec<String> {
+    crate::document::meta::path(&[crate::document::meta::DISSEMINATION])
+}
+
+/// The topology and the dissemination the swarm records: its genesis node writes its
+/// own (`topology_path`, `dissemination_path`), a protocol writer may change them (every
+/// node then swaps to them, `refresh`). Where the document has none (a swarm from
+/// before they were recorded), or one this node doesn't know, this node's own (`own`).
+pub(crate) fn swarm_ways(
+    doc: &crate::document::Document,
+    own: (Kind, crate::swarm::dissemination::Kind),
+) -> (Kind, crate::swarm::dissemination::Kind) {
+    fn recorded<T: serde::de::DeserializeOwned>(
+        doc: &crate::document::Document,
+        path: &[String],
+        what: &str,
+        own: T,
+    ) -> T {
+        match crate::document::store::read_path(doc, path) {
+            Ok(Some(value)) => match serde_json::from_value(value.clone()) {
+                Ok(recorded) => recorded,
+                Err(_) => {
+                    warn_once(&format!(
+                        "The swarm's {what} ({value}) isn't one this node knows; using --{what}"
+                    ));
+                    own
+                }
+            },
+            _ => own,
+        }
+    }
+    (
+        recorded(doc, &topology_path(), "topology", own.0),
+        recorded(doc, &dissemination_path(), "dissemination", own.1),
+    )
+}
+
+/// Swaps to the topology and the dissemination `kinds` if they aren't the ones in use:
+/// fresh, the new topology starting from our neighbours (`Topology::seed`).
+fn switch_ways(state: &AppState, kinds: (Kind, crate::swarm::dissemination::Kind)) {
+    let neighbours: BTreeSet<String> = our_neighbours(state).into_iter().collect();
+    let mut ways = state.ways.write_or_recover();
+    if ways.kinds == kinds {
+        return;
+    }
+    let before = (ways.topology.name(), ways.dissemination.name());
+    if ways.kinds.0 != kinds.0 {
+        ways.topology = kinds.0.make();
+        ways.topology.seed(&neighbours);
+    }
+    if ways.kinds.1 != kinds.1 {
+        ways.dissemination = kinds.1.make();
+    }
+    ways.kinds = kinds;
+    info!(
+        from_topology = before.0,
+        from_dissemination = before.1,
+        topology = ways.topology.name(),
+        dissemination = ways.dissemination.name(),
+        "The swarm switched how neighbours are chosen, or how changes pass on"
+    );
+    if ways
+        .topology
+        .graph(&[], 2, &|_| false, &|_| false)
+        .is_none()
+        && kinds.1 == crate::swarm::dissemination::Kind::WritersTrees
+    {
+        warn_once(
+            "Writers' trees need a graph every node shares, which this topology has none of: changes pass on by announcements alone (a protocol writer may record dissemination \"plumtree\")",
+        );
+    }
+}
+
+/// The topologies to choose from (`--topology`); one per swarm.
+#[derive(Clone, Copy, Debug, Default, PartialEq, Eq, serde::Serialize, serde::Deserialize)]
+#[serde(rename_all = "kebab-case")]
+pub(crate) enum Kind {
+    /// The member list's graph (`RingAndPairs`).
+    #[default]
+    RingAndPairs,
+    /// HyParView's active view, slimmed down (`swarm::hyparview`).
+    Hyparview,
+}
+
+impl Kind {
+    pub(crate) fn make(self) -> Arc<dyn Topology> {
+        match self {
+            Kind::RingAndPairs => Arc::new(RingAndPairs),
+            Kind::Hyparview => Arc::new(crate::swarm::hyparview::HyParView::default()),
+        }
+    }
 }
 
 /// What a node's neighbours are chosen from (`Topology::neighbours`): the members, by
@@ -128,8 +266,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,
+    /// The topology they were computed with, and its version (`Topology::version`).
+    version: (&'static str, 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.
@@ -173,7 +311,7 @@ pub(crate) fn member_list_graph(
                     .map(|(member, _)| member.clone())
                     .collect();
                 let graph = state
-                    .topology
+                    .topology()
                     .graph(&on_it, view.k, &|_| false, &|m| view.stable.contains(m))?;
                 for (member, neighbours) in graph {
                     whole.entry(member).or_default().extend(neighbours);
@@ -362,6 +500,7 @@ pub(crate) fn is_neighbour(state: &AppState, peer: &str) -> bool {
 #[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 topology = state.topology();
     let view = state.neighbours.read_or_recover();
     let view = view.as_ref()?;
     let graph = view
@@ -370,7 +509,7 @@ pub(crate) fn intended_graph(state: &AppState) -> Option<BTreeMap<String, BTreeS
         .map(|member| {
             (
                 member.clone(),
-                view.neighbours_of(&*state.topology, state.local_id(), member),
+                view.neighbours_of(&*topology, state.local_id(), member),
             )
         })
         .collect();
@@ -453,8 +592,8 @@ pub(crate) fn change_peer<T>(
 /// of notifications coalesces into one recomputation.
 pub(crate) async fn run_worker(state: AppState) {
     info!(
-        topology = state.topology.name(),
-        dissemination = state.dissemination.name(),
+        topology = state.topology().name(),
+        dissemination = state.dissemination().name(),
         "Choosing neighbours"
     );
     loop {
@@ -471,6 +610,8 @@ pub(crate) fn affects_neighbours(path: &[String]) -> bool {
     path.is_empty()
         || path.starts_with(&members::members_dir())
         || path.starts_with(&neighbours_path())
+        || path.starts_with(&topology_path())
+        || path.starts_with(&dissemination_path())
 }
 
 /// The most neighbours a swarm may ask each member to keep (`swarm_neighbours`).
@@ -543,6 +684,7 @@ pub(crate) async fn refresh(state: &AppState) -> Option<NeighbourChange> {
         let replica = state.doc.read().await;
         members::apply(state, &replica.doc);
         let k = swarm_neighbours(&replica.doc, state.neighbours_k);
+        switch_ways(state, swarm_ways(&replica.doc, state.configured_ways));
         let current = members::members(&replica.doc);
         // Those that said they are alive within the electorate window.
         let alive: BTreeSet<String> = members::electorate(
@@ -646,8 +788,9 @@ pub(crate) async fn refresh(state: &AppState) -> Option<NeighbourChange> {
         unreachable: &unreachable_of,
         stable: &stable_of,
     };
-    state.topology.maintain(state, &inputs).await;
-    let version = state.topology.version();
+    let topology = state.topology();
+    topology.maintain(state, &inputs).await;
+    let version = (topology.name(), topology.version());
     let previous = {
         let view = state.neighbours.read_or_recover();
         if view.as_ref().is_some_and(|view| {
@@ -694,7 +837,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 = state.topology.neighbours(state.local_id(), &inputs);
+    let mut computed = topology.neighbours(state.local_id(), &inputs);
     computed.extend(outbound.iter().cloned());
     #[cfg(test)]
     let listed_neighbours = computed.len();