add writer trees

1 files changed, 36 insertions, 20 deletions

diff --git a/src/swarm/relay.rs b/src/swarm/relay.rs
index 69d73d3..7dbc5b0 100644
--- a/src/swarm/relay.rs
+++ b/src/swarm/relay.rs
@@ -59,25 +59,33 @@ const PULL_TIMEOUT: Duration = Duration::from_secs(15);
 const ANNOUNCE_DELAY: Duration = Duration::from_millis(200);
 const ANNOUNCE_JITTER: Duration = Duration::from_millis(200);
 
-/// After changes from `from` were applied and our heads changed: tell our other
-/// neighbours, in the next announcement round.
-pub(crate) async fn announce(state: &AppState, from: &str) {
+/// 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>,
+}
+
+/// 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.
+pub(crate) async fn announce(state: &AppState, from: &str, writer: Option<u64>) {
     if !state.initial_sync_done.load(Ordering::Acquire) {
         return;
     }
     forward_eagerly(state, from).await;
+    let sent = match writer {
+        Some(writer) => crate::swarm::tree::forward(state, writer, from).await,
+        None => Vec::new(),
+    };
     let schedule = {
         let mut pending = state.announce_pending.lock_or_recover();
-        match pending.as_mut() {
-            Some(senders) => {
-                senders.insert(from.to_owned());
-                false
-            }
-            None => {
-                *pending = Some(HashSet::from([from.to_owned()]));
-                true
-            }
-        }
+        let schedule = pending.is_none();
+        let pending = pending.get_or_insert_with(Pending::default);
+        pending.senders.insert(from.to_owned());
+        pending.sent.extend(sent);
+        schedule
     };
     if !schedule {
         return;
@@ -89,12 +97,12 @@ pub(crate) async fn announce(state: &AppState, from: &str) {
                 .unwrap_or(1)
                 .max(1);
         tokio::time::sleep(ANNOUNCE_DELAY + Duration::from_millis(jitter)).await;
-        let senders = state
+        let pending = state
             .announce_pending
             .lock_or_recover()
             .take()
             .unwrap_or_default();
-        announce_now(&state, &senders).await;
+        announce_now(&state, &pending).await;
     });
 }
 
@@ -115,8 +123,8 @@ async fn forward_eagerly(state: &AppState, from: &str) {
 /// One announcement round. The peer whose changes we announce has them already, so a
 /// 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, senders: &HashSet<String>) {
-    let neighbours = unreached(state, senders).await;
+async fn announce_now(state: &AppState, pending: &Pending) {
+    let neighbours = unreached(state, &pending.senders, &pending.sent).await;
     if neighbours.is_empty() {
         return;
     }
@@ -132,8 +140,13 @@ async fn announce_now(state: &AppState, senders: &HashSet<String>) {
 /// announce has them already, so with a single sender it is skipped; with several, each
 /// lacks the others' changes. So is a neighbour whose latest heads (its STATUS, an
 /// announcement of its own, or its STATUS_REPLY) are ours now: it has all we have; and
-/// one we forward to eagerly, which was sent the changes already.
-async fn unreached(state: &AppState, senders: &HashSet<String>) -> Vec<String> {
+/// one we forward to eagerly, or `sent` the changes along a writer's tree, which has
+/// them already.
+async fn unreached(
+    state: &AppState,
+    senders: &HashSet<String>,
+    sent: &HashSet<String>,
+) -> Vec<String> {
     let skip = match senders.len() {
         1 => senders.iter().next().cloned(),
         _ => None,
@@ -144,7 +157,10 @@ async fn unreached(state: &AppState, senders: &HashSet<String>) -> Vec<String> {
     };
     let mut unreached = Vec::new();
     for (peer, peer_state) in state.peers_snapshot().await {
-        if skip.as_deref() == Some(peer.as_str()) || !topology::is_neighbour(state, &peer) {
+        if skip.as_deref() == Some(peer.as_str())
+            || sent.contains(&peer)
+            || !topology::is_neighbour(state, &peer)
+        {
             continue;
         }
         let peer_state = peer_state.lock().await;