introduce eager links

1 files changed, 13 insertions, 5 deletions

diff --git a/src/swarm/sync.rs b/src/swarm/sync.rs
index 3bfbf7d..37c3b0a 100644
--- a/src/swarm/sync.rs
+++ b/src/swarm/sync.rs
@@ -372,11 +372,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, a duplicate)) 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 carries_changes = message.carries_changes();
             let mut replica = doc.blocking_write();
 
             // Authoritative check, under the same lock as the apply: the document may
@@ -406,6 +407,9 @@ 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;
+            // Changes we had already, over a link we asked to be eager: the link isn't
+            // needed (`relay::prune`).
+            let duplicate = carries_changes && !changed && peer_state.eager_from;
             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.
@@ -426,13 +430,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,
+                duplicate,
             )))
         })
         .await
         .context("Sync task panicked")??
     };
 
-    let Some((caught_up, changed)) = applied else {
+    let Some((caught_up, changed, duplicate)) = applied else {
         debug!(peer = %peer, epoch = %epoch_label, "Dropped sync message from another epoch");
         return Ok(());
     };
@@ -451,7 +456,10 @@ pub(crate) async fn handle_sync(
 
     state.notify_store_changed();
 
-    crate::swarm::relay::pulled(state, peer);
+    crate::swarm::relay::pulled(state, peer, changed).await;
+    if duplicate {
+        crate::swarm::relay::prune(state, peer).await;
+    }
     // Caught up: our other neighbours may lack what the source had.
     if changed || just_caught_up {
         crate::swarm::relay::announce(state, peer).await;