delta from held heads

5 files changed, 88 insertions, 30 deletions

diff --git a/docs/peers.md b/docs/peers.md
index eb1eb5c..70437ab 100644
--- a/docs/peers.md
+++ b/docs/peers.md
@@ -162,8 +162,9 @@ A sync message larger than 16 chunks (16 × `--blob-chunk`, 1 MiB by default) is
 sent: it would be all or nothing, sent again whole after a lost copy, and on a slow link
 may not arrive within the transport's delivery limit (Mycelium's five minutes). The sender takes it back and offers a
 *delta* instead: the changes in the history of its heads but not of the peer's (as far
-as it knows them). The peer fetches them as a blob, in chunks, from the sender and any
-other peer holding those heads (every such peer produces the same bytes), loads them,
+as it knows them; of those, the ones it holds, and for any it lacks, the heads they
+share). The peer fetches them as a blob, in chunks, from the sender and any other peer
+holding both sets of heads (every such peer produces the same bytes), loads them,
 and syncs on; the rest goes in ordinary messages. Until the peer answers, the sender
 sends it nothing; if the answer is overdue, it offers again (the same offer, without
 generating the message again, while neither side's heads moved). This is how a join without a
diff --git a/src/document/changes.rs b/src/document/changes.rs
index 345f151..0409c50 100644
--- a/src/document/changes.rs
+++ b/src/document/changes.rs
@@ -67,24 +67,20 @@ pub(crate) fn change_from_bytes(bytes: Vec<u8>, hash: ChangeHash) -> Result<Chan
 
 /// The bytes of the delta from `from` to `to` in `doc`: the changes in the history of
 /// `to` but not of `from`, each as encoded, sorted by hash and concatenated
-/// (`load_delta` reads that). The same on every node holding `to`, whatever else it
-/// has. None if `doc` lacks part of `to`.
+/// (`load_delta` reads that). The same on every node holding `to` and `from`, whatever
+/// else it has. None if `doc` lacks part of either: without one of `from`, the delta
+/// would carry that one's history too, where `to` shares it, and differ from another
+/// node's (a download takes chunks from several, checked against one's manifest).
 pub(crate) fn delta_bytes(
     doc: &Document,
     from: &[ChangeHash],
     to: &[ChangeHash],
 ) -> Option<Vec<u8>> {
-    if !missing(doc, to).is_empty() {
+    if !missing(doc, to).is_empty() || !missing(doc, from).is_empty() {
         return None;
     }
-    // Heads of `from` we don't have can't hide anything we have.
-    let from: Vec<ChangeHash> = from
-        .iter()
-        .copied()
-        .filter(|head| doc.get_change_by_hash(head).is_some())
-        .collect();
     let since: HashMap<ChangeHash, (Vec<ChangeHash>, Vec<u8>)> = doc
-        .get_changes(&from)
+        .get_changes(from)
         .iter()
         .map(|change| {
             (
@@ -181,7 +177,12 @@ mod tests {
     fn no_delta_without_its_heads() {
         let mut doc = Document::new();
         change(&mut doc, |doc| doc.put(ROOT, "k", 1_i64).unwrap());
+        let heads = doc.get_heads();
         assert!(delta_bytes(&doc, &[], &[ChangeHash([7; 32])]).is_none());
+        assert!(
+            delta_bytes(&doc, &[ChangeHash([7; 32])], &heads).is_none(),
+            "nor without where it starts"
+        );
     }
 
     /// A change loads from its bytes only as the change it was asked for.
diff --git a/src/document/sync.rs b/src/document/sync.rs
index 8265a02..0aeec83 100644
--- a/src/document/sync.rs
+++ b/src/document/sync.rs
@@ -273,14 +273,18 @@ impl SyncSession {
     /// 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
-    /// still awaited, and if it doesn't come, the offer is made again. Returns the
-    /// peer's heads as far as we know them: where the delta starts.
+    /// still awaited, and if it doesn't come, the offer is made again. Returns where
+    /// the delta starts (`delta_from`).
     ///
     /// An answer is awaited even where none was (the peer's heads were ours: `generate`),
     /// so the offer lapses at the reply timeout at the latest: nothing else ends it but a
     /// message from the peer, and one that has everything has nothing to say. Without
     /// that, no message would go to it again.
-    pub(crate) fn withdraw_for_delta(&mut self, now: Instant) -> Vec<crate::document::ChangeHash> {
+    pub(crate) fn withdraw_for_delta(
+        &mut self,
+        doc: &Document,
+        now: Instant,
+    ) -> Vec<crate::document::ChangeHash> {
         if self.reply.is_none() {
             self.expect_reply(Vec::new(), now, SyncBytes::default(), 0);
         }
@@ -296,7 +300,24 @@ impl SyncSession {
         self.state.in_flight = false;
         (self.state.last_sent_heads, self.state.have_responded) = self.before_generate.clone();
         self.delta_offered = true;
-        self.state.their_heads.clone().unwrap_or_default()
+        self.delta_from(doc)
+    }
+
+    /// Where a delta to the peer starts, sorted: its heads as far as we know them, of
+    /// those we hold. Every node that serves the delta must hold all of them
+    /// (`document::changes::delta_bytes`), so the same bytes come from each; we do. In
+    /// place of heads we lack (changes it has and we don't), the heads we share: the
+    /// delta may then carry some it has, but nothing it lacks.
+    fn delta_from(&self, doc: &Document) -> Vec<crate::document::ChangeHash> {
+        let known = |hash: &&crate::document::ChangeHash| doc.get_change_by_hash(hash).is_some();
+        let theirs = self.state.their_heads.as_deref().unwrap_or_default();
+        let mut from: Vec<_> = theirs.iter().filter(known).copied().collect();
+        if from.len() < theirs.len() {
+            from.extend(self.state.shared_heads.iter().filter(known).copied());
+        }
+        from.sort_unstable();
+        from.dedup();
+        from
     }
 
     /// We offered the delta from `from` to `to` in place of a message of `bytes`.
@@ -310,11 +331,12 @@ impl SyncSession {
     }
 
     /// The delta to offer again, if nothing changed since the last offer (`offered`):
-    /// the peer's heads as we know them, our `heads`, and the message too large for
+    /// where it starts (`delta_from`), our `heads`, and the message too large for
     /// `threshold` still. Then the message would be the same, and so the delta: its
     /// `from`, offered as by `withdraw_for_delta`, without generating the message.
     pub(crate) fn offer_again(
         &mut self,
+        doc: &Document,
         heads: &[crate::document::ChangeHash],
         threshold: usize,
         now: Instant,
@@ -326,8 +348,7 @@ impl SyncSession {
             heads.dedup();
             heads
         };
-        let theirs = sorted(self.state.their_heads.as_deref().unwrap_or_default());
-        if *bytes <= threshold || theirs != *from || sorted(heads) != *to {
+        if *bytes <= threshold || self.delta_from(doc) != *from || sorted(heads) != *to {
             return None;
         }
         let from = from.clone();
@@ -827,7 +848,7 @@ mod tests {
         a.sending = sending();
         assert!(a.reply.is_none());
         let now = Instant::now();
-        a.withdraw_for_delta(now);
+        a.withdraw_for_delta(&Document::new(), now);
         assert!(a.delta_offered);
         assert!(
             a.take_overdue_reply(now + Duration::from_secs(3600))
@@ -849,24 +870,32 @@ mod tests {
         a.state.their_heads = Some(Vec::new());
         let now = Instant::now();
         assert_eq!(
-            a.offer_again(&heads, 1000, now),
+            a.offer_again(&ours, &heads, 1000, now),
             None,
             "nothing offered yet"
         );
 
         a.offered(&[], &heads, 5000);
-        assert_eq!(a.offer_again(&heads, 1000, now), Some(Vec::new()));
+        assert_eq!(a.offer_again(&ours, &heads, 1000, now), Some(Vec::new()));
         assert!(a.delta_offered && a.reply.is_some() && a.sending.is_some());
-        assert_eq!(a.offer_again(&heads, 5000, now), None, "it would fit now");
+        assert_eq!(
+            a.offer_again(&ours, &heads, 5000, now),
+            None,
+            "it would fit now"
+        );
 
         change(&mut ours, |doc| doc.put(ROOT, "k", 2).unwrap());
         assert_eq!(
-            a.offer_again(&ours.get_heads(), 1000, now),
+            a.offer_again(&ours, &ours.get_heads(), 1000, now),
             None,
             "ours moved"
         );
         a.state.their_heads = Some(heads.clone());
-        assert_eq!(a.offer_again(&heads, 1000, now), None, "theirs moved");
+        assert_eq!(
+            a.offer_again(&ours, &heads, 1000, now),
+            None,
+            "theirs moved"
+        );
     }
 
     /// A message too large to send is taken back: the peer, after the offer and an
@@ -900,7 +929,7 @@ mod tests {
         let carried = carrying.0.len();
         assert!(carried > 4_000, "the changes took {carried} bytes");
 
-        let from = a.withdraw_for_delta(now);
+        let from = a.withdraw_for_delta(&ours, now);
         assert!(from.is_empty());
         assert!(a.delta_offered);
         assert!(!a.send_succeeded(a.id()));
@@ -1157,6 +1186,30 @@ mod tests {
         }
     }
 
+    /// A delta starts at the peer's heads where we hold them all; where it has one we
+    /// lack, at those we hold and the heads we share, so every server of the delta
+    /// holds where it starts.
+    #[test]
+    fn a_delta_starts_where_we_hold_its_start() {
+        use crate::document::testing::change;
+        use automerge::{ROOT, transaction::Transactable};
+        let mut doc = Document::new();
+        change(&mut doc, |doc| doc.put(ROOT, "k", 1).unwrap());
+        let shared = doc.get_heads();
+        change(&mut doc, |doc| doc.put(ROOT, "k", 2).unwrap());
+        let known = doc.get_heads();
+        let unknown = crate::document::ChangeHash([7; 32]);
+        let mut a = session();
+        a.state.shared_heads = shared.clone();
+
+        a.state.their_heads = Some(known.clone());
+        assert_eq!(a.delta_from(&doc), known);
+        a.state.their_heads = Some(vec![known[0], unknown]);
+        let mut expected = vec![known[0], shared[0]];
+        expected.sort_unstable();
+        assert_eq!(a.delta_from(&doc), expected);
+    }
+
     /// A delta offered from heads we since moved past, by changes from the offerer,
     /// would bring those again; our own, which it lacks, don't count.
     #[test]
@@ -1228,7 +1281,7 @@ mod tests {
         b.send_succeeded(b.id());
         a.receive(&mut ours, decode(&opening).unwrap()).unwrap();
         assert!(a.generate(&ours, now, bytes).is_some());
-        a.withdraw_for_delta(now);
+        a.withdraw_for_delta(&ours, now);
         a.send_succeeded(a.id());
 
         // b loads the delta and tells a its heads, which are a's now.
diff --git a/src/swarm/sync.rs b/src/swarm/sync.rs
index 3cbead1..2a08aae 100644
--- a/src/swarm/sync.rs
+++ b/src/swarm/sync.rs
@@ -162,6 +162,7 @@ async fn deliver_sync_to_peer(state: &AppState, peer: &str) -> Result<()> {
                     // (it didn't answer in time): offered again without generating the
                     // message it replaces, which may take seconds when large.
                     let again = peer_state.sync.offer_again(
+                        &replica.doc,
                         replica.heads(),
                         delta_threshold,
                         Instant::now(),
@@ -186,7 +187,9 @@ async fn deliver_sync_to_peer(state: &AppState, peer: &str) -> Result<()> {
                             // Too large for one message: all or nothing, and on a slow link
                             // within the transport's delivery limit. The peer fetches it in
                             // chunks.
-                            let from = peer_state.sync.withdraw_for_delta(Instant::now());
+                            let from = peer_state
+                                .sync
+                                .withdraw_for_delta(&replica.doc, Instant::now());
                             let spec = transport::DeltaSpec::new(from, replica.heads().to_vec());
                             if spec.to.iter().all(|head| spec.from.contains(head)) {
                                 // It has our heads: nothing to offer (Automerge thought it
diff --git a/src/transport/mod.rs b/src/transport/mod.rs
index 4ddfe32..0f009cb 100644
--- a/src/transport/mod.rs
+++ b/src/transport/mod.rs
@@ -690,8 +690,8 @@ pub(crate) enum SyncBody {
 }
 
 /// The changes in the history of `to` but not of `from`: what a peer at `from` lacks of
-/// a document at `to`. Every node holding `to` produces the same bytes for it
-/// (`document::changes::delta_bytes`).
+/// a document at `to`. Every node holding `to` and `from` produces the same bytes for
+/// it; others serve none (`document::changes::delta_bytes`).
 #[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
 pub(crate) struct DeltaSpec {
     /// Sorted.