any message taken as delivered

1 files changed, 82 insertions, 26 deletions

diff --git a/src/document/sync.rs b/src/document/sync.rs
index c432173..14adca7 100644
--- a/src/document/sync.rs
+++ b/src/document/sync.rs
@@ -234,16 +234,19 @@ impl SyncSession {
         })
     }
 
-    /// The message just generated passes a change on along a writer's tree, to a peer
-    /// that had nothing we lack (`have_all_theirs`), and no answer was awaited before it:
-    /// once it has the message, it holds exactly our `heads`, and answers nothing
-    /// (`answered_along_tree`). So it is taken as delivered: the peer's heads are ours,
-    /// and no answer is awaited (about half the sync messages, otherwise, are those
-    /// answers). Should it be lost, the announcements make the peer pull: the slow round
-    /// of its other neighbours, or ours where it reaches no other member
-    /// (`swarm::tree::Plan::alone`); should the peer have changes we lack after all, it
-    /// answers with them, as its heads differ from ours. Undone if our transport refuses
-    /// the message (`send_failed`).
+    /// The message just generated goes to a peer that had nothing we lack
+    /// (`have_all_theirs`), and no answer was awaited before it: once it has the message,
+    /// it holds exactly our `heads`, and answers nothing (`answered_as_delivered`; the
+    /// envelope says so, `SyncEnvelope::delivered`). So it is taken as delivered: the
+    /// peer's heads are ours, and no answer is awaited (otherwise about half the sync
+    /// messages are those answers: along trees, to pulls, to what is sent again). Should
+    /// it be lost, the peer still lacks it: where it asked (a pull), it asks again after
+    /// its own reply timeout; otherwise the announcements make it pull (the slow round of
+    /// its other neighbours, or ours where it reaches no other member:
+    /// `swarm::tree::Plan::alone`), or our next message, which it can't apply, makes it
+    /// ask. Should the peer have changes we lack after all, it answers with them, as its
+    /// 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 Some(sending) = &mut self.sending else {
             return;
@@ -258,11 +261,11 @@ impl SyncSession {
         self.reply = None;
     }
 
-    /// We applied a message passing a change on along a writer's tree: if we hold
-    /// exactly its sender's heads now, its sender took it as delivered
-    /// (`assume_delivered`), and 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_along_tree(&mut self, heads: &[crate::document::ChangeHash]) -> bool {
+    /// 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| {
@@ -619,10 +622,10 @@ mod tests {
         (a, b)
     }
 
-    /// A message along a tree to a peer that had everything else: taken as delivered,
-    /// and not answered; neither side has more to say.
+    /// A message to a peer that had everything else: taken as delivered, and not
+    /// answered; neither side has more to say.
     #[test]
-    fn tree_messages_need_no_answer() {
+    fn delivered_messages_need_no_answer() {
         use crate::document::testing::change;
         use automerge::{ROOT, transaction::Transactable};
         let (mut ours, mut theirs) = (Document::new(), Document::new());
@@ -638,7 +641,7 @@ mod tests {
         assert!(a.reply.is_none(), "no answer awaited");
 
         b.receive(&mut theirs, decode(&message).unwrap()).unwrap();
-        assert!(b.answered_along_tree(&theirs.get_heads()));
+        assert!(b.answered_as_delivered(&theirs.get_heads()));
         assert!(
             b.generate(&theirs, now, bytes).is_none(),
             "nothing to answer"
@@ -653,7 +656,7 @@ mod tests {
     /// The peer wrote meanwhile: its heads differ from ours after our message, so it
     /// answers, with its change.
     #[test]
-    fn tree_messages_are_answered_with_changes_we_lack() {
+    fn delivered_messages_are_answered_with_changes_we_lack() {
         use crate::document::testing::change;
         use automerge::{ROOT, transaction::Transactable};
         let (mut ours, mut theirs) = (Document::new(), Document::new());
@@ -668,17 +671,17 @@ mod tests {
         a.send_succeeded(a.id());
 
         b.receive(&mut theirs, decode(&message).unwrap()).unwrap();
-        assert!(!b.answered_along_tree(&theirs.get_heads()));
+        assert!(!b.answered_as_delivered(&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();
         assert_eq!(ours.get_heads(), theirs.get_heads());
     }
 
-    /// A message along a tree is lost, and the next one arrives: the peer can't apply
-    /// it, answers what it needs, and gets the lost change then.
+    /// A message taken as delivered is lost, and the next one arrives: the peer can't
+    /// apply it, answers what it needs, and gets the lost change then.
     #[test]
-    fn a_lost_tree_message_is_asked_for() {
+    fn a_lost_delivered_message_is_asked_for() {
         use crate::document::testing::change;
         use automerge::{ROOT, transaction::Transactable};
         let (mut ours, mut theirs) = (Document::new(), Document::new());
@@ -700,7 +703,7 @@ mod tests {
         a.send_succeeded(a.id());
 
         b.receive(&mut theirs, decode(&next).unwrap()).unwrap();
-        assert!(!b.answered_along_tree(&theirs.get_heads()));
+        assert!(!b.answered_as_delivered(&theirs.get_heads()));
         let ask = b
             .generate(&theirs, now, bytes)
             .expect("b asks for what it lacks");
@@ -713,10 +716,63 @@ mod tests {
         assert_eq!(ours.get_heads(), theirs.get_heads());
     }
 
+    /// Two messages taken as delivered arrive in reverse order: the peer asks for the
+    /// first, which arrives meanwhile, and what is sent again is taken as delivered too.
+    /// Nobody waits for an answer that won't come.
+    #[test]
+    fn reordered_delivered_messages_leave_nothing_awaited() {
+        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());
+        let send = |a: &mut SyncSession, ours: &Document| {
+            assert!(a.reply.is_none() && a.have_all_theirs(ours));
+            let message = a.generate(ours, now, bytes).unwrap();
+            a.assume_delivered(&ours.get_heads());
+            a.send_succeeded(a.id());
+            message
+        };
+
+        change(&mut ours, |doc| doc.put(ROOT, "first", 2).unwrap());
+        let first = send(&mut a, &ours);
+        change(&mut ours, |doc| doc.put(ROOT, "second", 3).unwrap());
+        let second = send(&mut a, &ours);
+
+        b.receive(&mut theirs, decode(&second).unwrap()).unwrap();
+        assert!(!b.answered_as_delivered(&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());
+
+        a.receive(&mut ours, decode(&ask).unwrap()).unwrap();
+        a.resend_if_needed("b");
+        let again = send(&mut a, &ours);
+        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.generate(&theirs, now, bytes).is_none(),
+            "nothing to answer"
+        );
+        assert!(a.reply.is_none(), "nothing awaited");
+        assert_eq!(ours.get_heads(), theirs.get_heads());
+    }
+
     /// Our transport refused a message taken as delivered: the peer's heads are as
     /// before, so the next message carries the change again.
     #[test]
-    fn a_refused_tree_message_goes_again() {
+    fn a_refused_delivered_message_goes_again() {
         use crate::document::testing::change;
         use automerge::{ROOT, transaction::Transactable};
         let (mut ours, mut theirs) = (Document::new(), Document::new());