announce what trees miss

5 files changed, 125 insertions, 36 deletions

diff --git a/docs/peers.md b/docs/peers.md
index be77d5a..59b45d0 100644
--- a/docs/peers.md
+++ b/docs/peers.md
@@ -316,7 +316,12 @@ cycles; taking only one it reaches routes around links that don't work: where a
 behind NAT reaches a single member, that one comes first among those as far from it,
 and the others take it as their parent. A sync message that passes a
 writer's change on says whose; whoever applies one sends it at once to its children in
-that writer's tree, however it got it, and leaves them out of its announcement round.
+that writer's tree, however it got it. Its announcement round then goes only to the
+neighbours the tree leaves out (no member they reach comes before them, or the member
+list's graph doesn't have them); all others are told in a slow round 2 s later, which
+catches what a tree missed (a parent that went away, nodes whose views of the tree
+differed) and covers everything that came meanwhile: while changes keep coming, each
+neighbour gets an announcement about every 2 s, not with every change.
 A member without a parent it reaches, or whose parent missed the change, gets it by
 announcement and pull as before, then passes it on to its own children. With every link
 working, each member gets each change once (`bench_writer_trees`: 1999 messages per
diff --git a/src/node.rs b/src/node.rs
index b923185..b376baa 100644
--- a/src/node.rs
+++ b/src/node.rs
@@ -141,6 +141,7 @@ impl Node {
             peers_due: Default::default(),
             pulling: Default::default(),
             announce_pending: Default::default(),
+            announce_slow: Default::default(),
             reach: Default::default(),
             trees: Default::default(),
             rtt: Default::default(),
diff --git a/src/state/mod.rs b/src/state/mod.rs
index 57dab7d..53d8178 100644
--- a/src/state/mod.rs
+++ b/src/state/mod.rs
@@ -151,6 +151,9 @@ pub(crate) struct AppState {
     pub(crate) pulling: Arc<std::sync::Mutex<crate::swarm::relay::Pulling>>,
     /// An announcement round is scheduled (`swarm::relay`).
     pub(crate) announce_pending: Arc<std::sync::Mutex<Option<crate::swarm::relay::Pending>>>,
+    /// The slow round after changes passed on along writers' trees is scheduled
+    /// (`swarm::relay`).
+    pub(crate) announce_slow: Arc<std::sync::Mutex<Option<crate::swarm::relay::Pending>>>,
     /// Distances in writers' trees (`swarm::tree`).
     pub(crate) trees: Arc<std::sync::Mutex<crate::swarm::tree::Trees>>,
     /// Whom we reach, as told our neighbours (`swarm::reach`).
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)
 }