summaryrefslogtreecommitdiffstats
path: root/src/state/session.rs
diff options
context:
space:
mode:
Diffstat (limited to 'src/state/session.rs')
-rw-r--r--src/state/session.rs57
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));
+ }
}