1 files changed, 68 insertions, 15 deletions

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.