heads heard forget sent

2 files changed, 31 insertions, 1 deletions

diff --git a/src/swarm/peer/inbound.rs b/src/swarm/peer/inbound.rs
index 1542d4f..8a424b8 100644
--- a/src/swarm/peer/inbound.rs
+++ b/src/swarm/peer/inbound.rs
@@ -12,7 +12,8 @@ use super::handshake::{
     discover_peers, ensure_handshake, establish_peer, peer_swarm, remember_peer, send_hello_reply,
 };
 use super::keepalive::{
-    heal_if_diverged, heard_from, local_status, note_alive, note_status, status_received,
+    heal_if_diverged, heard_from, heard_heads, local_status, note_alive, note_status,
+    status_received,
 };
 use super::lanes::SyncLanes;
 use crate::locks::MutexExt;
@@ -199,6 +200,7 @@ pub(crate) async fn handle_message(
             // no reason, as changes on their way to us make them differ while writes go on.
             if !remote.announcing {
                 status_received(state, &peer).await;
+                heard_heads(state, &peer, &remote);
             }
             let local = on_status(state, &peer, &remote, reactivated).await;
             if remote.announcing {
diff --git a/src/swarm/peer/keepalive.rs b/src/swarm/peer/keepalive.rs
index d698585..693a370 100644
--- a/src/swarm/peer/keepalive.rs
+++ b/src/swarm/peer/keepalive.rs
@@ -218,6 +218,34 @@ pub(super) async fn status_received(state: &AppState, peer: &str) {
     }
 }
 
+/// A keepalive STATUS said the peer's heads: what it has counts as sent to it no more
+/// (`SyncSession::heard_heads`). Off the consumer: it takes the document's lock.
+pub(super) fn heard_heads(state: &AppState, peer: &str, remote: &transport::Status) {
+    let Some(heads) = remote.heads.heads() else {
+        return;
+    };
+    if heads.is_empty() || remote.epoch != state.local_epoch() {
+        return;
+    }
+    let (state, peer, heads, epoch) =
+        (state.clone(), peer.to_owned(), heads.to_vec(), remote.epoch);
+    tokio::spawn(async move {
+        let Some(peer_state) = state.get_peer_state(&peer).await else {
+            return;
+        };
+        let replica = state.doc.read().await;
+        // The same epoch still: a switch meanwhile makes the heads another document's.
+        if replica.epoch != epoch {
+            return;
+        }
+        peer_state
+            .lock()
+            .await
+            .sync
+            .heard_heads(&replica.doc, &heads);
+    });
+}
+
 /// What a peer's status says about it.
 pub(super) async fn note_status(state: &AppState, peer: &str, remote: &transport::Status) {
     if let Some(peer_state) = state.get_peer_state(peer).await {