no reset for heads still on their way

1 files changed, 44 insertions, 1 deletions

diff --git a/src/document/sync.rs b/src/document/sync.rs
index 29efb5d..f2d6669 100644
--- a/src/document/sync.rs
+++ b/src/document/sync.rs
@@ -223,7 +223,43 @@ impl SyncSession {
     ) -> 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)
+        doc.receive_sync_message_log_patches(&mut self.state, message.0, log)?;
+        self.known_last_sync(doc);
+        Ok(())
+    }
+
+    /// The peer's `have` names what it last shared with us by heads (`last_sync`), and
+    /// one that took a message to us as delivered names the heads that message brought.
+    /// Where the next message overtook it, we don't have those heads yet: Automerge
+    /// would take that for a document that lost changes and answer with a reset, and
+    /// the peer would forget all it ever sent us and send it again, what is still on its
+    /// way included (on slow links, every change several times). So heads in it that we
+    /// lack give way to the heads we share with the peer (`shared_heads`). The bloom
+    /// filter stays: it covers the changes since its own heads, which leaves out at
+    /// worst a few that go again.
+    fn known_last_sync(&mut self, doc: &Document) {
+        let Some(haves) = &mut self.state.their_have else {
+            return;
+        };
+        for have in haves.iter_mut() {
+            if have
+                .last_sync
+                .iter()
+                .all(|hash| doc.get_change_by_hash(hash).is_some())
+            {
+                continue;
+            }
+            let mut known: Vec<crate::document::ChangeHash> = have
+                .last_sync
+                .iter()
+                .filter(|hash| doc.get_change_by_hash(hash).is_some())
+                .chain(self.state.shared_heads.iter())
+                .copied()
+                .collect();
+            known.sort_unstable();
+            known.dedup();
+            have.last_sync = known;
+        }
     }
 
     /// The message just generated is too large to send in one piece: take back what it
@@ -862,6 +898,13 @@ mod tests {
         let ask = b
             .generate(&theirs, now, bytes)
             .expect("b asks for the first");
+        // The second named the heads the first brought as shared, which b lacks: no
+        // reset for that, else a forgets all it sent.
+        let flags = sync::Message::decode(&ask.0).unwrap().flags;
+        assert!(
+            !flags.is_some_and(|flags| flags.contains(sync::MessageFlags::SYNC_RESET)),
+            "b asked with a reset"
+        );
         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");