writers from authors

5 files changed, 75 insertions, 32 deletions

diff --git a/docs/peers.md b/docs/peers.md
index 377d8aa..127b2ad 100644
--- a/docs/peers.md
+++ b/docs/peers.md
@@ -316,11 +316,13 @@ member's parent is the neighbour it reaches that comes before it, the closest to
 writer first, then one that reaches closer, then by pair score. Coming before rules out
 cycles; taking only one it reaches routes around links that don't work: where a writer
 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. So does one that had the change already: it may
-have come another way first without saying whose (with an answer to one of its own
-messages, say), and then nothing was passed on. Its announcement round then goes only to the
+and the others take it as their parent. Whose a change is, its author says (every
+node writes as itself, its node ID): whoever applies changes sends them at once to its
+children in each of their writers' trees, however they came (along a tree, with an
+answer to one of its own messages, pulled). A sync message along a tree also says
+whose; one that had those changes already passes them on too (they may have come
+another way first, when it wasn't yet anyone's turn to pass them on), so a child that
+lacks them still gets them. 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
diff --git a/src/document/changes.rs b/src/document/changes.rs
index 92b7744..faba895 100644
--- a/src/document/changes.rs
+++ b/src/document/changes.rs
@@ -27,6 +27,28 @@ pub(crate) fn commit_times(doc: &Document, since: &[ChangeHash]) -> Vec<i64> {
         .collect()
 }
 
+/// Who wrote the changes after `since`: the authors their writers set (a node's ID,
+/// `replica::authored`), each once; and whether any change has none (a snapshot, or a
+/// change of a node that set none).
+pub(crate) fn authors_since(doc: &Document, since: &[ChangeHash]) -> (Vec<String>, bool) {
+    let mut authors: Vec<String> = Vec::new();
+    let mut unknown = false;
+    for change in doc.get_changes_meta(since) {
+        match change
+            .author
+            .as_ref()
+            .and_then(|author| std::str::from_utf8(author.as_bytes()).ok())
+        {
+            Some(author) if !authors.iter().any(|known| known == author) => {
+                authors.push(author.to_owned());
+            }
+            Some(_) => {}
+            None => unknown = true,
+        }
+    }
+    (authors, unknown)
+}
+
 /// The change `hash` as the document encodes it; None if `doc` doesn't have it.
 pub(crate) fn change_bytes(doc: &Document, hash: &ChangeHash) -> Option<Vec<u8>> {
     doc.get_change_by_hash(hash)
diff --git a/src/swarm/relay.rs b/src/swarm/relay.rs
index a8560a5..53c1065 100644
--- a/src/swarm/relay.rs
+++ b/src/swarm/relay.rs
@@ -77,30 +77,33 @@ pub(crate) struct Pending {
 }
 
 /// 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; 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>) {
+/// the trees of their `writers` (by their authors' pair keys); then announced, in the
+/// next round to those the trees leave out, and to the others in the slow round. To all
+/// in the next round if any change has no tree to go along: none known whose
+/// (`unauthored`), a writer not in the member list, or we have no part in its tree.
+pub(crate) async fn announce(state: &AppState, from: &str, writers: &[u64], unauthored: bool) {
     if !state.initial_sync_done.load(Ordering::Acquire) {
         return;
     }
-    let plan = match writer {
-        Some(writer) => crate::swarm::tree::forward(state, writer, from).await,
-        None => None,
-    };
-    let tree = plan.is_some();
+    let mut plans = Vec::new();
+    let mut treeless = unauthored || writers.is_empty();
+    for writer in writers {
+        match crate::swarm::tree::forward(state, *writer, from).await {
+            Some(plan) => plans.push(plan),
+            None => treeless = true,
+        }
+    }
+    let tree = !plans.is_empty();
     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());
-        match &plan {
-            Some(plan) => {
-                pending.sent.extend(plan.children.iter().cloned());
-                pending.targets.extend(plan.uncovered.iter().cloned());
-            }
-            None => pending.all = true,
+        for plan in &plans {
+            pending.sent.extend(plan.children.iter().cloned());
+            pending.targets.extend(plan.uncovered.iter().cloned());
         }
+        pending.all |= treeless;
         schedule
     };
     if schedule {
@@ -115,7 +118,7 @@ pub(crate) async fn announce(state: &AppState, from: &str, writer: Option<u64>)
         let slow = slow.get_or_insert_with(Pending::default);
         slow.senders.insert(from.to_owned());
         slow.all = true;
-        if let Some(plan) = &plan {
+        for plan in &plans {
             slow.sent.extend(plan.children.iter().cloned());
         }
         schedule
diff --git a/src/swarm/sync.rs b/src/swarm/sync.rs
index dd5444b..b20f7ef 100644
--- a/src/swarm/sync.rs
+++ b/src/swarm/sync.rs
@@ -376,9 +376,11 @@ pub(crate) async fn handle_sync(
         let clock = Arc::clone(&state.clock);
         let ahead_ms = state.clock_ahead_ms.load(Ordering::Relaxed);
 
-        // Some((caught_up, heads changed, carried changes)) when applied, None when the
-        // message belongs to another epoch.
-        tokio::task::spawn_blocking(move || -> Result<Option<(bool, bool, bool)>> {
+        // Some((caught_up, heads changed, carried changes, the new changes' authors and
+        // whether any has none)) when applied, None when the message belongs to another
+        // epoch.
+        type Applied = (bool, bool, bool, (Vec<String>, bool));
+        tokio::task::spawn_blocking(move || -> Result<Option<Applied>> {
             // Before the lock: decoding a large message takes a while.
             let message = crate::document::sync::decode(&message)?;
             let carried = message.carries_changes();
@@ -411,6 +413,13 @@ pub(crate) async fn handle_sync(
             let heads_after = replica.heads().to_vec();
             stats.heads_seen(heads_after.len());
             let changed = heads_after != heads_before;
+            // Whose they are, to pass them on along their writers' trees, however they
+            // came (along a tree, with an answer, pulled).
+            let authors = if changed {
+                crate::document::changes::authors_since(&replica.doc, &heads_before)
+            } else {
+                (Vec::new(), false)
+            };
             if changed && measure_propagation {
                 // Now by the peers' clocks (their median): this node's own clock skew
                 // doesn't count; the writer's still does.
@@ -432,13 +441,14 @@ pub(crate) async fn handle_sync(
                 check_caught_up && from_ready && peer_state.sync.have_all_theirs(&replica.doc),
                 changed,
                 carried,
+                authors,
             )))
         })
         .await
         .context("Sync task panicked")??
     };
 
-    let Some((caught_up, changed, carried)) = applied else {
+    let Some((caught_up, changed, carried, (authors, unauthored))) = applied else {
         debug!(peer = %peer, epoch = %epoch_label, "Dropped sync message from another epoch");
         return Ok(());
     };
@@ -460,7 +470,11 @@ pub(crate) async fn handle_sync(
     crate::swarm::relay::pulled(state, peer);
     // Caught up: our other neighbours may lack what the source had.
     if changed || just_caught_up {
-        crate::swarm::relay::announce(state, peer, writer.filter(|_| changed)).await;
+        let writers: Vec<u64> = authors
+            .iter()
+            .map(|author| crate::swarm::topology::pair_key(author))
+            .collect();
+        crate::swarm::relay::announce(state, peer, &writers, unauthored).await;
     } else if carried
         && let Some(writer) = writer
         && state.initial_sync_done.load(Ordering::Acquire)
@@ -496,7 +510,7 @@ fn fetch_delta(state: &AppState, peer: &str, epoch: transport::Epoch, spec: tran
             Ok(changed) => {
                 state.notify_store_changed();
                 if changed {
-                    crate::swarm::relay::announce(&state, &peer, None).await;
+                    crate::swarm::relay::announce(&state, &peer, &[], true).await;
                 }
             }
             Err(error) => {
diff --git a/src/swarm/tree.rs b/src/swarm/tree.rs
index 8031a69..8093c91 100644
--- a/src/swarm/tree.rs
+++ b/src/swarm/tree.rs
@@ -12,11 +12,13 @@
 //! it reaches routes around links that don't work: where a writer behind NAT reaches a
 //! single member, that one comes first and the others take it.
 //!
-//! 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.
+//! Whose a change is, its author says (every node writes as itself: its node ID, by
+//! pair key here): whoever applies changes passes them on to its own children in each of
+//! their writers' trees (`forward`), however they came; so does the writer, to all its
+//! neighbours, with its own writes. A sync message along a tree also says whose
+//! (`SyncEnvelope::writer`), for one that had them already (`swarm::sync`). 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).