tree changes had already

2 files changed, 21 insertions, 4 deletions

diff --git a/src/document/sync.rs b/src/document/sync.rs
index 44ad833..4a3fe5e 100644
--- a/src/document/sync.rs
+++ b/src/document/sync.rs
@@ -22,6 +22,13 @@ use crate::transport::SyncPayload;
 /// A sync message from a peer, decoded (`decode`), to apply (`SyncSession::receive`).
 pub(crate) struct Received(sync::Message);
 
+impl Received {
+    /// Whether it carries changes (it may carry only heads and what the peer needs).
+    pub(crate) fn carries_changes(&self) -> bool {
+        !self.0.changes.is_empty()
+    }
+}
+
 /// `payload` decoded. Before taking the document's lock: decoding a large message takes
 /// a while.
 pub(crate) fn decode(payload: &SyncPayload) -> anyhow::Result<Received> {
diff --git a/src/swarm/sync.rs b/src/swarm/sync.rs
index 187dacc..dd5444b 100644
--- a/src/swarm/sync.rs
+++ b/src/swarm/sync.rs
@@ -376,11 +376,12 @@ 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)) when applied, None when the message belongs
-        // to another epoch.
-        tokio::task::spawn_blocking(move || -> Result<Option<(bool, bool)>> {
+        // 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)>> {
             // Before the lock: decoding a large message takes a while.
             let message = crate::document::sync::decode(&message)?;
+            let carried = message.carries_changes();
             let mut replica = doc.blocking_write();
 
             // Authoritative check, under the same lock as the apply: the document may
@@ -430,13 +431,14 @@ pub(crate) async fn handle_sync(
                 // before it registered itself).
                 check_caught_up && from_ready && peer_state.sync.have_all_theirs(&replica.doc),
                 changed,
+                carried,
             )))
         })
         .await
         .context("Sync task panicked")??
     };
 
-    let Some((caught_up, changed)) = applied else {
+    let Some((caught_up, changed, carried)) = applied else {
         debug!(peer = %peer, epoch = %epoch_label, "Dropped sync message from another epoch");
         return Ok(());
     };
@@ -459,6 +461,14 @@ pub(crate) async fn handle_sync(
     // 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;
+    } else if carried
+        && let Some(writer) = writer
+        && state.initial_sync_done.load(Ordering::Acquire)
+    {
+        // A writer's changes along its tree that we had already: they came another way
+        // first, not saying whose (with an answer to one of our messages, say), so we
+        // passed nothing on then. Our children in its tree may lack them still.
+        crate::swarm::tree::forward(state, writer, peer).await;
     }
 
     state