announce what trees miss

2 files changed, 115 insertions, 35 deletions

diff --git a/src/swarm/relay.rs b/src/swarm/relay.rs
index f4eaf5a..f79c601 100644
--- a/src/swarm/relay.rs
+++ b/src/swarm/relay.rs
@@ -20,8 +20,13 @@
 //! data they announce (about 15 to 1 at 250 nodes).
 //!
 //! Changes that say whose they are (the writer's own, and what passes on along its tree)
-//! go at once to the receiver's children in the writer's tree (`swarm::tree`), which
-//! the announcement round then leaves out.
+//! go at once to the receiver's children in the writer's tree (`swarm::tree`). Then the
+//! tree carries them, and the round after `ANNOUNCE_DELAY` goes only to the neighbours
+//! the tree leaves out (`Plan::uncovered`: no member they reach comes before them). The
+//! others are told in a slow round (`SLOW_ANNOUNCE_DELAY`): in case their parent missed
+//! the change (it went away, or the nodes' views of the tree differed), coalescing all
+//! that came meanwhile. Changes that don't say whose (pulled, a delta, from a node that
+//! has no part in the writer's tree) are announced to all, as before.
 //!
 //! In a full mesh too: the member list makes every member the writer's neighbour, but
 //! that says nothing about whether the writer reaches them. One behind NAT may reach a
@@ -53,48 +58,87 @@ const PULL_TIMEOUT: Duration = Duration::from_secs(15);
 const ANNOUNCE_DELAY: Duration = Duration::from_millis(200);
 const ANNOUNCE_JITTER: Duration = Duration::from_millis(200);
 
+/// How long the slow round after changes passed on along writers' trees waits: what a
+/// tree misses arrives this much later, and while changes keep coming, every neighbour
+/// is told at most this often.
+const SLOW_ANNOUNCE_DELAY: Duration = Duration::from_secs(2);
+
 /// An announcement round scheduled: for changes from these peers, and the neighbours
 /// that were sent them along a writer's tree meanwhile (`swarm::tree`).
 #[derive(Default)]
 pub(crate) struct Pending {
     senders: HashSet<String>,
     sent: HashSet<String>,
+    /// Some change had no tree to go along: every neighbour is told, not only `targets`.
+    all: bool,
+    /// The neighbours writers' trees leave out (`Plan::uncovered`).
+    targets: HashSet<String>,
 }
 
 /// After changes from `from` were applied and our heads changed: passed on at once along
-/// the tree of their `writer` if the message said whose, and our other neighbours told
-/// in the next announcement round.
+/// the tree of their `writer` if the message said whose; then announced, in the next
+/// round to those the tree leaves out (to all if there's no tree), and to the others in
+/// the slow round.
 pub(crate) async fn announce(state: &AppState, from: &str, writer: Option<u64>) {
     if !state.initial_sync_done.load(Ordering::Acquire) {
         return;
     }
-    let sent = match writer {
+    let plan = match writer {
         Some(writer) => crate::swarm::tree::forward(state, writer, from).await,
-        None => Vec::new(),
+        None => None,
     };
+    let tree = plan.is_some();
     let schedule = {
         let mut pending = state.announce_pending.lock_or_recover();
         let schedule = pending.is_none();
         let pending = pending.get_or_insert_with(Pending::default);
         pending.senders.insert(from.to_owned());
-        pending.sent.extend(sent);
+        match &plan {
+            Some(plan) => {
+                pending.sent.extend(plan.children.iter().cloned());
+                pending.targets.extend(plan.uncovered.iter().cloned());
+            }
+            None => pending.all = true,
+        }
         schedule
     };
-    if !schedule {
+    if schedule {
+        round(state, &state.announce_pending, ANNOUNCE_DELAY);
+    }
+    if !tree {
         return;
     }
+    let schedule = {
+        let mut slow = state.announce_slow.lock_or_recover();
+        let schedule = slow.is_none();
+        let slow = slow.get_or_insert_with(Pending::default);
+        slow.senders.insert(from.to_owned());
+        slow.all = true;
+        if let Some(plan) = &plan {
+            slow.sent.extend(plan.children.iter().cloned());
+        }
+        schedule
+    };
+    if schedule {
+        round(state, &state.announce_slow, SLOW_ANNOUNCE_DELAY);
+    }
+}
+
+/// An announcement round of what `pending` collects, `delay` (and a jitter) from now.
+fn round(
+    state: &AppState,
+    pending: &std::sync::Arc<std::sync::Mutex<Option<Pending>>>,
+    delay: Duration,
+) {
     let state = state.clone();
+    let pending = std::sync::Arc::clone(pending);
     tokio::spawn(async move {
         let jitter = crate::transport::paths::random_nonce()
             % u64::try_from(ANNOUNCE_JITTER.as_millis())
                 .unwrap_or(1)
                 .max(1);
-        tokio::time::sleep(ANNOUNCE_DELAY + Duration::from_millis(jitter)).await;
-        let pending = state
-            .announce_pending
-            .lock_or_recover()
-            .take()
-            .unwrap_or_default();
+        tokio::time::sleep(delay + Duration::from_millis(jitter)).await;
+        let pending = pending.lock_or_recover().take().unwrap_or_default();
         announce_now(&state, &pending).await;
     });
 }
@@ -103,7 +147,10 @@ pub(crate) async fn announce(state: &AppState, from: &str, writer: Option<u64>)
 /// round with a single sender skips it; with several, each lacks the others' changes,
 /// so every neighbour is told.
 async fn announce_now(state: &AppState, pending: &Pending) {
-    let neighbours = unreached(state, &pending.senders, &pending.sent).await;
+    let mut neighbours = unreached(state, &pending.senders, &pending.sent).await;
+    if !pending.all {
+        neighbours.retain(|peer| pending.targets.contains(peer));
+    }
     if neighbours.is_empty() {
         return;
     }
diff --git a/src/swarm/tree.rs b/src/swarm/tree.rs
index d362f74..ee46271 100644
--- a/src/swarm/tree.rs
+++ b/src/swarm/tree.rs
@@ -95,13 +95,30 @@ fn distances(state: &AppState, writer: u64) -> Option<(Arc<Graph>, Arc<HashMap<S
 /// others take them as parents. What a member reaches we know of ourselves and our
 /// neighbours, as they said; of others, the graph is taken as right (if we're wrong,
 /// a change goes twice or by announcement).
+#[cfg(test)]
 pub(crate) async fn children(state: &AppState, writer: u64) -> Vec<String> {
-    let Some((graph, distance)) = distances(state, writer) else {
-        return Vec::new();
-    };
+    plan(state, writer)
+        .await
+        .map(|plan| plan.children)
+        .unwrap_or_default()
+}
+
+/// What `writer`'s tree means for us: our children in it, and our neighbours it leaves
+/// out (`uncovered`).
+pub(crate) struct Plan {
+    pub(crate) children: Vec<String>,
+    /// Our neighbours that no member they reach comes before (the tree reaches them only
+    /// by announcement), and those not in the member list's graph (whom no tree has).
+    pub(crate) uncovered: Vec<String>,
+}
+
+/// Our part in `writer`'s tree (`children`, `Plan`); None while the member list isn't
+/// known, or the writer or we aren't in it.
+pub(crate) async fn plan(state: &AppState, writer: u64) -> Option<Plan> {
+    let (graph, distance) = distances(state, writer)?;
     let me = state.local_id();
     if !distance.contains_key(me) {
-        return Vec::new();
+        return None;
     }
     // Whom we and our neighbours reach, by pair key.
     let mut reaches: HashMap<&str, Vec<u64>> = HashMap::new();
@@ -131,15 +148,14 @@ pub(crate) async fn children(state: &AppState, writer: u64) -> Vec<String> {
             .get(member)
             .map(|d| (*d, !closer(member), topology::pair_key(member)))
     };
-    let Some(my_rank) = rank(me) else {
-        return Vec::new();
-    };
+    let my_rank = rank(me)?;
     let mut children = Vec::new();
+    let mut uncovered = Vec::new();
     for member in graph.get(me).into_iter().flatten() {
         let Some(its_rank) = rank(member) else {
             continue;
         };
-        if its_rank <= my_rank {
+        if distance[member.as_str()] == 0 {
             continue;
         }
         let parent = graph
@@ -152,22 +168,39 @@ pub(crate) async fn children(state: &AppState, writer: u64) -> Vec<String> {
                 let (d, far, _) = rank(candidate).expect("ranked");
                 (d, far, topology::pair_score(candidate, member))
             });
-        if parent.is_some_and(|parent| parent == me) && topology::reaches(state, member).await {
-            children.push(member.clone());
+        match parent {
+            None => uncovered.push(member.clone()),
+            Some(parent)
+                if parent == me && its_rank > my_rank && topology::reaches(state, member).await =>
+            {
+                children.push(member.clone());
+            }
+            Some(_) => {}
+        }
+    }
+    // Neighbours the member list's graph doesn't give us (not listed yet, or only
+    // connecting out): no tree has them.
+    for (peer, _) in state.peers_snapshot().await {
+        if topology::is_neighbour(state, &peer)
+            && !graph.get(me).is_some_and(|ours| ours.contains(&peer))
+        {
+            uncovered.push(peer);
         }
     }
-    children
+    Some(Plan {
+        children,
+        uncovered,
+    })
 }
 
 /// 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 {
+/// tree (but `from`), a sync message each, saying whose. Returns our part in the tree
+/// (`Plan`), None if we have none (announcements then do it all).
+pub(crate) async fn forward(state: &AppState, writer: u64, from: &str) -> Option<Plan> {
+    let mut plan = plan(state, writer).await?;
+    plan.children.retain(|child| child != from);
+    let children = &plan.children;
+    for child in children {
         if let Some(peer_state) = state.get_peer_state(child).await {
             peer_state.lock().await.tree_writer = Some(writer);
         }
@@ -181,5 +214,5 @@ pub(crate) async fn forward(state: &AppState, writer: u64, from: &str) -> Vec<St
             "Passing a change on along the writer's tree"
         );
     }
-    children
+    Some(plan)
 }