pushed heads as tips

3 files changed, 63 insertions, 8 deletions

diff --git a/src/document/sync.rs b/src/document/sync.rs
index e7a2665..9415bd0 100644
--- a/src/document/sync.rs
+++ b/src/document/sync.rs
@@ -6,6 +6,7 @@
 //! forgetting what was sent when the peer still needs it) live here with it, so an
 //! Automerge upgrade that changes the state touches this file only.
 
+use std::collections::HashSet;
 use std::sync::atomic::{AtomicU64, Ordering};
 use std::time::Duration;
 
@@ -22,6 +23,29 @@ use crate::transport::SyncPayload;
 /// A sync message from a peer, decoded (`decode`), to apply (`SyncSession::receive`).
 pub(crate) struct Received(sync::Message);
 
+/// The changes none of the others depend on, sorted as heads are (`get_heads`).
+fn tips<'a>(
+    changes: impl Iterator<
+        Item = (
+            crate::document::ChangeHash,
+            &'a [crate::document::ChangeHash],
+        ),
+    >,
+) -> Vec<crate::document::ChangeHash> {
+    let changes: Vec<_> = changes.collect();
+    let depended: HashSet<crate::document::ChangeHash> = changes
+        .iter()
+        .flat_map(|(_, deps)| deps.iter().copied())
+        .collect();
+    let mut tips: Vec<crate::document::ChangeHash> = changes
+        .into_iter()
+        .map(|(hash, _)| hash)
+        .filter(|hash| !depended.contains(hash))
+        .collect();
+    tips.sort_unstable();
+    tips
+}
+
 impl Received {
     /// Whether it carries changes (it may carry only heads and what the peer needs).
     pub(crate) fn carries_changes(&self) -> bool {
@@ -32,11 +56,23 @@ impl Received {
     /// 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).
+    ///
+    /// No heads stand for the tips of the changes (`SyncSession::push`). Changes that
+    /// don't decode leave the heads out; applying them fails as it would have anyway.
     pub(crate) fn pushed(
         heads: Vec<crate::document::ChangeHash>,
         changes: Vec<Vec<u8>>,
         ours: Vec<crate::document::ChangeHash>,
     ) -> Self {
+        let heads = if heads.is_empty() {
+            let decoded: Vec<automerge::Change> = changes
+                .iter()
+                .filter_map(|bytes| automerge::Change::try_from(bytes.as_slice()).ok())
+                .collect();
+            tips(decoded.iter().map(|change| (change.hash(), change.deps())))
+        } else {
+            heads
+        };
         Self(sync::Message {
             heads,
             need: Vec::new(),
@@ -352,8 +388,11 @@ impl SyncSession {
             regenerate: false,
             assumed: None,
         });
+        // Our heads are mostly the tips of the changes pushed: then they go as none, and
+        // the receiver finds them in the changes (`Received::pushed`).
+        let tips = tips(changes.iter().map(|change| (change.hash(), change.deps())));
         Some((
-            heads,
+            if tips == heads { Vec::new() } else { heads },
             changes
                 .iter()
                 .map(|change| change.raw_bytes().to_vec())
@@ -1453,16 +1492,18 @@ mod tests {
         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);
+        assert!(
+            heads.is_empty(),
+            "our heads are the change's: they go as none"
+        );
         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();
+        let message = Received::pushed(heads, changes, before.clone());
+        assert_eq!(message.heads(), ours.get_heads(), "found in the changes");
+        b.receive(&mut theirs, message).unwrap();
         b.came_from_peer(&theirs, &before);
         assert_eq!(theirs.get_heads(), ours.get_heads());
         assert!(b.answered_as_delivered(&theirs, &theirs.get_heads()));
@@ -1472,6 +1513,19 @@ mod tests {
         );
     }
 
+    /// Tips are the changes none of the others depend on, sorted as heads are.
+    #[test]
+    fn tips_are_what_nothing_depends_on() {
+        let hash = |n: u8| crate::document::ChangeHash([n; 32]);
+        let (a, b, c, d) = (hash(1), hash(2), hash(3), hash(4));
+        let deps = [vec![], vec![a], vec![a], vec![b, c, hash(9)]];
+        let all = [a, b, c, d];
+        let tips_of = |which: &[usize]| tips(which.iter().map(|&i| (all[i], deps[i].as_slice())));
+        assert_eq!(tips_of(&[0, 1, 2, 3]), vec![d]);
+        assert_eq!(tips_of(&[2, 0, 1]), vec![b, c]);
+        assert_eq!(tips_of(&[]), Vec::new());
+    }
+
     /// Changes larger than the limit aren't pushed: they go as `generate` sends them.
     #[test]
     fn large_changes_are_not_pushed() {
diff --git a/src/transport/mod.rs b/src/transport/mod.rs
index 7e5d41b..2595257 100644
--- a/src/transport/mod.rs
+++ b/src/transport/mod.rs
@@ -746,7 +746,8 @@ pub(crate) enum SyncBody {
     /// Changes passed on along a writer's tree as they are, with the sender's heads,
     /// instead of an Automerge sync message (`AppState::push_changes`): where the sender
     /// takes the message as delivered, it has everything the receiver has, and needs
-    /// none of the rest of one (`SyncSession::push`).
+    /// none of the rest of one (`SyncSession::push`). No heads stand for the tips of
+    /// the changes, as the sender's heads mostly are.
     Changes {
         #[serde(with = "crate::transport::wire::codec::change_hashes")]
         heads: Vec<ChangeHash>,
diff --git a/src/transport/wire.rs b/src/transport/wire.rs
index 70ac3ce..580712e 100644
--- a/src/transport/wire.rs
+++ b/src/transport/wire.rs
@@ -14,7 +14,7 @@ use super::{EncodedMessage, MessageKind, ParsedMessage};
 
 /// The wire format's version, the first byte of every payload. Bump it with every change
 /// to a message type or a codec (`golden_bytes` fails then).
-pub(crate) const VERSION: u8 = 33;
+pub(crate) const VERSION: u8 = 34;
 
 /// What the second byte of a payload says.
 fn tag(kind: MessageKind) -> u8 {