diff options
Diffstat (limited to 'src')
| -rw-r--r-- | src/document/sync.rs | 63 | ||||
| -rw-r--r-- | src/swarm/tree.rs | 21 |
2 files changed, 78 insertions, 6 deletions
diff --git a/src/document/sync.rs b/src/document/sync.rs index befb8e5..82b37f5 100644 --- a/src/document/sync.rs +++ b/src/document/sync.rs @@ -120,6 +120,10 @@ pub(crate) struct SyncSession { /// peer never learns our heads (a joining node then never finishes its initial /// sync). before_generate: (Vec<crate::document::ChangeHash>, bool), + /// Changes counted as sent because another parent brings them (`known_to_peer`), + /// and when: still in flight perhaps, so they go along with a change of ours that + /// builds on them (`with_predecessors`). + known_elsewhere: std::collections::HashMap<crate::document::ChangeHash, Instant>, /// The answer we expect to our last sync message(s), if any. pub(crate) reply: Option<SyncReplyWait>, @@ -127,6 +131,10 @@ pub(crate) struct SyncSession { pub(crate) reply_backoff: Backoff, } +/// How long a change counted as sent because another parent brings it may still be in +/// flight to the peer (`SyncSession::with_predecessors`): a slow round trip, and more. +const KNOWN_ELSEWHERE_FOR: Duration = Duration::from_secs(2); + /// Sync messages we sent and expect an answer to. Automerge assumes reliable delivery: /// a lost message leaves both sides waiting (`in_flight`), and changes it carried are /// never sent again (`sent_hashes`). We resend after a timeout instead. @@ -156,6 +164,7 @@ impl SyncSession { last_offer: None, before_receive: None, before_generate: (Vec::new(), false), + known_elsewhere: std::collections::HashMap::new(), reply: None, reply_backoff, } @@ -479,8 +488,36 @@ impl SyncSession { /// The peer has `changes`, or will without us (another parent in their writer's /// tree brings them, `swarm::tree::forward`): counted as sent, they don't go to it, /// unless it asks for them (`resend_if_needed`). - pub(crate) fn known_to_peer(&mut self, changes: &[crate::document::ChangeHash]) { - self.state.sent_hashes.extend(changes.iter().copied()); + pub(crate) fn known_to_peer(&mut self, changes: &[crate::document::ChangeHash], now: Instant) { + self.known_elsewhere + .retain(|_, at| now.saturating_duration_since(*at) < KNOWN_ELSEWHERE_FOR); + for hash in changes { + if self.state.sent_hashes.insert(*hash) { + self.known_elsewhere.insert(*hash, now); + } + } + } + + /// We pass `changes` on to the peer: of their predecessors, those counted as sent + /// lately only because another parent brings them (`known_to_peer`) go along. That + /// one may not have arrived yet (with writers in turn, each change builds on the + /// last writer's, which comes along another tree): without them, the peer couldn't + /// apply ours, and would ask for them, a round trip later. + pub(crate) fn with_predecessors( + &mut self, + doc: &Document, + changes: &[crate::document::ChangeHash], + ) { + for hash in changes { + let Some(change) = doc.get_change_by_hash(hash) else { + continue; + }; + for dep in change.deps() { + if self.known_elsewhere.remove(dep).is_some() { + self.state.sent_hashes.remove(dep); + } + } + } } /// What the sync state knows of the peer (`PeerView`). @@ -1257,7 +1294,7 @@ mod tests { let (now, bytes) = (Instant::now(), SyncBytes::default()); change(&mut ours, |doc| doc.put(ROOT, "elsewhere", 2).unwrap()); - a.known_to_peer(&ours.get_heads()); + a.known_to_peer(&ours.get_heads(), Instant::now()); let message = a.generate(&ours, now, bytes).expect("our heads moved"); a.send_succeeded(a.id()); let (carried, _) = decode(&message).unwrap().sizes(); @@ -1274,6 +1311,26 @@ mod tests { assert_eq!(theirs.get_heads(), ours.get_heads()); } + /// A change counted as known to the peer lately goes along with one of ours that + /// builds on it: the peer applies both at once, without asking. + #[test] + fn predecessors_known_elsewhere_go_along() { + use crate::document::testing::change; + use automerge::{ROOT, transaction::Transactable}; + let (mut ours, mut theirs) = (Document::new(), Document::new()); + change(&mut ours, |doc| doc.put(ROOT, "early", 1).unwrap()); + let (mut a, mut b) = in_step(&mut ours, &mut theirs); + let (now, bytes) = (Instant::now(), SyncBytes::default()); + + change(&mut ours, |doc| doc.put(ROOT, "elsewhere", 2).unwrap()); + a.known_to_peer(&ours.get_heads(), now); + change(&mut ours, |doc| doc.put(ROOT, "along", 3).unwrap()); + a.with_predecessors(&ours, &ours.get_heads()); + let message = a.generate(&ours, now, bytes).expect("our heads moved"); + b.receive(&mut theirs, decode(&message).unwrap()).unwrap(); + assert_eq!(theirs.get_heads(), ours.get_heads(), "both in one message"); + } + /// Our transport refused a message taken as delivered: the peer's heads are as /// before, so the next message carries the change again. #[test] diff --git a/src/swarm/tree.rs b/src/swarm/tree.rs index 20df2ec..ffdf80e 100644 --- a/src/swarm/tree.rs +++ b/src/swarm/tree.rs @@ -361,7 +361,9 @@ pub(crate) async fn plan(state: &AppState, writer: u64) -> Option<Plan> { /// them (`SyncSession::known_to_peer`): else, with writers in turn, each change coming /// along another tree, the next message we pass on to one would carry all we got since /// our last to it, mostly what its other parents brought it. Should one lack them after -/// all, it asks, and gets them (`resend_if_needed`). +/// all, it asks, and gets them (`resend_if_needed`). But to our children, those of +/// them that `changes` build on go along, if they were counted so lately +/// (`SyncSession::with_predecessors`). pub(crate) async fn forward( state: &AppState, writer: u64, @@ -372,15 +374,28 @@ pub(crate) async fn forward( if !changes.is_empty() { for neighbour in &plan.covered { if let Some(peer_state) = state.get_peer_state(neighbour).await { - peer_state.lock().await.sync.known_to_peer(changes); + peer_state + .lock() + .await + .sync + .known_to_peer(changes, tokio::time::Instant::now()); } } } plan.children.retain(|child| child != from); let children = &plan.children; + let replica = if changes.is_empty() { + None + } else { + Some(state.doc.read().await) + }; for child in children { if let Some(peer_state) = state.get_peer_state(child).await { - peer_state.lock().await.tree_writer = Some(writer); + let mut peer_state = peer_state.lock().await; + peer_state.tree_writer = Some(writer); + if let Some(replica) = &replica { + peer_state.sync.with_predecessors(&replica.doc, changes); + } } state.stats.tree_forward(); state.request_sync(child); |