heads heard forget sent

3 files changed, 121 insertions, 1 deletions

diff --git a/src/document/sync.rs b/src/document/sync.rs
index 5d139e2..5669ce3 100644
--- a/src/document/sync.rs
+++ b/src/document/sync.rs
@@ -825,6 +825,56 @@ impl SyncSession {
         true
     }
 
+    /// The peer said its `heads` outside the sync (a keepalive STATUS): it has every
+    /// change in their history. Automerge forgets changes as sent only when the peer's
+    /// sync message says it has them, and a neighbour another parent brings the changes
+    /// to (`known_to_peer`) sends us none while it has nothing to write: counted as sent,
+    /// every change would stay so. So those in the history of `heads` count as sent no
+    /// more, and the peer as having all up to `heads`: the next message to it carries
+    /// what lies beyond them, not those. Only forward: where `heads` aren't all ours, or
+    /// don't cover what the peer said before or what we take it to have (a message of
+    /// ours still on its way, `assume_delivered`), nothing changes. Returns how many
+    /// changes count as sent no more.
+    pub(crate) fn heard_heads(
+        &mut self,
+        doc: &Document,
+        heads: &[crate::document::ChangeHash],
+    ) -> usize {
+        use crate::document::changes::{covers, holds_all};
+        if heads.is_empty() || !holds_all(doc, heads) {
+            return 0;
+        }
+        let behind = |older: &[crate::document::ChangeHash]| !covers(doc, heads, older);
+        if self.said_heads.as_deref().is_some_and(behind)
+            || self.state.their_heads.as_deref().is_some_and(behind)
+            || self
+                .state
+                .their_have
+                .iter()
+                .flatten()
+                .any(|have| behind(&have.last_sync))
+        {
+            return 0;
+        }
+        let beyond: HashSet<crate::document::ChangeHash> = doc
+            .get_changes_meta(heads)
+            .iter()
+            .map(|change| change.hash)
+            .collect();
+        let before = self.state.sent_hashes.len();
+        self.state
+            .sent_hashes
+            .retain(|hash| beyond.contains(hash) || doc.get_change_by_hash(hash).is_none());
+        self.state.their_heads = Some(heads.to_vec());
+        self.state.shared_heads = heads.to_vec();
+        self.state.their_have = Some(vec![sync::Have {
+            last_sync: heads.to_vec(),
+            bloom: sync::BloomFilter::default(),
+        }]);
+        self.said_heads = Some(heads.to_vec());
+        before - self.state.sent_hashes.len()
+    }
+
     /// The peer has heads we lack (its keepalive said so): ask for what they need,
     /// even though Automerge thinks a message is still in flight.
     pub(crate) fn pull_from(&mut self, heads: Vec<crate::document::ChangeHash>) {
@@ -1340,6 +1390,46 @@ mod tests {
         assert_eq!(a.sent_hashes_len(), before);
     }
 
+    /// A neighbour another parent brings our changes to sends us no sync message: what
+    /// counts as sent to it is forgotten once its keepalive says its heads, and nothing
+    /// goes to it again.
+    #[test]
+    fn heads_heard_forget_what_counts_as_sent() {
+        use crate::document::testing::change;
+        use automerge::{ROOT, transaction::Transactable};
+        let (mut ours, mut theirs) = (Document::new(), Document::new());
+        change(&mut ours, |doc| doc.put(ROOT, "early", 1).unwrap());
+        let (mut a, _b) = in_step(&mut ours, &mut theirs);
+        let (now, bytes) = (Instant::now(), SyncBytes::default());
+        for i in 0..100 {
+            change(&mut ours, |doc| doc.put(ROOT, "key", i).unwrap());
+            a.known_to_peer(&ours.get_heads(), now);
+        }
+        theirs.merge(&mut ours.clone()).unwrap();
+        assert_eq!(a.sent_hashes_len(), 100);
+
+        // Heads it said before we had them all: nothing changes.
+        assert_eq!(
+            a.heard_heads(&ours, &[crate::document::ChangeHash([7; 32])]),
+            0
+        );
+        assert_eq!(a.heard_heads(&ours, &theirs.get_heads()), 100);
+        assert_eq!(a.sent_hashes_len(), 0);
+        let carries = |message: Option<SyncPayload>| {
+            message.is_some_and(|message| decode(&message).unwrap().carries_changes())
+        };
+        assert!(!carries(a.generate(&ours, now, bytes)));
+
+        // Older heads than those, said later (a STATUS overtaken): nothing changes.
+        change(&mut ours, |doc| doc.put(ROOT, "key", 100).unwrap());
+        a.known_to_peer(&ours.get_heads(), now);
+        let older = theirs.get_heads();
+        theirs.merge(&mut ours.clone()).unwrap();
+        assert_eq!(a.heard_heads(&ours, &theirs.get_heads()), 1);
+        assert_eq!(a.heard_heads(&ours, &older), 0);
+        assert!(!carries(a.generate(&ours, now, bytes)));
+    }
+
     /// Automerge, as we rely on it: a message with no heads says the peer lost all it
     /// had, and what we counted as its goes again (why a push never goes without heads:
     /// `Received::pushed`, and an opening never goes empty: `leave_opening`).
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 {