writers from authors

1 files changed, 20 insertions, 6 deletions

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) => {