diff options
| -rw-r--r-- | docs/peers.md | 3 | ||||
| -rw-r--r-- | src/document/changes.rs | 43 | ||||
| -rw-r--r-- | src/document/sync.rs | 5 | ||||
| -rw-r--r-- | src/state/peer.rs | 4 | ||||
| -rw-r--r-- | src/swarm/relay.rs | 40 | ||||
| -rw-r--r-- | src/swarm/sync.rs | 3 |
6 files changed, 83 insertions, 15 deletions
diff --git a/docs/peers.md b/docs/peers.md index 9ef570b..0f7635e 100644 --- a/docs/peers.md +++ b/docs/peers.md @@ -299,7 +299,8 @@ has the changes already sends nothing back, even if its heads differ, as they do writes go on and newer changes are still on their way to it; nor is a neighbour told whose latest heads, as its own announcement or STATUS brought them, are this node's: announcements wait 200 to 400 ms, at random, so of two neighbours with the same news one is usually first, and -the other skips it), and a neighbour that lacks those heads pulls them (it sets them as the peer's heads in its sync state, so Automerge asks +the other skips it; nor one whose latest heads, as it said them in a STATUS or a sync +message, have all of this node's in their history: it is as far or further), and a neighbour that lacks those heads pulls them (it sets them as the peer's heads in its sync state, so Automerge asks for what's missing), if it still lacks them after a round trip to the announcer (at most 1 s): while writes go on, the newest are usually on their way to it along a writer's tree, and would otherwise come twice. It pulls from one neighbour at a time diff --git a/src/document/changes.rs b/src/document/changes.rs index 19d4db6..1091261 100644 --- a/src/document/changes.rs +++ b/src/document/changes.rs @@ -8,6 +8,25 @@ use anyhow::{Context, Result, anyhow, bail}; use super::{Change, ChangeHash, Document, ReadDoc}; +/// Whether the history of `theirs` (of them, those `doc` has) holds all of `ours` +/// (heads of `doc`): a peer at `theirs` has every change we have. +pub(crate) fn covers(doc: &Document, theirs: &[ChangeHash], ours: &[ChangeHash]) -> bool { + let known: Vec<ChangeHash> = theirs + .iter() + .copied() + .filter(|head| doc.get_change_by_hash(head).is_some()) + .collect(); + if known.is_empty() { + return ours.is_empty(); + } + let beyond: std::collections::HashSet<ChangeHash> = doc + .get_changes_meta(&known) + .into_iter() + .map(|change| change.hash) + .collect(); + ours.iter().all(|head| !beyond.contains(head)) +} + /// Those of `heads` that `doc` doesn't have. pub(crate) fn missing(doc: &impl ReadDoc, heads: &[ChangeHash]) -> Vec<ChangeHash> { heads @@ -176,6 +195,30 @@ mod tests { }); } + /// Heads cover ours when their history holds all of ours: as far as ours, or + /// further; not when behind or beside; heads we don't know count for nothing. + #[test] + fn heads_that_cover_ours() { + let mut doc = Document::new(); + change(&mut doc, |doc| doc.put(ROOT, "a", 1_i64).unwrap()); + let early = doc.get_heads(); + change(&mut doc, |doc| doc.put(ROOT, "b", 2_i64).unwrap()); + let ours = doc.get_heads(); + let mut further = doc.fork(); + change(&mut further, |doc| doc.put(ROOT, "c", 3_i64).unwrap()); + doc.merge(&mut further.clone()).unwrap(); + let unknown = ChangeHash([7; 32]); + + assert!(covers(&doc, &ours, &ours)); + assert!(covers(&doc, &further.get_heads(), &ours), "further"); + assert!(!covers(&doc, &early, &ours), "behind"); + assert!(!covers(&doc, &[unknown], &ours), "unknown"); + assert!( + covers(&doc, &[unknown, ours[0]], &ours), + "the known one does" + ); + } + #[test] fn no_delta_without_its_heads() { let mut doc = Document::new(); diff --git a/src/document/sync.rs b/src/document/sync.rs index 8c04632..e3ff157 100644 --- a/src/document/sync.rs +++ b/src/document/sync.rs @@ -28,6 +28,11 @@ impl Received { !self.0.changes.is_empty() } + /// The sender's heads, as it says. + pub(crate) fn heads(&self) -> &[crate::document::ChangeHash] { + &self.0.heads + } + /// The bytes of the changes it carries, and of its bloom filters (`have`). pub(crate) fn sizes(&self) -> (usize, usize) { let changes = self.0.changes.iter().map(<[u8]>::len).sum(); diff --git a/src/state/peer.rs b/src/state/peer.rs index a84e069..231f320 100644 --- a/src/state/peer.rs +++ b/src/state/peer.rs @@ -150,8 +150,8 @@ pub(crate) struct PeerState { /// How many heads its most recent keepalive listed (None if it sent a digest). pub(crate) reported_heads: Option<usize>, /// The heads its most recent STATUS or STATUS_REPLY carried (an announcement's too), - /// with its epoch: once they are ours, it has every change we would announce to it - /// (`swarm::relay`). + /// or its latest sync message, with its epoch: what it said it has. Once those cover + /// ours, it has every change we would announce to it (`swarm::relay`). pub(crate) last_heads: Option<(transport::Epoch, transport::HeadsSummary)>, /// How long its sync messages took to come (`swarm::hop`). pub(crate) hop_delays: crate::swarm::hop::HopDelays, diff --git a/src/swarm/relay.rs b/src/swarm/relay.rs index e049846..31df0d3 100644 --- a/src/swarm/relay.rs +++ b/src/swarm/relay.rs @@ -251,8 +251,11 @@ async fn announce_now(state: &AppState, pending: &Pending, slow: bool) { /// The neighbours to tell of the changes `senders` sent us. The peer whose changes we /// announce has them already, so with a single sender it is skipped; with several, each /// lacks the others' changes. So is a neighbour whose latest heads (its STATUS, an -/// announcement of its own, or its STATUS_REPLY) are ours now: it has all we have; and -/// one `sent` the changes along a writer's tree, which has them already. +/// announcement of its own, its STATUS_REPLY, or its latest sync message) are ours, or +/// have all of ours in their history: it has all we have; and one `sent` the changes +/// along a writer's tree, which has them already. Only what the neighbour said counts, +/// not what we sent it: a message along a tree may be lost, which is what the slow +/// round is for. async fn unreached( state: &AppState, senders: &HashSet<String>, @@ -262,11 +265,13 @@ async fn unreached( 1 => senders.iter().next().cloned(), _ => None, }; - let ours = { + let (epoch, heads) = { let stamp = state.stamp.borrow(); - (stamp.epoch, HeadsSummary::of(stamp.heads.to_vec())) + (stamp.epoch, stamp.heads.to_vec()) }; - let mut unreached = Vec::new(); + let ours = HeadsSummary::of(heads.clone()); + // What each neighbour said it has. + let mut said = Vec::new(); for (peer, peer_state) in state.peers_snapshot().await { if skip.as_deref() == Some(peer.as_str()) || sent.contains(&peer) @@ -274,17 +279,28 @@ async fn unreached( { continue; } - let has_ours = peer_state + let last = peer_state .lock() .await .last_heads - .as_ref() - .is_some_and(|(epoch, heads)| *epoch == ours.0 && *heads == ours.1); - if !has_ours { - unreached.push(peer); - } + .clone() + .filter(|(theirs, _)| *theirs == epoch) + .map(|(_, heads)| heads); + said.push((peer, last)); + } + let replica = state.doc.read().await; + if replica.epoch != epoch { + return said.into_iter().map(|(peer, _)| peer).collect(); } - unreached + said.into_iter() + .filter(|(_, last)| { + !last.as_ref().is_some_and(|theirs| { + *theirs == ours + || crate::document::changes::covers(&replica.doc, theirs.listed(), &heads) + }) + }) + .map(|(peer, _)| peer) + .collect() } /// Pulls in progress: one neighbour at a time, so each change arrives about once. The diff --git a/src/swarm/sync.rs b/src/swarm/sync.rs index ed4c4da..8660742 100644 --- a/src/swarm/sync.rs +++ b/src/swarm/sync.rs @@ -484,6 +484,9 @@ pub(crate) async fn handle_sync( let mut peer_state = peer_state_mutex.blocking_lock(); peer_state.reports_ready = Some(from_ready); + // What it says it has (`swarm::relay::unreached`). + peer_state.last_heads = + Some((epoch, transport::HeadsSummary::of(message.heads().to_vec()))); let heads_before = replica.heads().to_vec(); let sync = &mut peer_state.sync; |