older delivered messages leave us level

1 files changed, 73 insertions, 35 deletions

diff --git a/src/document/sync.rs b/src/document/sync.rs
index 628fd4f..f1582a7 100644
--- a/src/document/sync.rs
+++ b/src/document/sync.rs
@@ -48,12 +48,14 @@ pub(crate) struct Sending {
     pub(crate) regenerate: bool,
     /// Taken as delivered (`assume_delivered`): what we knew of the peer before, put
     /// back if the send fails.
-    pub(crate) assumed: Option<Assumed>,
+    pub(crate) assumed: Option<PeerView>,
 }
 
-/// What `assume_delivered` replaces in the sync state, as it was.
+/// What the sync state knows of the peer: its heads, the heads we share, what it has
+/// and needs. Saved to be put back: by `assume_delivered` if the send fails, and before
+/// receiving, for a message older than the last (`answered_as_delivered`).
 #[derive(Clone)]
-pub(crate) struct Assumed {
+pub(crate) struct PeerView {
     their_heads: Option<Vec<crate::document::ChangeHash>>,
     shared_heads: Vec<crate::document::ChangeHash>,
     their_have: Option<Vec<sync::Have>>,
@@ -89,6 +91,9 @@ pub(crate) struct SyncSession {
     /// (`withdraw_for_delta`): no more messages until it answers, or the answer is
     /// overdue (then the offer goes out again).
     pub(crate) delta_offered: bool,
+    /// What the sync state knew of the peer before the last message from it
+    /// (`answered_as_delivered`).
+    before_receive: Option<PeerView>,
     /// What the last `generate` changed in the sync state besides the hashes sent: the
     /// heads it last sent, and whether it had answered yet. Restored when the message is
     /// withdrawn for a delta: otherwise Automerge believes the peer heard our heads, and
@@ -129,6 +134,7 @@ impl SyncSession {
             retry_at: None,
             send_backoff: Backoff::new(SEND_RETRY, SEND_RETRY_MAX),
             delta_offered: false,
+            before_receive: None,
             before_generate: (Vec::new(), false),
             reply: None,
             reply_backoff,
@@ -204,6 +210,7 @@ impl SyncSession {
         log: &mut automerge::PatchLog,
     ) -> Result<(), AutomergeError> {
         self.delta_offered = false;
+        self.before_receive = Some(self.peer_view());
         doc.receive_sync_message_log_patches(&mut self.state, message.0, log)
     }
 
@@ -254,15 +261,11 @@ impl SyncSession {
     /// heads differ from ours. Undone if our transport refuses the message
     /// (`send_failed`).
     pub(crate) fn assume_delivered(&mut self, heads: &[crate::document::ChangeHash]) {
+        let view = self.peer_view();
         let Some(sending) = &mut self.sending else {
             return;
         };
-        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(),
-        });
+        sending.assumed = Some(view);
         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
@@ -278,20 +281,55 @@ impl SyncSession {
         self.reply = None;
     }
 
+    /// What the sync state knows of the peer (`PeerView`).
+    fn peer_view(&self) -> PeerView {
+        PeerView {
+            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(),
+        }
+    }
+
     /// We applied a message its sender took as delivered (`assume_delivered`,
-    /// `SyncEnvelope::delivered`): if we hold exactly its sender's heads now, Automerge's
-    /// answer, our heads, would tell it nothing. So none goes: we count as having said
-    /// our `heads`. Returns whether so.
-    pub(crate) fn answered_as_delivered(&mut self, heads: &[crate::document::ChangeHash]) -> bool {
-        let mut ours = heads.to_vec();
-        ours.sort_unstable();
-        let theirs = self.state.their_heads.as_ref().map(|theirs| {
-            let mut theirs = theirs.clone();
-            theirs.sort_unstable();
-            theirs
-        });
-        if theirs.as_ref() != Some(&ours) {
-            return false;
+    /// `SyncEnvelope::delivered`) to `doc`: if we hold exactly its sender's heads now,
+    /// Automerge's answer, our heads, would tell it nothing. So none goes: we count as
+    /// having said our `heads`. Returns whether so.
+    ///
+    /// So also where the message is older than the last one from the peer (they
+    /// overtake each other, as no answers are awaited) and we hold exactly the heads
+    /// that one said: the older one tells nothing new of the peer, all its heads are in
+    /// our history, and we are level with the peer as of the newer. Otherwise the
+    /// older one's view of what we have would stand, and the answer to it would carry
+    /// back to the peer the changes the newer one brought us.
+    pub(crate) fn answered_as_delivered(
+        &mut self,
+        doc: &Document,
+        heads: &[crate::document::ChangeHash],
+    ) -> bool {
+        let sorted = |heads: &[crate::document::ChangeHash]| {
+            let mut heads = heads.to_vec();
+            heads.sort_unstable();
+            heads
+        };
+        let ours = sorted(heads);
+        let level = |theirs: &Option<Vec<crate::document::ChangeHash>>| {
+            theirs.as_deref().map(sorted).as_ref() == Some(&ours)
+        };
+        let before = self.before_receive.take();
+        if !level(&self.state.their_heads) {
+            let older = self.state.their_heads.as_ref().is_some_and(|theirs| {
+                theirs
+                    .iter()
+                    .all(|hash| doc.get_change_by_hash(hash).is_some())
+            });
+            let Some(before) = before.filter(|before| older && level(&before.their_heads)) else {
+                return false;
+            };
+            self.state.their_heads = before.their_heads;
+            self.state.their_have = before.their_have;
+            self.state.their_need = before.their_need;
+            self.state.shared_heads = heads.to_vec();
         }
         self.state.last_sent_heads = heads.to_vec();
         self.state.have_responded = true;
@@ -690,7 +728,7 @@ mod tests {
         assert!(a.reply.is_none(), "no answer awaited");
 
         b.receive(&mut theirs, decode(&message).unwrap()).unwrap();
-        assert!(b.answered_as_delivered(&theirs.get_heads()));
+        assert!(b.answered_as_delivered(&theirs, &theirs.get_heads()));
         assert!(
             b.generate(&theirs, now, bytes).is_none(),
             "nothing to answer"
@@ -720,7 +758,7 @@ mod tests {
         a.send_succeeded(a.id());
 
         b.receive(&mut theirs, decode(&message).unwrap()).unwrap();
-        assert!(!b.answered_as_delivered(&theirs.get_heads()));
+        assert!(!b.answered_as_delivered(&theirs, &theirs.get_heads()));
         let answer = b.generate(&theirs, now, bytes).expect("b answers");
         b.send_succeeded(b.id());
         a.receive(&mut ours, decode(&answer).unwrap()).unwrap();
@@ -752,7 +790,7 @@ mod tests {
         a.send_succeeded(a.id());
 
         b.receive(&mut theirs, decode(&next).unwrap()).unwrap();
-        assert!(!b.answered_as_delivered(&theirs.get_heads()));
+        assert!(!b.answered_as_delivered(&theirs, &theirs.get_heads()));
         let ask = b
             .generate(&theirs, now, bytes)
             .expect("b asks for what it lacks");
@@ -790,17 +828,20 @@ mod tests {
         let second = send(&mut a, &ours);
 
         b.receive(&mut theirs, decode(&second).unwrap()).unwrap();
-        assert!(!b.answered_as_delivered(&theirs.get_heads()));
+        assert!(!b.answered_as_delivered(&theirs, &theirs.get_heads()));
         let ask = b
             .generate(&theirs, now, bytes)
             .expect("b asks for the first");
         b.send_succeeded(b.id());
         b.receive(&mut theirs, decode(&first).unwrap()).unwrap();
         assert_eq!(theirs.get_heads(), ours.get_heads(), "b has both now");
-        // The first says older heads than b's now: b tells its own, once.
-        assert!(!b.answered_as_delivered(&theirs.get_heads()));
-        let level = b.generate(&theirs, now, bytes);
-        b.send_succeeded(b.id());
+        // The first is older than the second, whose heads b holds now: level with a as
+        // of that, b says nothing, and doesn't send a the second back.
+        assert!(b.answered_as_delivered(&theirs, &theirs.get_heads()));
+        assert!(
+            b.generate(&theirs, now, bytes).is_none(),
+            "nothing sent back"
+        );
 
         a.receive(&mut ours, decode(&ask).unwrap()).unwrap();
         a.resend_if_needed(&ours, "b");
@@ -810,11 +851,8 @@ mod tests {
             carried, 1,
             "only the first again, not the second, on its way"
         );
-        if let Some(level) = level {
-            a.receive(&mut ours, decode(&level).unwrap()).unwrap();
-        }
         b.receive(&mut theirs, decode(&again).unwrap()).unwrap();
-        assert!(b.answered_as_delivered(&theirs.get_heads()));
+        assert!(b.answered_as_delivered(&theirs, &theirs.get_heads()));
         assert!(
             b.generate(&theirs, now, bytes).is_none(),
             "nothing to answer"
@@ -840,7 +878,7 @@ mod tests {
             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()));
+            assert!(b.answered_as_delivered(&theirs, &theirs.get_heads()));
         }
         // As `resend_if_needed` does when the peer needs what we sent.
         a.state.sent_hashes.clear();