push changes experiment

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]