aboutsummaryrefslogtreecommitdiffstats
path: root/src
diff options
context:
space:
mode:
Diffstat (limited to 'src')
-rw-r--r--src/document/changes.rs25
-rw-r--r--src/document/sync.rs36
-rw-r--r--src/swarm/relay.rs14
-rw-r--r--src/swarm/sync.rs21
-rw-r--r--src/swarm/tree.rs35
5 files changed, 106 insertions, 25 deletions
diff --git a/src/document/changes.rs b/src/document/changes.rs
index 0409c50..19d4db6 100644
--- a/src/document/changes.rs
+++ b/src/document/changes.rs
@@ -28,10 +28,13 @@ pub(crate) fn commit_times(doc: &Document, since: &[ChangeHash]) -> Vec<i64> {
}
/// Who wrote the changes after `since`: the authors their writers set (a node's ID,
-/// `replica::authored`), each once; and whether any change has none (a snapshot, or a
-/// change of a node that set none).
-pub(crate) fn authors_since(doc: &Document, since: &[ChangeHash]) -> (Vec<String>, bool) {
- let mut authors: Vec<String> = Vec::new();
+/// `replica::authored`), each once, with the hashes of their changes; and whether any
+/// change has none (a snapshot, or a change of a node that set none).
+pub(crate) fn authors_since(
+ doc: &Document,
+ since: &[ChangeHash],
+) -> (Vec<(String, Vec<ChangeHash>)>, bool) {
+ let mut authors: Vec<(String, Vec<ChangeHash>)> = Vec::new();
let mut unknown = false;
for change in doc.get_changes_meta(since) {
match change
@@ -39,10 +42,10 @@ pub(crate) fn authors_since(doc: &Document, since: &[ChangeHash]) -> (Vec<String
.as_ref()
.and_then(|author| std::str::from_utf8(author.as_bytes()).ok())
{
- Some(author) if !authors.iter().any(|known| known == author) => {
- authors.push(author.to_owned());
- }
- Some(_) => {}
+ Some(author) => match authors.iter_mut().find(|(known, _)| known == author) {
+ Some((_, hashes)) => hashes.push(change.hash),
+ None => authors.push((author.to_owned(), vec![change.hash])),
+ },
None => unknown = true,
}
}
@@ -221,7 +224,11 @@ mod tests {
merged.merge(&mut b).unwrap();
let (mut authors, unknown) = authors_since(&merged, &since);
authors.sort();
- assert_eq!(authors, ["node-a", "node-b"]);
+ let counts: Vec<(&str, usize)> = authors
+ .iter()
+ .map(|(author, hashes)| (author.as_str(), hashes.len()))
+ .collect();
+ assert_eq!(counts, [("node-a", 2), ("node-b", 1)]);
assert!(!unknown);
// A change no node authored (a snapshot's, say).
diff --git a/src/document/sync.rs b/src/document/sync.rs
index 1f61f70..befb8e5 100644
--- a/src/document/sync.rs
+++ b/src/document/sync.rs
@@ -476,6 +476,13 @@ impl SyncSession {
known.len()
}
+ /// 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());
+ }
+
/// What the sync state knows of the peer (`PeerView`).
fn peer_view(&self) -> PeerView {
PeerView {
@@ -1238,6 +1245,35 @@ mod tests {
assert_eq!(b.known_beyond(&theirs, &theirs.get_heads()), 0);
}
+ /// Changes the peer gets from elsewhere (another parent in their writer's tree)
+ /// don't go to it, but do if it asks.
+ #[test]
+ fn changes_known_to_the_peer_go_only_when_asked() {
+ 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());
+ let message = a.generate(&ours, now, bytes).expect("our heads moved");
+ a.send_succeeded(a.id());
+ let (carried, _) = decode(&message).unwrap().sizes();
+ assert_eq!(carried, 0, "not sent: the peer gets it from elsewhere");
+
+ // It didn't: it asks, and gets it.
+ b.receive(&mut theirs, decode(&message).unwrap()).unwrap();
+ let asking = b.generate(&theirs, now, bytes).expect("it asks");
+ b.send_succeeded(b.id());
+ a.receive(&mut ours, decode(&asking).unwrap()).unwrap();
+ a.resend_if_needed(&ours, "b");
+ let answer = a.generate(&ours, now, bytes).expect("answered");
+ b.receive(&mut theirs, decode(&answer).unwrap()).unwrap();
+ assert_eq!(theirs.get_heads(), ours.get_heads());
+ }
+
/// 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/relay.rs b/src/swarm/relay.rs
index 58196d3..e049846 100644
--- a/src/swarm/relay.rs
+++ b/src/swarm/relay.rs
@@ -140,18 +140,24 @@ pub(crate) struct Pending {
}
/// After changes from `from` were applied and our heads changed: passed on at once along
-/// the trees of their `writers` (by their authors' pair keys); then announced, in the
+/// the trees of their `writers` (by their authors' pair keys, with their changes);
+/// then announced, in the
/// next round to those the trees leave out, and to the others in the slow round. To all
/// in the next round if any change has no tree to go along: none known whose
/// (`unauthored`), a writer not in the member list, or we have no part in its tree.
-pub(crate) async fn announce(state: &AppState, from: &str, writers: &[u64], unauthored: bool) {
+pub(crate) async fn announce(
+ state: &AppState,
+ from: &str,
+ writers: &[(u64, Vec<crate::document::ChangeHash>)],
+ unauthored: bool,
+) {
if !state.initial_sync_done.load(Ordering::Acquire) {
return;
}
let mut plans = Vec::new();
let mut treeless = unauthored || writers.is_empty();
- for writer in writers {
- match crate::swarm::tree::forward(state, *writer, from).await {
+ for (writer, changes) in writers {
+ match crate::swarm::tree::forward(state, *writer, from, changes).await {
Some(plan) => plans.push(plan),
None => treeless = true,
}
diff --git a/src/swarm/sync.rs b/src/swarm/sync.rs
index 9a63c8b..c76c16b 100644
--- a/src/swarm/sync.rs
+++ b/src/swarm/sync.rs
@@ -446,10 +446,15 @@ pub(crate) async fn handle_sync(
let clock = Arc::clone(&state.clock);
let ahead_ms = state.clock_ahead_ms.load(Ordering::Relaxed);
- // Some((caught_up, heads changed, carried changes, the new changes' authors and
- // whether any has none)) when applied, None when the message belongs to another
- // epoch.
- type Applied = (bool, bool, bool, (Vec<String>, bool));
+ // Some((caught_up, heads changed, carried changes, the new changes' authors with
+ // theirs, and whether any has none)) when applied, None when the message belongs
+ // to another epoch.
+ type Applied = (
+ bool,
+ bool,
+ bool,
+ (Vec<(String, Vec<crate::document::ChangeHash>)>, bool),
+ );
tokio::task::spawn_blocking(move || -> Result<Option<Applied>> {
// Before the lock: decoding a large message takes a while.
let payload_len = message.0.len();
@@ -572,9 +577,9 @@ pub(crate) async fn handle_sync(
crate::swarm::relay::pulled(state, peer);
// Caught up: our other neighbours may lack what the source had.
if changed || just_caught_up {
- let writers: Vec<u64> = authors
- .iter()
- .map(|author| crate::swarm::topology::pair_key(author))
+ let writers: Vec<(u64, Vec<crate::document::ChangeHash>)> = authors
+ .into_iter()
+ .map(|(author, changes)| (crate::swarm::topology::pair_key(&author), changes))
.collect();
crate::swarm::relay::announce(state, peer, &writers, unauthored).await;
} else if carried
@@ -584,7 +589,7 @@ pub(crate) async fn handle_sync(
// A writer's changes along its tree that we had already: they came another way
// first, not saying whose (with an answer to one of our messages, say), so we
// passed nothing on then. Our children in its tree may lack them still.
- crate::swarm::tree::forward(state, writer, peer).await;
+ crate::swarm::tree::forward(state, writer, peer, &[]).await;
}
state
diff --git a/src/swarm/tree.rs b/src/swarm/tree.rs
index 015d62c..20df2ec 100644
--- a/src/swarm/tree.rs
+++ b/src/swarm/tree.rs
@@ -284,6 +284,9 @@ pub(crate) struct Plan {
/// Our neighbours that no member they reach comes before (the tree reaches them only
/// by announcement), and those not in the member list's graph (whom no tree has).
pub(crate) uncovered: Vec<String>,
+ /// Our neighbours that get the writer's changes from another parent, and the writer
+ /// itself: they have them, or will, without us (`forward`).
+ pub(crate) covered: Vec<String>,
}
/// Our part in `writer`'s tree (`children`, `Plan`); None while the member list isn't
@@ -305,11 +308,13 @@ pub(crate) async fn plan(state: &AppState, writer: u64) -> Option<Plan> {
let mut children = Vec::new();
let mut alone = Vec::new();
let mut uncovered = Vec::new();
+ let mut covered = Vec::new();
for member in graph.get(me).into_iter().flatten() {
let Some(its_rank) = rank(&graph, &distance, &reached, member) else {
continue;
};
if distance[member.as_str()] == 0 {
+ covered.push(member.clone());
continue;
}
match parent(&graph, &distance, &reached, member) {
@@ -325,6 +330,8 @@ pub(crate) async fn plan(state: &AppState, writer: u64) -> Option<Plan> {
}
children.push(member.clone());
}
+ Some(parent) if parent != me => covered.push(member.clone()),
+ // Ours by the graph, but we don't reach it, or it comes before us.
Some(_) => {}
}
}
@@ -341,14 +348,34 @@ pub(crate) async fn plan(state: &AppState, writer: u64) -> Option<Plan> {
children,
alone,
uncovered,
+ covered,
})
}
-/// We applied a change of `writer`'s that `from` passed on: to our children in its
-/// tree (but `from`), a sync message each, saying whose. Returns our part in the tree
-/// (`Plan`), None if we have none (announcements then do it all).
-pub(crate) async fn forward(state: &AppState, writer: u64, from: &str) -> Option<Plan> {
+/// We applied `changes` of `writer`'s that `from` passed on (none if we had them
+/// already): to our children in its tree (but `from`), a sync message each, saying
+/// whose. Returns our part in the tree (`Plan`), None if we have none (announcements
+/// then do it all).
+///
+/// Our neighbours that get them from another parent (`Plan::covered`) count as having
+/// 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`).
+pub(crate) async fn forward(
+ state: &AppState,
+ writer: u64,
+ from: &str,
+ changes: &[crate::document::ChangeHash],
+) -> Option<Plan> {
let mut plan = plan(state, writer).await?;
+ 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);
+ }
+ }
+ }
plan.children.retain(|child| child != from);
let children = &plan.children;
for child in children {