add writer trees

1 files changed, 159 insertions, 0 deletions

diff --git a/src/swarm/tree.rs b/src/swarm/tree.rs
new file mode 100644
index 0000000..489bc22
--- /dev/null
+++ b/src/swarm/tree.rs
@@ -0,0 +1,159 @@
+//! Passing a writer's changes on along its tree: every member gets them once, over a
+//! link that works, and as soon as the member before it has them.
+//!
+//! The tree is the same on every node, computed from what they share: the graph the
+//! member list gives (`topology::member_list_graph`: every member's neighbours, as if
+//! all reached all) and what each member tells its neighbours it reaches
+//! (`swarm::reach`). Every member's distance from the writer over that graph orders the
+//! members, by distance and then pair key; a member's parent is the neighbour it reaches
+//! that comes before it, the closest to the writer first, then by pair score
+//! (`topology::pair_score`). Coming before rules out cycles, and taking only a neighbour
+//! it reaches routes around links that don't work: one behind NAT that reaches a single
+//! member takes that one, at whatever distance.
+//!
+//! A sync message that passes a writer's change on says whose (`SyncEnvelope::writer`).
+//! Whoever applies one passes it on to its own children in that writer's tree
+//! (`forward`), however it got it; so does the writer, to all its neighbours, with its
+//! own writes. A member no member it reaches comes before (or whose parent didn't get
+//! the change) gets it from the announcements (`swarm::relay`), as before.
+//!
+//! The graph is computed once per member list (tens of milliseconds for thousands of
+//! members), the distances once per writer and graph (`MAX_WRITERS` kept).
+
+use std::collections::{BTreeMap, BTreeSet, HashMap, VecDeque};
+use std::sync::Arc;
+
+use tracing::trace;
+
+use crate::locks::MutexExt;
+use crate::state::AppState;
+use crate::swarm::{reach, topology};
+
+/// The writers whose distances are kept at most.
+const MAX_WRITERS: usize = 256;
+
+type Graph = BTreeMap<String, BTreeSet<String>>;
+
+/// The distances from each writer, over the graph they were computed for.
+#[derive(Default)]
+pub(crate) struct Trees {
+    graph: Option<Arc<Graph>>,
+    /// The members by pair key, of that graph.
+    by_key: HashMap<u64, String>,
+    distances: HashMap<u64, Arc<HashMap<String, u32>>>,
+}
+
+/// The graph, and every member's distance from `writer` (by pair key) over it; None
+/// while the member list isn't known, or `writer` isn't in it.
+fn distances(state: &AppState, writer: u64) -> Option<(Arc<Graph>, Arc<HashMap<String, u32>>)> {
+    let graph = topology::member_list_graph(state)?;
+    let mut trees = state.trees.lock_or_recover();
+    if !trees
+        .graph
+        .as_ref()
+        .is_some_and(|known| Arc::ptr_eq(known, &graph))
+    {
+        trees.by_key = graph
+            .keys()
+            .map(|member| (topology::pair_key(member), member.clone()))
+            .collect();
+        trees.distances.clear();
+        trees.graph = Some(Arc::clone(&graph));
+    }
+    if let Some(known) = trees.distances.get(&writer) {
+        return Some((graph, Arc::clone(known)));
+    }
+    let from = trees.by_key.get(&writer)?.clone();
+    let mut distance = HashMap::from([(from.clone(), 0u32)]);
+    let mut queue = VecDeque::from([from]);
+    while let Some(member) = queue.pop_front() {
+        let next = distance[&member] + 1;
+        for neighbour in graph.get(&member).into_iter().flatten() {
+            if !distance.contains_key(neighbour) {
+                distance.insert(neighbour.clone(), next);
+                queue.push_back(neighbour.clone());
+            }
+        }
+    }
+    if trees.distances.len() >= MAX_WRITERS {
+        trees.distances.clear();
+    }
+    let distance = Arc::new(distance);
+    trees.distances.insert(writer, Arc::clone(&distance));
+    Some((graph, distance))
+}
+
+/// Our children in `writer`'s tree: our neighbours in the graph whose parent we are,
+/// and whom we reach.
+pub(crate) async fn children(state: &AppState, writer: u64) -> Vec<String> {
+    let Some((graph, distance)) = distances(state, writer) else {
+        return Vec::new();
+    };
+    let rank = |member: &str| {
+        distance
+            .get(member)
+            .map(|d| (*d, topology::pair_key(member)))
+    };
+    let me = state.local_id();
+    let Some(my_rank) = rank(me) else {
+        return Vec::new();
+    };
+    let mut children = Vec::new();
+    for member in graph.get(me).into_iter().flatten() {
+        let Some(its_rank) = rank(member) else {
+            continue;
+        };
+        if its_rank <= my_rank {
+            continue;
+        }
+        // Whom it reaches, as it told us; not knowing, as the graph says.
+        let reaches = match state.get_peer_state(member).await {
+            Some(peer_state) => reach::of(state, peer_state.lock().await.reaches.as_ref()),
+            None => None,
+        };
+        let parent = graph
+            .get(member)
+            .into_iter()
+            .flatten()
+            .filter(|candidate| rank(candidate).is_some_and(|r| r < its_rank))
+            .filter(|candidate| {
+                reaches
+                    .as_ref()
+                    .is_none_or(|keys| keys.contains(&topology::pair_key(candidate)))
+            })
+            .min_by_key(|candidate| {
+                (
+                    distance[*candidate],
+                    topology::pair_score(candidate, member),
+                )
+            });
+        if parent.is_some_and(|parent| parent == me) && topology::reaches(state, member).await {
+            children.push(member.clone());
+        }
+    }
+    children
+}
+
+/// We applied a change of `writer`'s that `from` passed on: to our children in its
+/// tree (but `from`), a sync message each, saying whose. Returns them.
+pub(crate) async fn forward(state: &AppState, writer: u64, from: &str) -> Vec<String> {
+    let children: Vec<String> = children(state, writer)
+        .await
+        .into_iter()
+        .filter(|child| child != from)
+        .collect();
+    for child in &children {
+        if let Some(peer_state) = state.get_peer_state(child).await {
+            peer_state.lock().await.tree_writer = Some(writer);
+        }
+        state.request_sync(child);
+    }
+    if !children.is_empty() {
+        trace!(
+            writer,
+            children = children.len(),
+            "Passing a change on along the writer's tree"
+        );
+    }
+    children
+}