From c3faf6ecde8157154a01daa2ddc0ac02e6c761f1 Mon Sep 17 00:00:00 2001 From: tv Date: Wed, 4 Nov 2026 19:46:17 +0100 Subject: tree messages need no answer --- src/document/sync.rs | 194 +++++++++++++++++++++++++++++++++++++++++++++++++++ 1 file changed, 194 insertions(+) (limited to 'src/document') diff --git a/src/document/sync.rs b/src/document/sync.rs index 4a3fe5e..c432173 100644 --- a/src/document/sync.rs +++ b/src/document/sync.rs @@ -46,6 +46,12 @@ pub(crate) struct Sending { /// A sync pass was requested meanwhile. Once this one is handed over, the peer must /// be enqueued again so the deferred changes go out. pub(crate) regenerate: bool, + /// Taken as delivered (`assume_delivered`): the peer's heads and the shared heads as + /// they were before, put back if the send fails. + pub(crate) assumed: Option<( + Option>, + Vec, + )>, } /// Session IDs, unique within the process. @@ -165,6 +171,7 @@ impl SyncSession { self.sending = Some(Sending { sent, regenerate: false, + assumed: None, }); } payload @@ -227,6 +234,50 @@ 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`). + pub(crate) fn assume_delivered(&mut self, heads: &[crate::document::ChangeHash]) { + let Some(sending) = &mut self.sending else { + return; + }; + sending.assumed = Some(( + self.state.their_heads.clone(), + self.state.shared_heads.clone(), + )); + self.state.their_heads = Some(heads.to_vec()); + self.state.shared_heads = heads.to_vec(); + self.state.in_flight = false; + 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 { + 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; + } + self.state.last_sent_heads = heads.to_vec(); + self.state.have_responded = true; + true + } + /// The peer has heads we lack (its keepalive said so): ask for what they need, /// even though Automerge thinks a message is still in flight. pub(crate) fn pull_from(&mut self, heads: Vec) { @@ -318,6 +369,10 @@ impl SyncSession { for hash in &sending.sent { self.state.sent_hashes.remove(hash); } + if let Some((their_heads, shared_heads)) = sending.assumed { + self.state.their_heads = their_heads; + self.state.shared_heads = shared_heads; + } if let Some(wait) = &mut self.reply { wait.sent.retain(|hash| !sending.sent.contains(hash)); } @@ -471,6 +526,7 @@ mod tests { Some(Sending { sent: Vec::new(), regenerate: false, + assumed: None, }) } @@ -540,6 +596,144 @@ mod tests { assert!(carried_again > 4_000, "sent again in {carried_again} bytes"); } + /// 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()); + let (now, bytes) = (Instant::now(), SyncBytes::default()); + for _ in 0..10 { + let to_b = a.generate(ours, now, bytes); + a.send_succeeded(a.id()); + let to_a = b.generate(theirs, now, bytes); + b.send_succeeded(b.id()); + if to_b.is_none() && to_a.is_none() { + break; + } + if let Some(m) = to_b { + b.receive(theirs, decode(&m).unwrap()).unwrap(); + } + if let Some(m) = to_a { + a.receive(ours, decode(&m).unwrap()).unwrap(); + } + } + assert_eq!(ours.get_heads(), theirs.get_heads()); + (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. + #[test] + fn tree_messages_need_no_answer() { + 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, "news", 2).unwrap()); + assert!(a.have_all_theirs(&ours)); + let message = a.generate(&ours, now, bytes).unwrap(); + a.assume_delivered(&ours.get_heads()); + a.send_succeeded(a.id()); + 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.generate(&theirs, now, bytes).is_none(), + "nothing to answer" + ); + assert!( + a.generate(&ours, now, bytes).is_none(), + "nothing more to send" + ); + assert_eq!(ours.get_heads(), theirs.get_heads()); + } + + /// 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() { + 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, "ours", 2).unwrap()); + change(&mut theirs, |doc| doc.put(ROOT, "theirs", 3).unwrap()); + let message = a.generate(&ours, now, bytes).unwrap(); + a.assume_delivered(&ours.get_heads()); + a.send_succeeded(a.id()); + + b.receive(&mut theirs, decode(&message).unwrap()).unwrap(); + assert!(!b.answered_along_tree(&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. + #[test] + fn a_lost_tree_message_is_asked_for() { + 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, "lost", 2).unwrap()); + let _lost = a.generate(&ours, now, bytes).unwrap(); + a.assume_delivered(&ours.get_heads()); + a.send_succeeded(a.id()); + change(&mut ours, |doc| doc.put(ROOT, "next", 3).unwrap()); + assert!( + a.have_all_theirs(&ours), + "a takes the lost one as delivered" + ); + let next = a.generate(&ours, now, bytes).unwrap(); + a.assume_delivered(&ours.get_heads()); + a.send_succeeded(a.id()); + + b.receive(&mut theirs, decode(&next).unwrap()).unwrap(); + assert!(!b.answered_along_tree(&theirs.get_heads())); + let ask = b + .generate(&theirs, now, bytes) + .expect("b asks for what it lacks"); + b.send_succeeded(b.id()); + a.receive(&mut ours, decode(&ask).unwrap()).unwrap(); + a.resend_if_needed("b"); + let answer = a.generate(&ours, now, bytes).expect("a sends the lost one"); + a.send_succeeded(a.id()); + b.receive(&mut theirs, decode(&answer).unwrap()).unwrap(); + 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() { + 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, "news", 2).unwrap()); + let _refused = a.generate(&ours, now, bytes).unwrap(); + a.assume_delivered(&ours.get_heads()); + assert!(a.send_failed(a.id(), now).is_some()); + let again = a.generate(&ours, now, bytes).expect("sent again"); + a.send_succeeded(a.id()); + b.receive(&mut theirs, decode(&again).unwrap()).unwrap(); + assert_eq!(ours.get_heads(), theirs.get_heads()); + } + /// A joining node opens, the answer is too large and becomes a delta, the node gets /// the delta and says so: the answer to that carries our heads, so the node knows it /// has everything (`have_all_theirs`). -- cgit v1.3.1