summaryrefslogtreecommitdiffstats
path: root/src/state
diff options
context:
space:
mode:
Diffstat (limited to 'src/state')
-rw-r--r--src/state/session.rs84
1 files changed, 84 insertions, 0 deletions
diff --git a/src/state/session.rs b/src/state/session.rs
index f09ae41..8885699 100644
--- a/src/state/session.rs
+++ b/src/state/session.rs
@@ -220,6 +220,13 @@ impl SyncSession {
);
self.state.sent_hashes.clear();
}
+ if answer_for_everything_since_shared(&mut self.state) {
+ debug!(
+ peer = %peer,
+ need = self.state.their_need.as_ref().map_or(0, Vec::len),
+ "Peer lacks the changes ours build on; sending everything since what we share"
+ );
+ }
}
/// For the simulator's dump: the peer's heads as Automerge knows them, and whether
@@ -329,10 +336,87 @@ impl SyncSession {
}
}
+/// A peer that got a change of ours without the changes it builds on (they were lost,
+/// a later one arrived) asks for those it lacks, and while it does, Automerge leaves
+/// out what it has (its bloom filter: `have` waits until all it needs is in our heads).
+/// Without a `have`, Automerge sends exactly what was asked for, and the peer, finding
+/// each change's own predecessors missing in turn, asks again: one round trip per
+/// missing change, hundreds after a short outage. So answer as if it had said it has
+/// nothing since the heads we share: everything since then goes in one message (a
+/// delta when large: `swarm::sync`), and what it had already only twice. Returns
+/// whether it did. Call after receiving.
+fn answer_for_everything_since_shared(state: &mut sync::State) -> bool {
+ let asks_without_saying = state.their_have.as_ref().is_some_and(Vec::is_empty)
+ && state
+ .their_need
+ .as_ref()
+ .is_some_and(|need| !need.is_empty());
+ if !asks_without_saying {
+ return false;
+ }
+ state.their_have = Some(vec![sync::Have {
+ last_sync: state.shared_heads.clone(),
+ bloom: sync::BloomFilter::default(),
+ }]);
+ // What we sent since may be what was lost.
+ state.sent_hashes.clear();
+ true
+}
+
#[cfg(test)]
mod tests {
use super::*;
+ /// After a loss, the changes the peer lacks go in one message, not one per round
+ /// trip (`answer_for_everything_since_shared`).
+ #[test]
+ fn missing_predecessors_go_at_once() {
+ use automerge::transaction::Transactable;
+ let mut a = AutoCommit::new();
+ let mut b = AutoCommit::new();
+ let (mut at_a, mut at_b) = (sync::State::new(), sync::State::new());
+ a.put(automerge::ROOT, "early", 1).unwrap();
+ // Until in step.
+ for _ in 0..10 {
+ let to_b = a.sync().generate_sync_message(&mut at_a);
+ let to_a = b.sync().generate_sync_message(&mut at_b);
+ if to_b.is_none() && to_a.is_none() {
+ break;
+ }
+ if let Some(m) = to_b {
+ b.sync().receive_sync_message(&mut at_b, m).unwrap();
+ }
+ if let Some(m) = to_a {
+ a.sync().receive_sync_message(&mut at_a, m).unwrap();
+ }
+ }
+ assert_eq!(a.get_heads(), b.get_heads());
+ // Five changes, whose message is lost; then a sixth, which arrives alone.
+ for i in 0..5 {
+ a.put(automerge::ROOT, format!("lost{i}"), i).unwrap();
+ a.commit();
+ }
+ let _lost = a.sync().generate_sync_message(&mut at_a).unwrap();
+ a.put(automerge::ROOT, "last", 6).unwrap();
+ a.commit();
+ at_a.in_flight = false; // As after our reply timeout.
+ let last = a.sync().generate_sync_message(&mut at_a).unwrap();
+ assert_eq!(last.changes.len(), 1);
+ b.sync().receive_sync_message(&mut at_b, last).unwrap();
+ // b asks for the fifth, saying nothing of what it has.
+ let ask = b.sync().generate_sync_message(&mut at_b).unwrap();
+ assert!(ask.have.is_empty() && ask.need.len() == 1);
+ a.sync().receive_sync_message(&mut at_a, ask).unwrap();
+ assert!(answer_for_everything_since_shared(&mut at_a));
+ // All since the shared heads, in one message (here the whole document: more
+ // than a third of it is missing).
+ let answer = a.sync().generate_sync_message(&mut at_a).unwrap();
+ b.sync().receive_sync_message(&mut at_b, answer).unwrap();
+ assert_eq!(b.get_heads(), a.get_heads());
+ // A message with what it has is left alone.
+ assert!(!answer_for_everything_since_shared(&mut at_b));
+ }
+
fn session() -> SyncSession {
SyncSession::new(Backoff::new(
Duration::from_secs(1),