delivered means has all up to our heads

1 files changed, 59 insertions, 13 deletions

diff --git a/src/document/sync.rs b/src/document/sync.rs
index 14adca7..fda8889 100644
--- a/src/document/sync.rs
+++ b/src/document/sync.rs
@@ -46,12 +46,18 @@ pub(crate) struct Sending {
     /// A sync pass was requested meanwhile. Once this one is handed over, the peer must
     /// be enqueued again so the deferred changes go out.
     pub(crate) regenerate: bool,
-    /// Taken as delivered (`assume_delivered`): the peer's heads and the shared heads as
-    /// they were before, put back if the send fails.
-    pub(crate) assumed: Option<(
-        Option<Vec<crate::document::ChangeHash>>,
-        Vec<crate::document::ChangeHash>,
-    )>,
+    /// Taken as delivered (`assume_delivered`): what we knew of the peer before, put
+    /// back if the send fails.
+    pub(crate) assumed: Option<Assumed>,
+}
+
+/// What `assume_delivered` replaces in the sync state, as it was.
+#[derive(Clone)]
+pub(crate) struct Assumed {
+    their_heads: Option<Vec<crate::document::ChangeHash>>,
+    shared_heads: Vec<crate::document::ChangeHash>,
+    their_have: Option<Vec<sync::Have>>,
+    their_need: Option<Vec<crate::document::ChangeHash>>,
 }
 
 /// Session IDs, unique within the process.
@@ -251,12 +257,23 @@ impl SyncSession {
         let Some(sending) = &mut self.sending else {
             return;
         };
-        sending.assumed = Some((
-            self.state.their_heads.clone(),
-            self.state.shared_heads.clone(),
-        ));
+        sending.assumed = Some(Assumed {
+            their_heads: self.state.their_heads.clone(),
+            shared_heads: self.state.shared_heads.clone(),
+            their_have: self.state.their_have.clone(),
+            their_need: self.state.their_need.clone(),
+        });
         self.state.their_heads = Some(heads.to_vec());
         self.state.shared_heads = heads.to_vec();
+        // And it has all up to our heads, needing nothing: the next message carries
+        // what came since, and if what we sent is forgotten (`resend_if_needed`), only
+        // that goes again, not all since the peer's last message, which may be long ago
+        // (no answers come).
+        self.state.their_have = Some(vec![sync::Have {
+            last_sync: heads.to_vec(),
+            bloom: sync::BloomFilter::default(),
+        }]);
+        self.state.their_need = Some(Vec::new());
         self.state.in_flight = false;
         self.reply = None;
     }
@@ -372,9 +389,11 @@ impl SyncSession {
         for hash in &sending.sent {
             self.state.sent_hashes.remove(hash);
         }
-        if let Some((their_heads, shared_heads)) = sending.assumed {
-            self.state.their_heads = their_heads;
-            self.state.shared_heads = shared_heads;
+        if let Some(assumed) = sending.assumed {
+            self.state.their_heads = assumed.their_heads;
+            self.state.shared_heads = assumed.shared_heads;
+            self.state.their_have = assumed.their_have;
+            self.state.their_need = assumed.their_need;
         }
         if let Some(wait) = &mut self.reply {
             wait.sent.retain(|hash| !sending.sent.contains(hash));
@@ -769,6 +788,33 @@ mod tests {
         assert_eq!(ours.get_heads(), theirs.get_heads());
     }
 
+    /// After messages taken as delivered (no answers, so nothing new heard of what the
+    /// peer has), forgetting what was sent resends only what came since the last of
+    /// them, not all since the peer's last message.
+    #[test]
+    fn after_delivered_messages_only_news_goes_again() {
+        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, mut b) = in_step(&mut ours, &mut theirs);
+        let (now, bytes) = (Instant::now(), SyncBytes::default());
+        for i in 0..5 {
+            change(&mut ours, |doc| doc.put(ROOT, format!("k{i}"), i).unwrap());
+            let message = a.generate(&ours, now, bytes).unwrap();
+            a.assume_delivered(&ours.get_heads());
+            a.send_succeeded(a.id());
+            b.receive(&mut theirs, decode(&message).unwrap()).unwrap();
+            assert!(b.answered_as_delivered(&theirs.get_heads()));
+        }
+        // As `resend_if_needed` does when the peer needs what we sent.
+        a.state.sent_hashes.clear();
+        change(&mut ours, |doc| doc.put(ROOT, "news", 6).unwrap());
+        let message = a.generate(&ours, now, bytes).unwrap();
+        let carried = sync::Message::decode(&message.0).unwrap().changes.len();
+        assert_eq!(carried, 1, "only the news");
+    }
+
     /// Our transport refused a message taken as delivered: the peer's heads are as
     /// before, so the next message carries the change again.
     #[test]