diff options
Diffstat (limited to 'src/swarm')
| -rw-r--r-- | src/swarm/mod.rs | 1 | ||||
| -rw-r--r-- | src/swarm/peer.rs | 1 | ||||
| -rw-r--r-- | src/swarm/relay.rs | 126 | ||||
| -rw-r--r-- | src/swarm/sync.rs | 19 | ||||
| -rw-r--r-- | src/swarm/topology.rs | 17 |
5 files changed, 159 insertions, 5 deletions
diff --git a/src/swarm/mod.rs b/src/swarm/mod.rs index 0997f7c..5616fe8 100644 --- a/src/swarm/mod.rs +++ b/src/swarm/mod.rs @@ -6,6 +6,7 @@ pub(crate) mod bootstrap; pub(crate) mod members; pub(crate) mod pair; pub(crate) mod peer; +pub(crate) mod relay; pub(crate) mod source; pub(crate) mod sync; pub(crate) mod topology; diff --git a/src/swarm/peer.rs b/src/swarm/peer.rs index a42f5a4..2fd248e 100644 --- a/src/swarm/peer.rs +++ b/src/swarm/peer.rs @@ -133,6 +133,7 @@ async fn note_keepalive(state: &AppState, peer: &str, remote: &transport::Keepal peer_state.syncs_with_us = Some(remote.syncs_with_you); peer_state.reported_heads = remote.heads.heads().map(<[_]>::len); drop(peer_state); + crate::swarm::relay::pull_if_behind(state, peer, remote).await; if became_ready { // We may have left the opening to it while it wasn't ready, and nothing else // would make us look again. diff --git a/src/swarm/relay.rs b/src/swarm/relay.rs new file mode 100644 index 0000000..a0fa6e1 --- /dev/null +++ b/src/swarm/relay.rs @@ -0,0 +1,126 @@ +//! Passing changes on in a partial mesh (`swarm::topology`). +//! +//! A node pushes its own changes to its neighbours, but nothing it receives goes any +//! further by itself. So after applying changes from one neighbour, a node *announces* +//! its new heads to its other neighbours with a PING (a keepalive: heads, about 100 +//! bytes, not the changes). A neighbour that sees heads it lacks *pulls*: it enters the +//! announced heads as the peer's heads in its sync state, Automerge asks for what's +//! missing (`need`), and the announcer answers with exactly those changes. Each node +//! receives a change about once; the rest is announcements. Regular keepalives work the +//! same way, so any gap is noticed within a keepalive interval. +//! +//! In a full mesh the writer reaches everyone itself, so nothing is announced. + +use std::sync::atomic::Ordering; +use std::time::Duration; + +use tokio::time::Instant; +use tracing::debug; + +use crate::state::{AppState, Standing}; +use crate::swarm::peer::local_keepalive; +use crate::swarm::sync::SyncEvent; +use crate::swarm::topology; +use crate::transport::{self, HeadsSummary, ParsedMessage}; + +/// A pull that brought nothing for this long no longer blocks pulling from another peer. +const PULL_TIMEOUT: Duration = Duration::from_secs(15); + +/// After changes from `from` were applied and our heads changed: tell our other +/// neighbours. +pub(crate) async fn announce(state: &AppState, from: &str) { + if !state.initial_sync_done.load(Ordering::Acquire) || topology::is_full_mesh(state) { + return; + } + let neighbours: Vec<String> = state + .peer_states + .read() + .await + .keys() + .filter(|peer| peer.as_str() != from && topology::is_neighbour(state, peer)) + .cloned() + .collect(); + for peer in neighbours { + let keepalive = local_keepalive(state, &peer).await; + if let Err(error) = state + .transport + .send_message(&peer, &ParsedMessage::Ping(keepalive)) + .await + { + debug!(peer = %peer, error = %format!("{error:#}"), "Announcing our heads failed"); + } + } +} + +/// A keepalive from `peer`: if it has heads we lack, pull them (one peer at a time). +pub(crate) async fn pull_if_behind(state: &AppState, peer: &str, remote: &transport::Keepalive) { + if remote.epoch != state.local_epoch() || !topology::is_neighbour(state, peer) { + return; + } + let missing = match &remote.heads { + HeadsSummary::Heads(heads) => { + // AutoCommit's lookups take &mut self. + let mut replica = state.doc.write().await; + let missing = heads + .iter() + .any(|head| replica.doc.get_change_by_hash(head).is_none()); + if !missing { + return; + } + Some(heads.clone()) + } + // Can't tell which heads are missing; the digest differs from ours, though. + HeadsSummary::Digest(_) => None, + }; + + let now = Instant::now(); + { + let mut pulling = state + .pulling + .lock() + .unwrap_or_else(|poisoned| poisoned.into_inner()); + let busy = matches!( + &*pulling, + Some((other, since)) if other != peer && now < *since + PULL_TIMEOUT + ); + if busy { + return; + } + *pulling = Some((peer.to_owned(), now)); + } + + let Some(peer_state) = state.get_peer_state(peer).await else { + return; + }; + { + let mut peer_state = peer_state.lock().await; + if !matches!(peer_state.standing, Standing::Member { .. }) { + return; + } + match missing { + Some(heads) => { + // Automerge asks for whatever these heads need that we don't have. + peer_state.sync.state.their_heads = Some(heads); + peer_state.sync.state.in_flight = false; + } + // A fresh session's opening carries our heads and bloom filter; the answer + // brings what we lack. + None => peer_state.reset_sync(), + } + } + debug!(peer = %peer, "Neighbour has changes we lack; pulling"); + let _ = state + .sync_events + .send(SyncEvent::PeerNeedsSync(peer.to_owned())); +} + +/// A sync message from `peer` was applied: a pull from it is answered. +pub(crate) fn pulled(state: &AppState, peer: &str) { + let mut pulling = state + .pulling + .lock() + .unwrap_or_else(|poisoned| poisoned.into_inner()); + if pulling.as_ref().is_some_and(|(p, _)| p == peer) { + *pulling = None; + } +} diff --git a/src/swarm/sync.rs b/src/swarm/sync.rs index b8d16aa..383cdaf 100644 --- a/src/swarm/sync.rs +++ b/src/swarm/sync.rs @@ -310,8 +310,9 @@ pub(crate) async fn handle_sync( let message = envelope.message; let peer = peer.to_owned(); - // Some(caught_up) when applied, None when the message belongs to another epoch. - tokio::task::spawn_blocking(move || -> Result<Option<bool>> { + // Some((caught_up, heads changed)) when applied, None when the message belongs + // to another epoch. + tokio::task::spawn_blocking(move || -> Result<Option<(bool, bool)>> { let mut replica = doc.blocking_write(); // Authoritative check, under the same lock as the apply: the document may @@ -322,6 +323,7 @@ pub(crate) async fn handle_sync( let mut peer_state = peer_state_mutex.blocking_lock(); peer_state.reports_ready = Some(from_ready); + let heads_before = replica.doc.get_heads(); replica .doc @@ -332,8 +334,9 @@ pub(crate) async fn handle_sync( peer_state.sync.reply_settled(); resend_if_needed(&peer, &mut peer_state.sync.state); + let changed = replica.doc.get_heads() != heads_before; - Ok(Some( + Ok(Some(( // Only from a ready peer: a peer still catching up has empty heads, which // every document "contains". check_caught_up @@ -348,13 +351,14 @@ pub(crate) async fn handle_sync( .iter() .all(|h| replica.doc.get_change_by_hash(h).is_some()) }), - )) + changed, + ))) }) .await .context("Sync task panicked")?? }; - let Some(caught_up) = applied else { + let Some((caught_up, changed)) = applied else { debug!(peer = %peer, epoch = %epoch_label, "Dropped sync message from another epoch"); return Ok(()); }; @@ -368,6 +372,11 @@ pub(crate) async fn handle_sync( state.notify_store_changed(); + crate::swarm::relay::pulled(state, peer); + if changed { + crate::swarm::relay::announce(state, peer).await; + } + state .sync_events .send(SyncEvent::MessageReceived(peer.to_owned())) diff --git a/src/swarm/topology.rs b/src/swarm/topology.rs index b5880cf..dddf81b 100644 --- a/src/swarm/topology.rs +++ b/src/swarm/topology.rs @@ -33,6 +33,18 @@ pub(crate) struct NeighbourView { k: usize, unreachable: BTreeSet<String>, neighbours: BTreeSet<String>, + /// Every reachable member is a neighbour. + full_mesh: bool, +} + +/// Whether every reachable member is our neighbour (also while the view is unknown). +pub(crate) fn is_full_mesh(state: &AppState) -> bool { + state + .neighbours + .read() + .unwrap_or_else(|poisoned| poisoned.into_inner()) + .as_ref() + .is_none_or(|view| view.full_mesh) } /// Whether we initiate contact with `peer`: HELLOs, keepalives, dormancy checks, sync @@ -113,10 +125,15 @@ pub(crate) async fn refresh(state: &AppState) -> Option<NeighbourChange> { removed = change.removed.len(), "Neighbours computed from the member list" ); + let reachable_others = current + .iter() + .filter(|m| m.as_str() != state.local_id() && !unreachable.contains(*m)) + .count(); *state .neighbours .write() .unwrap_or_else(|poisoned| poisoned.into_inner()) = Some(NeighbourView { + full_mesh: computed.len() >= reachable_others, members: current, k, unreachable, |