diff options
Diffstat (limited to 'src/document')
| -rw-r--r-- | src/document/sync.rs | 108 |
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()); |
