fix deadlock in delta handshake

1 files changed, 57 insertions, 0 deletions

diff --git a/src/state/session.rs b/src/state/session.rs
index cd1ab30..f09ae41 100644
--- a/src/state/session.rs
+++ b/src/state/session.rs
@@ -57,6 +57,13 @@ pub(crate) struct SyncSession {
     /// (`withdraw_for_delta`): no more messages until it answers, or the answer is
     /// overdue (then the offer goes out again).
     pub(crate) delta_offered: bool,
+    /// What the last `generate` changed in the sync state besides the hashes sent: the
+    /// heads it last sent, and whether it had answered yet. Restored when the message is
+    /// withdrawn for a delta: otherwise Automerge believes the peer heard our heads, and
+    /// once the peer has the delta and says its heads are ours, never answers it, so the
+    /// peer never learns our heads (a joining node then never finishes its initial
+    /// sync).
+    before_generate: (Vec<automerge::ChangeHash>, bool),
 
     /// The answer we expect to our last sync message(s), if any.
     pub(crate) reply: Option<SyncReplyWait>,
@@ -87,6 +94,7 @@ impl SyncSession {
             retry_at: None,
             send_backoff: Backoff::new(SEND_RETRY, SEND_RETRY_MAX),
             delta_offered: false,
+            before_generate: (Vec::new(), false),
             reply: None,
             reply_backoff,
         }
@@ -110,6 +118,10 @@ impl SyncSession {
         bytes: SyncBytes,
     ) -> Option<sync::Message> {
         let sent_before = self.state.sent_hashes.clone();
+        self.before_generate = (
+            self.state.last_sent_heads.clone(),
+            self.state.have_responded,
+        );
         let message = doc.sync().generate_sync_message(&mut self.state);
         let in_step = self.state.their_heads.as_ref() == Some(&doc.get_heads());
         let sent: Vec<_> = self
@@ -161,6 +173,7 @@ 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()
     }
@@ -399,4 +412,48 @@ mod tests {
         let carried_again = again.encode().len();
         assert!(carried_again > 4_000, "sent again in {carried_again} bytes");
     }
+
+    /// A joining node opens, the answer is too large and becomes a delta, the node gets
+    /// the delta and says so: the answer to that carries our heads, so the node knows it
+    /// has everything (`have_all_theirs`).
+    #[test]
+    fn after_a_delta_the_peer_learns_our_heads() {
+        use automerge::{ROOT, transaction::Transactable};
+
+        let mut rng = crate::testrand::Rng::new(2);
+        let big: String = (0..1024)
+            .map(|_| format!("{:016x}", rng.next_u64()))
+            .collect();
+        let mut ours = AutoCommit::new();
+        ours.put(ROOT, "big", big).unwrap();
+        ours.commit();
+        let mut theirs = AutoCommit::new();
+        let (mut a, mut b) = (session(), session());
+        let now = Instant::now();
+        let bytes = SyncBytes::default();
+
+        // b (joining) opens; a's answer would carry everything: withdrawn for a delta.
+        let opening = b.generate(&mut theirs, now, bytes).unwrap();
+        b.send_succeeded(b.id());
+        a.receive(&mut ours, opening).unwrap();
+        assert!(a.generate(&mut ours, now, bytes).is_some());
+        a.withdraw_for_delta();
+        a.send_succeeded(a.id());
+
+        // b loads the delta and tells a its heads, which are a's now.
+        theirs.merge(&mut ours.clone()).unwrap();
+        assert!(
+            !b.have_all_theirs(&mut theirs),
+            "b hasn't heard a's heads yet"
+        );
+        let after_delta = b.generate(&mut theirs, now, bytes).unwrap();
+        b.send_succeeded(b.id());
+        a.receive(&mut ours, after_delta).unwrap();
+
+        let answer = a
+            .generate(&mut ours, now, bytes)
+            .expect("a answers with its heads");
+        b.receive(&mut theirs, answer).unwrap();
+        assert!(b.have_all_theirs(&mut theirs));
+    }
 }