diff options
Diffstat (limited to 'src/swarm')
| -rw-r--r-- | src/swarm/peer.rs | 65 | ||||
| -rw-r--r-- | src/swarm/source.rs | 169 |
2 files changed, 208 insertions, 26 deletions
diff --git a/src/swarm/peer.rs b/src/swarm/peer.rs index 1e8b91d..f98bf17 100644 --- a/src/swarm/peer.rs +++ b/src/swarm/peer.rs @@ -6,7 +6,6 @@ use std::time::Duration; use anyhow::{Context, Result, anyhow, bail}; use automerge::AutoCommit; -use sha2::{Digest, Sha256}; use tokio::sync::Mutex; use tokio::time::{Instant, MissedTickBehavior}; use tracing::{debug, info, instrument, trace, warn}; @@ -98,23 +97,40 @@ async fn apply_forget_records(state: &AppState) { // --- keepalive and self-healing ---------------------------------------------- -fn heads_digest(doc: &mut AutoCommit) -> transport::HeadsDigest { - let mut heads = doc.get_heads(); - heads.sort_unstable(); - let mut hasher = Sha256::new(); - for head in &heads { - hasher.update(head.0); +/// Our keepalive for `peer`. +pub(crate) async fn local_keepalive(state: &AppState, peer: &str) -> transport::Keepalive { + // AutoCommit::get_heads takes &mut self (it closes the pending transaction). + // Heads and epoch come from the same locked replica, so they always match. + let (heads, epoch) = { + let mut replica = state.doc.write().await; + ( + transport::HeadsSummary::of(replica.doc.get_heads()), + replica.epoch, + ) + }; + let syncs_with_you = match state.get_peer_state(peer).await { + Some(peer_state) => { + let peer_state = peer_state.lock().await; + peer_state.liveness == Liveness::Active + && matches!(peer_state.standing, Standing::Member { .. }) + && peer_state.epoch == Some(epoch) + } + None => false, + }; + transport::Keepalive { + heads, + epoch, + ready: state.is_ready().await, + syncs_with_you, } - hasher.finalize().into() } -pub(crate) async fn local_keepalive(state: &AppState) -> transport::Keepalive { - // AutoCommit::get_heads takes &mut self (it closes the pending transaction). - // Digest and epoch come from the same locked replica, so they always match. - let mut replica = state.doc.write().await; - transport::Keepalive { - heads_digest: heads_digest(&mut replica.doc), - epoch: replica.epoch, +/// What a peer's keepalive says about it. +async fn note_keepalive(state: &AppState, peer: &str, remote: &transport::Keepalive) { + if let Some(peer_state) = state.get_peer_state(peer).await { + let mut peer_state = peer_state.lock().await; + peer_state.reports_ready = Some(remote.ready); + peer_state.syncs_with_us = Some(remote.syncs_with_you); } } @@ -124,7 +140,7 @@ async fn heal_if_diverged( local: &transport::Keepalive, remote: &transport::Keepalive, ) { - // Digests of documents from different epochs are never equal and never converge. + // Heads of documents from different epochs are never equal and never converge. if local.epoch != remote.epoch { return; } @@ -135,18 +151,13 @@ async fn heal_if_diverged( let heal = { let mut peer_state = peer_state_mutex.lock().await; - if local.heads_digest == remote.heads_digest { + if local.heads == remote.heads { // Same document: whatever answer we were waiting for, there is nothing left // to send. (Automerge answers only when it has something to say, so a sync // message to a peer that already has everything gets no reply at all.) peer_state.sync.reply_settled(); } - peer_state.needs_heal( - &local.heads_digest, - &remote.heads_digest, - &state.timing, - Instant::now(), - ) + peer_state.needs_heal(&local.heads, &remote.heads, &state.timing, Instant::now()) }; if heal { @@ -432,7 +443,7 @@ pub(crate) async fn run_peer_worker(state: AppState) { } if ping_due { - let local = local_keepalive(&state).await; + let local = local_keepalive(&state, peer).await; if let Err(error) = state .transport .send_message(peer, &ParsedMessage::Ping(local)) @@ -747,7 +758,8 @@ pub(crate) async fn handle_message( if reactivated || peer_swarm(state, &peer).await.is_none() { ensure_handshake(state, &peer).await; } - let local = local_keepalive(state).await; + note_keepalive(state, &peer, &remote).await; + let local = local_keepalive(state, &peer).await; heal_if_diverged(state, &peer, &local, &remote).await; if let Err(error) = state .transport @@ -761,7 +773,8 @@ pub(crate) async fn handle_message( if reactivated || peer_swarm(state, &peer).await.is_none() { ensure_handshake(state, &peer).await; } - let local = local_keepalive(state).await; + note_keepalive(state, &peer, &remote).await; + let local = local_keepalive(state, &peer).await; heal_if_diverged(state, &peer, &local, &remote).await; } } diff --git a/src/swarm/source.rs b/src/swarm/source.rs new file mode 100644 index 0000000..6be62b7 --- /dev/null +++ b/src/swarm/source.rs @@ -0,0 +1,169 @@ +//! The catch-up source: while a node catches up (joining, or rejoining a newer epoch, +//! until its initial sync is done), it syncs with one peer only. +//! +//! Every peer would otherwise send us what we lack, all at once: a node with an empty +//! document gets the whole document from each of its peers, and a node that is merely +//! behind gets the missing changes from each. Automerge can't deduplicate across peers. +//! So the first suitable peer we would sync with becomes the source; the others wait +//! until the initial sync is done, and then only exchange heads. Messages from waiting +//! peers are still applied; we just don't answer them yet. +//! +//! A source must be ready and willing to sync with us, as far as we know (its sync +//! messages and keepalives say so). It counts as alive while any message from it +//! arrives, so a slow transfer of a large document never looks like a stall. When it +//! is quiet for `Timing::catch_up_probe`, we PING it; after three unanswered probes, or +//! when it says it isn't ready, or twice in a row that it wouldn't sync with us (the +//! first time may be a handshake still under way), the next peer takes over. + +use std::sync::atomic::Ordering; +use std::time::Duration; + +use tokio::time::Instant; +use tracing::{debug, info}; + +use crate::state::{AppState, CatchUpSource, PeerState}; +use crate::swarm::sync::SyncEvent; + +/// Probes without an answer before the source is given up. +const UNANSWERED_PROBES: u32 = 3; + +/// Keepalives in a row saying "I wouldn't sync with you" before the source is given up. +const REFUSALS: u32 = 2; + +fn catching_up(state: &AppState) -> bool { + !state.initial_sync_done.load(Ordering::Acquire) +} + +fn lock(state: &AppState) -> std::sync::MutexGuard<'_, Option<CatchUpSource>> { + state + .catch_up_source + .lock() + .unwrap_or_else(|poisoned| poisoned.into_inner()) +} + +/// Returns how long to hold back sync with `peer`, if at all: while catching up, with +/// everyone but the source. Picks the source if there is none (or it went quiet). +pub(crate) fn defer(state: &AppState, peer: &str, peer_state: &PeerState) -> Option<Duration> { + if !catching_up(state) { + return None; + } + let stall = state.timing.catch_up_probe * UNANSWERED_PROBES; + if peer_state.reports_ready == Some(false) || peer_state.syncs_with_us == Some(false) { + // Not a candidate; its next message re-evaluates this, the timer is a fallback. + return Some(stall); + } + + let now = Instant::now(); + let mut source = lock(state); + let quiet = match &*source { + Some(current) if current.peer == peer => return None, + Some(current) if now < current.last_heard + stall => { + return Some(current.last_heard + stall - now); + } + Some(current) => Some(current.peer.clone()), + None => None, + }; + match quiet { + Some(previous) => info!( + peer = %peer, + previous = %previous, + quiet_secs = stall.as_secs_f32(), + "Catch-up source stopped answering; catching up from another peer" + ), + None => debug!(peer = %peer, "Catching up from this peer first"), + } + *source = Some(CatchUpSource { + peer: peer.to_owned(), + last_heard: now, + last_probe: None, + refusals: 0, + }); + None +} + +/// Any message from `peer` arrived. +pub(crate) fn heard_from(state: &AppState, peer: &str) { + if let Some(current) = lock(state).as_mut().filter(|current| current.peer == peer) { + current.last_heard = Instant::now(); + } +} + +/// What `peer` said about itself, in a sync message (`syncs_with_us: None`) or a +/// keepalive. Gives up the source if that rules it out. +pub(crate) fn reported(state: &AppState, peer: &str, ready: bool, syncs_with_us: Option<bool>) { + if !catching_up(state) { + return; + } + let reason = { + let mut source = lock(state); + let Some(current) = source.as_mut().filter(|current| current.peer == peer) else { + return; + }; + let reason = if !ready { + Some("it isn't ready itself") + } else { + match syncs_with_us { + Some(false) => { + current.refusals += 1; + (current.refusals >= REFUSALS).then_some("it wouldn't sync with us") + } + Some(true) => { + current.refusals = 0; + None + } + None => None, + } + }; + if reason.is_some() { + *source = None; + } + reason + }; + if let Some(reason) = reason { + debug!(peer = %peer, reason, "Giving up the catch-up source; trying another peer"); + resync_everyone(state); + } +} + +/// Whether to PING `peer` now: it is the source and has been quiet for a probe interval. +pub(crate) fn probe_due(state: &AppState, peer: &str, now: Instant) -> bool { + if !catching_up(state) { + return false; + } + let probe = state.timing.catch_up_probe; + let mut source = lock(state); + let Some(current) = source.as_mut().filter(|current| current.peer == peer) else { + return false; + }; + let due = now >= current.last_heard + probe + && current.last_probe.is_none_or(|last| now >= last + probe); + if due { + current.last_probe = Some(now); + } + due +} + +/// The initial sync is done: no source any more, and the peers we held back only need +/// to learn our heads now. +pub(crate) fn finished(state: &AppState, from: &str) { + *lock(state) = None; + let state = state.clone(); + let from = from.to_owned(); + tokio::spawn(async move { + let peers: Vec<String> = state.peer_states.read().await.keys().cloned().collect(); + for other in peers.into_iter().filter(|other| *other != from) { + let _ = state.sync_events.send(SyncEvent::PeerNeedsSync(other)); + } + }); +} + +/// After giving up a source: let every peer be considered again. +fn resync_everyone(state: &AppState) { + let state = state.clone(); + tokio::spawn(async move { + let peers: Vec<String> = state.peer_states.read().await.keys().cloned().collect(); + for other in peers { + let _ = state.sync_events.send(SyncEvent::PeerNeedsSync(other)); + } + }); +} |