diff options
| -rw-r--r-- | docs/peers.md | 6 | ||||
| -rw-r--r-- | src/document/sync.rs | 63 | ||||
| -rw-r--r-- | src/swarm/tree.rs | 21 |
3 files changed, 83 insertions, 7 deletions
diff --git a/docs/peers.md b/docs/peers.md index 4ee3c16..9ef570b 100644 --- a/docs/peers.md +++ b/docs/peers.md @@ -361,7 +361,11 @@ doesn't carry them again. Without that, with writers in turn, each change coming another tree, the node's view of a neighbour stays as of its last message to it, and the next carries all it got since, nearly all of it brought by the neighbour's other parents (with 50 nodes and writers in turn, about ten changes per message, one of them -new). Should one lack them after all, it asks, as above. +new). Should one lack them after all, it asks, as above. But what the node passes on +to a child carries those of them the passed-on changes build on, if counted so within +the last 2 s: with writers in turn each change builds on the last writer's, which may +still be on its way along another tree, and the child would have to ask for it, a round +trip later. As nobody waits for these answers, messages overtake each other on slow links. Three things keep that from sending changes twice (`document::sync`): 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); |