pull missing changes

Nodes push changes to their neighbours but don't relay them directly.
In a partial mesh, nodes can learn about missing changes via keepalive
messages and can pull them from their neighbours.

1 files changed, 14 insertions, 5 deletions

diff --git a/src/swarm/sync.rs b/src/swarm/sync.rs
index b8d16aa..383cdaf 100644
--- a/src/swarm/sync.rs
+++ b/src/swarm/sync.rs
@@ -310,8 +310,9 @@ pub(crate) async fn handle_sync(
         let message = envelope.message;
         let peer = peer.to_owned();
 
-        // Some(caught_up) when applied, None when the message belongs to another epoch.
-        tokio::task::spawn_blocking(move || -> Result<Option<bool>> {
+        // Some((caught_up, heads changed)) when applied, None when the message belongs
+        // to another epoch.
+        tokio::task::spawn_blocking(move || -> Result<Option<(bool, bool)>> {
             let mut replica = doc.blocking_write();
 
             // Authoritative check, under the same lock as the apply: the document may
@@ -322,6 +323,7 @@ pub(crate) async fn handle_sync(
 
             let mut peer_state = peer_state_mutex.blocking_lock();
             peer_state.reports_ready = Some(from_ready);
+            let heads_before = replica.doc.get_heads();
 
             replica
                 .doc
@@ -332,8 +334,9 @@ pub(crate) async fn handle_sync(
             peer_state.sync.reply_settled();
 
             resend_if_needed(&peer, &mut peer_state.sync.state);
+            let changed = replica.doc.get_heads() != heads_before;
 
-            Ok(Some(
+            Ok(Some((
                 // Only from a ready peer: a peer still catching up has empty heads, which
                 // every document "contains".
                 check_caught_up
@@ -348,13 +351,14 @@ pub(crate) async fn handle_sync(
                                 .iter()
                                 .all(|h| replica.doc.get_change_by_hash(h).is_some())
                         }),
-            ))
+                changed,
+            )))
         })
         .await
         .context("Sync task panicked")??
     };
 
-    let Some(caught_up) = applied else {
+    let Some((caught_up, changed)) = applied else {
         debug!(peer = %peer, epoch = %epoch_label, "Dropped sync message from another epoch");
         return Ok(());
     };
@@ -368,6 +372,11 @@ pub(crate) async fn handle_sync(
 
     state.notify_store_changed();
 
+    crate::swarm::relay::pulled(state, peer);
+    if changed {
+        crate::swarm::relay::announce(state, peer).await;
+    }
+
     state
         .sync_events
         .send(SyncEvent::MessageReceived(peer.to_owned()))