diff options
Diffstat (limited to 'src/document/sync.rs')
| -rw-r--r-- | src/document/sync.rs | 123 |
1 files changed, 123 insertions, 0 deletions
diff --git a/src/document/sync.rs b/src/document/sync.rs index e3ff157..e7a2665 100644 --- a/src/document/sync.rs +++ b/src/document/sync.rs @@ -28,6 +28,28 @@ impl Received { !self.0.changes.is_empty() } + /// Changes pushed along a writer's tree (`SyncBody::Changes`), as the Automerge sync + /// message that would have brought them: the sender's `heads`, nothing needed, and + /// as what it has, everything we had (`ours`, our heads before: it takes the + /// message as delivered, so it had all of them). + pub(crate) fn pushed( + heads: Vec<crate::document::ChangeHash>, + changes: Vec<Vec<u8>>, + ours: Vec<crate::document::ChangeHash>, + ) -> Self { + Self(sync::Message { + heads, + need: Vec::new(), + have: vec![sync::Have { + last_sync: ours, + bloom: sync::BloomFilter::default(), + }], + changes: changes.into(), + flags: None, + version: sync::MessageVersion::V2, + }) + } + /// The sender's heads, as it says. pub(crate) fn heads(&self) -> &[crate::document::ChangeHash] { &self.0.heads @@ -287,6 +309,58 @@ impl SyncSession { } } + /// Instead of `generate`, where the message would be taken as delivered (the peer + /// has nothing we lack): the changes it would carry, as they are, with our heads + /// (`SyncBody::Changes`). Those since the peer's heads that don't count as sent + /// (as Automerge picks them; `known_to_peer`, `with_predecessors` alike), counted as + /// sent now, the message marked as handed over as `generate` does; the caller takes + /// it as delivered. None if there is nothing to push, the peer's heads aren't all + /// known, or the changes come to more than `limit` bytes (then `generate`). + pub(crate) fn push( + &mut self, + doc: &Document, + limit: usize, + ) -> Option<(Vec<crate::document::ChangeHash>, Vec<Vec<u8>>)> { + let theirs = self.state.their_heads.as_ref()?; + if theirs + .iter() + .any(|hash| doc.get_change_by_hash(hash).is_none()) + { + return None; + } + let changes: Vec<automerge::Change> = doc + .get_changes(theirs) + .into_iter() + .filter(|change| !self.state.sent_hashes.contains(&change.hash())) + .collect(); + let bytes: usize = changes.iter().map(|change| change.raw_bytes().len()).sum(); + if changes.is_empty() || bytes > limit { + return None; + } + let sent: Vec<crate::document::ChangeHash> = + changes.iter().map(automerge::Change::hash).collect(); + self.before_generate = ( + self.state.last_sent_heads.clone(), + self.state.have_responded, + ); + self.state.sent_hashes.extend(sent.iter().copied()); + let heads = doc.get_heads(); + self.state.last_sent_heads = heads.clone(); + self.state.have_responded = true; + self.sending = Some(Sending { + sent, + regenerate: false, + assumed: None, + }); + Some(( + heads, + changes + .iter() + .map(|change| change.raw_bytes().to_vec()) + .collect(), + )) + } + /// The message just generated is too large to send in one piece: take back what it /// carried, as if it had never been generated, and offer a delta instead. The /// message being handed over becomes the offer (it carries nothing); the answer is @@ -1365,6 +1439,55 @@ mod tests { assert_eq!(theirs.get_heads(), ours.get_heads(), "both in one message"); } + /// Pushed changes (`push`) do what the sync message they stand for would: the peer + /// applies them, is level with us, and has nothing to answer; nothing goes twice. + #[test] + fn pushed_changes_stand_for_a_message() { + 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 (heads, changes) = a.push(&ours, 1 << 20).expect("news to push"); + assert_eq!(changes.len(), 1); + a.assume_delivered(&ours.get_heads()); + a.send_succeeded(a.id()); + assert!(a.push(&ours, 1 << 20).is_none(), "sent already"); + + let before = theirs.get_heads(); + b.receive( + &mut theirs, + Received::pushed(heads, changes, before.clone()), + ) + .unwrap(); + b.came_from_peer(&theirs, &before); + assert_eq!(theirs.get_heads(), ours.get_heads()); + assert!(b.answered_as_delivered(&theirs, &theirs.get_heads())); + assert!( + b.generate(&theirs, now, bytes).is_none(), + "nothing to answer" + ); + } + + /// Changes larger than the limit aren't pushed: they go as `generate` sends them. + #[test] + fn large_changes_are_not_pushed() { + 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, _b) = in_step(&mut ours, &mut theirs); + change(&mut ours, |doc| { + doc.put(ROOT, "news", "x".repeat(100)).unwrap() + }); + assert!(a.push(&ours, 10).is_none()); + assert!(a.sending.is_none(), "nothing handed over"); + assert!(a.push(&ours, 1 << 20).is_some()); + } + /// Our transport refused a message taken as delivered: the peer's heads are as /// before, so the next message carries the change again. #[test] |