late messages keep newer heads

1 files changed, 73 insertions, 0 deletions

diff --git a/src/document/sync.rs b/src/document/sync.rs
index b3f8221..f9f0295 100644
--- a/src/document/sync.rs
+++ b/src/document/sync.rs
@@ -187,6 +187,10 @@ pub(crate) struct SyncSession {
     /// builds on them (`with_predecessors`).
     known_elsewhere: std::collections::HashMap<crate::document::ChangeHash, Instant>,
 
+    /// The heads the peer said last, in the newest of its messages so far (one that
+    /// comes late doesn't count: `older_keeps_newer`).
+    said_heads: Option<Vec<crate::document::ChangeHash>>,
+
     /// The answer we expect to our last sync message(s), if any.
     pub(crate) reply: Option<SyncReplyWait>,
     /// How long to wait for it; doubles while answers stay out.
@@ -230,6 +234,7 @@ impl SyncSession {
             before_receive: None,
             before_generate: (Vec::new(), false),
             known_elsewhere: std::collections::HashMap::new(),
+            said_heads: None,
             reply: None,
             reply_backoff,
         }
@@ -305,11 +310,43 @@ impl SyncSession {
     ) -> Result<(), AutomergeError> {
         self.delta_offered = false;
         self.before_receive = Some(self.peer_view());
+        let said = message.0.heads.clone();
         doc.receive_sync_message_log_patches(&mut self.state, message.0, log)?;
+        self.older_keeps_newer(doc, &said);
         self.known_last_sync(doc);
         Ok(())
     }
 
+    /// A message older than one the peer sent before (it came late: overtaken, or a
+    /// copy) names heads in the history of those it said then: Automerge takes them as
+    /// the peer's now, and what came since would seem to lack to it, the changes it
+    /// sent us among them (sent back to it, by a push above all: `push` sends what lies
+    /// beyond its heads). So the peer's heads, and those we share, are what it said then
+    /// (`said_heads`). Only by what it said: heads we took for its (a message of ours
+    /// taken as delivered, `assume_delivered`) may be more than it has, and a peer
+    /// back from away says older heads than those. Not where it says it has nothing (no
+    /// heads: it lost everything).
+    fn older_keeps_newer(&mut self, doc: &Document, said: &[crate::document::ChangeHash]) {
+        let newer = self.said_heads.clone();
+        let older = newer.as_ref().is_some_and(|newer| {
+            // Older: all its heads in our history, and in that of the heads said before
+            // (a message whose changes wait for predecessors names heads we lack:
+            // newer).
+            !said.is_empty()
+                && said != newer.as_slice()
+                && crate::document::changes::holds_all(doc, said)
+                && crate::document::changes::holds_all(doc, newer)
+                && crate::document::changes::covers(doc, newer, said)
+        });
+        match newer {
+            Some(newer) if older => {
+                self.state.their_heads = Some(newer.clone());
+                self.state.shared_heads = newer;
+            }
+            _ => self.said_heads = Some(said.to_vec()),
+        }
+    }
+
     /// 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
@@ -1293,6 +1330,42 @@ mod tests {
         assert!(b.have_all_theirs(&theirs));
     }
 
+    /// A message that comes late (a copy of one before, or overtaken) names heads older
+    /// than the peer said since: they stay its heads, and what it sent us since doesn't
+    /// go back to it (as it did by a push: beyond the older heads, and no longer counted
+    /// as sent once the peer's heads showed it).
+    #[test]
+    fn a_late_message_takes_the_peer_back_nowhere() {
+        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());
+
+        change(&mut ours, |doc| doc.put(ROOT, "first", 2).unwrap());
+        let first = a.generate(&ours, now, bytes).expect("the first");
+        a.send_succeeded(a.id());
+        change(&mut ours, |doc| doc.put(ROOT, "then", 3).unwrap());
+        let then = ours.get_heads();
+        let second = a.generate(&ours, now, bytes).expect("the second");
+        a.send_succeeded(a.id());
+
+        // Both arrive, the second twice (so it no longer counts as sent: the peer's
+        // heads show it); then the first again, late.
+        for message in [&first, &second, &second] {
+            let before = theirs.get_heads();
+            b.receive(&mut theirs, decode(message).unwrap()).unwrap();
+            b.came_from_peer(&theirs, &before);
+        }
+        b.receive(&mut theirs, decode(&first).unwrap()).unwrap();
+        assert_eq!(b.state.their_heads.as_deref(), Some(&then[..]));
+        assert!(
+            b.push(&theirs, 1 << 20).is_none(),
+            "nothing of a's goes back to it"
+        );
+    }
+
     /// Two sessions exchanging messages until neither has anything to say.
     fn in_step(ours: &mut Document, theirs: &mut Document) -> (SyncSession, SyncSession) {
         let (mut a, mut b) = (session(), session());