diff options
Diffstat (limited to 'src/swarm')
| -rw-r--r-- | src/swarm/relay.rs | 35 | ||||
| -rw-r--r-- | src/swarm/sync.rs | 26 | ||||
| -rw-r--r-- | src/swarm/tree.rs | 12 |
3 files changed, 46 insertions, 27 deletions
diff --git a/src/swarm/relay.rs b/src/swarm/relay.rs index a8560a5..53c1065 100644 --- a/src/swarm/relay.rs +++ b/src/swarm/relay.rs @@ -77,30 +77,33 @@ pub(crate) struct Pending { } /// After changes from `from` were applied and our heads changed: passed on at once along -/// the tree of their `writer` if the message said whose; then announced, in the next -/// round to those the tree leaves out (to all if there's no tree), and to the others in -/// the slow round. -pub(crate) async fn announce(state: &AppState, from: &str, writer: Option<u64>) { +/// the trees of their `writers` (by their authors' pair keys); 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) { if !state.initial_sync_done.load(Ordering::Acquire) { return; } - let plan = match writer { - Some(writer) => crate::swarm::tree::forward(state, writer, from).await, - None => None, - }; - let tree = plan.is_some(); + 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 { + Some(plan) => plans.push(plan), + None => treeless = true, + } + } + let tree = !plans.is_empty(); let schedule = { let mut pending = state.announce_pending.lock_or_recover(); let schedule = pending.is_none(); let pending = pending.get_or_insert_with(Pending::default); pending.senders.insert(from.to_owned()); - match &plan { - Some(plan) => { - pending.sent.extend(plan.children.iter().cloned()); - pending.targets.extend(plan.uncovered.iter().cloned()); - } - None => pending.all = true, + for plan in &plans { + pending.sent.extend(plan.children.iter().cloned()); + pending.targets.extend(plan.uncovered.iter().cloned()); } + pending.all |= treeless; schedule }; if schedule { @@ -115,7 +118,7 @@ pub(crate) async fn announce(state: &AppState, from: &str, writer: Option<u64>) let slow = slow.get_or_insert_with(Pending::default); slow.senders.insert(from.to_owned()); slow.all = true; - if let Some(plan) = &plan { + for plan in &plans { slow.sent.extend(plan.children.iter().cloned()); } schedule diff --git a/src/swarm/sync.rs b/src/swarm/sync.rs index dd5444b..b20f7ef 100644 --- a/src/swarm/sync.rs +++ b/src/swarm/sync.rs @@ -376,9 +376,11 @@ 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)) when applied, None when the - // message belongs to another epoch. - tokio::task::spawn_blocking(move || -> Result<Option<(bool, bool, bool)>> { + // 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)); + tokio::task::spawn_blocking(move || -> Result<Option<Applied>> { // Before the lock: decoding a large message takes a while. let message = crate::document::sync::decode(&message)?; let carried = message.carries_changes(); @@ -411,6 +413,13 @@ pub(crate) async fn handle_sync( let heads_after = replica.heads().to_vec(); stats.heads_seen(heads_after.len()); let changed = heads_after != heads_before; + // Whose they are, to pass them on along their writers' trees, however they + // came (along a tree, with an answer, pulled). + let authors = if changed { + crate::document::changes::authors_since(&replica.doc, &heads_before) + } else { + (Vec::new(), false) + }; if changed && measure_propagation { // Now by the peers' clocks (their median): this node's own clock skew // doesn't count; the writer's still does. @@ -432,13 +441,14 @@ pub(crate) async fn handle_sync( check_caught_up && from_ready && peer_state.sync.have_all_theirs(&replica.doc), changed, carried, + authors, ))) }) .await .context("Sync task panicked")?? }; - let Some((caught_up, changed, carried)) = applied else { + let Some((caught_up, changed, carried, (authors, unauthored))) = applied else { debug!(peer = %peer, epoch = %epoch_label, "Dropped sync message from another epoch"); return Ok(()); }; @@ -460,7 +470,11 @@ 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 { - crate::swarm::relay::announce(state, peer, writer.filter(|_| changed)).await; + let writers: Vec<u64> = authors + .iter() + .map(|author| crate::swarm::topology::pair_key(author)) + .collect(); + crate::swarm::relay::announce(state, peer, &writers, unauthored).await; } else if carried && let Some(writer) = writer && state.initial_sync_done.load(Ordering::Acquire) @@ -496,7 +510,7 @@ fn fetch_delta(state: &AppState, peer: &str, epoch: transport::Epoch, spec: tran Ok(changed) => { state.notify_store_changed(); if changed { - crate::swarm::relay::announce(&state, &peer, None).await; + crate::swarm::relay::announce(&state, &peer, &[], true).await; } } Err(error) => { diff --git a/src/swarm/tree.rs b/src/swarm/tree.rs index 8031a69..8093c91 100644 --- a/src/swarm/tree.rs +++ b/src/swarm/tree.rs @@ -12,11 +12,13 @@ //! it reaches routes around links that don't work: where a writer behind NAT reaches a //! single member, that one comes first and the others take it. //! -//! A sync message that passes a writer's change on says whose (`SyncEnvelope::writer`). -//! Whoever applies one passes it on to its own children in that writer's tree -//! (`forward`), however it got it; so does the writer, to all its neighbours, with its -//! own writes. A member no member it reaches comes before (or whose parent didn't get -//! the change) gets it from the announcements (`swarm::relay`), as before. +//! Whose a change is, its author says (every node writes as itself: its node ID, by +//! pair key here): whoever applies changes passes them on to its own children in each of +//! their writers' trees (`forward`), however they came; so does the writer, to all its +//! neighbours, with its own writes. A sync message along a tree also says whose +//! (`SyncEnvelope::writer`), for one that had them already (`swarm::sync`). A member no +//! member it reaches comes before (or whose parent didn't get the change) gets it from +//! the announcements (`swarm::relay`), as before. //! //! The graph is computed once per member list (tens of milliseconds for thousands of //! members), the distances once per writer and graph (`MAX_WRITERS` kept). |