announce what trees miss

1 files changed, 62 insertions, 15 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;
     }