aboutsummaryrefslogtreecommitdiffstats
path: root/src/swarm
diff options
context:
space:
mode:
Diffstat (limited to 'src/swarm')
-rw-r--r--src/swarm/relay.rs14
-rw-r--r--src/swarm/sync.rs21
-rw-r--r--src/swarm/tree.rs35
3 files changed, 54 insertions, 16 deletions
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 {