diff options
Diffstat (limited to 'src/document')
| -rw-r--r-- | src/document/sync.rs | 72 |
1 files changed, 59 insertions, 13 deletions
diff --git a/src/document/sync.rs b/src/document/sync.rs index 14adca7..fda8889 100644 --- a/src/document/sync.rs +++ b/src/document/sync.rs @@ -46,12 +46,18 @@ 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<crate::document::ChangeHash>>, - Vec<crate::document::ChangeHash>, - )>, + /// Taken as delivered (`assume_delivered`): what we knew of the peer before, put + /// back if the send fails. + pub(crate) assumed: Option<Assumed>, +} + +/// What `assume_delivered` replaces in the sync state, as it was. +#[derive(Clone)] +pub(crate) struct Assumed { + their_heads: Option<Vec<crate::document::ChangeHash>>, + shared_heads: Vec<crate::document::ChangeHash>, + their_have: Option<Vec<sync::Have>>, + their_need: Option<Vec<crate::document::ChangeHash>>, } /// Session IDs, unique within the process. @@ -251,12 +257,23 @@ impl SyncSession { let Some(sending) = &mut self.sending else { return; }; - sending.assumed = Some(( - self.state.their_heads.clone(), - self.state.shared_heads.clone(), - )); + sending.assumed = Some(Assumed { + their_heads: self.state.their_heads.clone(), + shared_heads: self.state.shared_heads.clone(), + their_have: self.state.their_have.clone(), + their_need: self.state.their_need.clone(), + }); self.state.their_heads = Some(heads.to_vec()); self.state.shared_heads = heads.to_vec(); + // And it has all up to our heads, needing nothing: the next message carries + // what came since, and if what we sent is forgotten (`resend_if_needed`), only + // that goes again, not all since the peer's last message, which may be long ago + // (no answers come). + self.state.their_have = Some(vec![sync::Have { + last_sync: heads.to_vec(), + bloom: sync::BloomFilter::default(), + }]); + self.state.their_need = Some(Vec::new()); self.state.in_flight = false; self.reply = None; } @@ -372,9 +389,11 @@ 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(assumed) = sending.assumed { + self.state.their_heads = assumed.their_heads; + self.state.shared_heads = assumed.shared_heads; + self.state.their_have = assumed.their_have; + self.state.their_need = assumed.their_need; } if let Some(wait) = &mut self.reply { wait.sent.retain(|hash| !sending.sent.contains(hash)); @@ -769,6 +788,33 @@ mod tests { assert_eq!(ours.get_heads(), theirs.get_heads()); } + /// After messages taken as delivered (no answers, so nothing new heard of what the + /// peer has), forgetting what was sent resends only what came since the last of + /// them, not all since the peer's last message. + #[test] + fn after_delivered_messages_only_news_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()); + for i in 0..5 { + change(&mut ours, |doc| doc.put(ROOT, format!("k{i}"), i).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_as_delivered(&theirs.get_heads())); + } + // As `resend_if_needed` does when the peer needs what we sent. + a.state.sent_hashes.clear(); + change(&mut ours, |doc| doc.put(ROOT, "news", 6).unwrap()); + let message = a.generate(&ours, now, bytes).unwrap(); + let carried = sync::Message::decode(&message.0).unwrap().changes.len(); + assert_eq!(carried, 1, "only the news"); + } + /// Our transport refused a message taken as delivered: the peer's heads are as /// before, so the next message carries the change again. #[test] |