1 files changed, 90 insertions, 0 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`).