diff options
Diffstat (limited to 'src/swarm')
| -rw-r--r-- | src/swarm/blob.rs | 8 | ||||
| -rw-r--r-- | src/swarm/bootstrap.rs | 386 | ||||
| -rw-r--r-- | src/swarm/members.rs | 6 | ||||
| -rw-r--r-- | src/swarm/mod.rs | 1 | ||||
| -rw-r--r-- | src/swarm/peer.rs | 96 | ||||
| -rw-r--r-- | src/swarm/source.rs | 111 | ||||
| -rw-r--r-- | src/swarm/state_file.rs | 189 | ||||
| -rw-r--r-- | src/swarm/sync.rs | 17 | ||||
| -rw-r--r-- | src/swarm/topology.rs | 82 |
9 files changed, 811 insertions, 85 deletions
diff --git a/src/swarm/blob.rs b/src/swarm/blob.rs index 8c891ae..42cfbf3 100644 --- a/src/swarm/blob.rs +++ b/src/swarm/blob.rs @@ -171,8 +171,10 @@ async fn prepare(state: &AppState, id: BlobId) -> Option<Arc<Prepared>> { } async fn reply(state: &AppState, peer: &str, id: BlobId, body: BlobBody) { + let (epoch, epoch_seq) = state.epoch_stamp().await; let message = ParsedMessage::Blob(BlobMessage { - epoch: state.local_epoch(), + epoch, + epoch_seq, id, body, }); @@ -223,8 +225,10 @@ impl Drop for Registration<'_> { } async fn request(state: &AppState, peer: &str, id: BlobId, chunks: Vec<u32>) { + let (epoch, epoch_seq) = state.epoch_stamp().await; let message = ParsedMessage::Blob(BlobMessage { - epoch: state.local_epoch(), + epoch, + epoch_seq, id, body: BlobBody::Request { chunks }, }); diff --git a/src/swarm/bootstrap.rs b/src/swarm/bootstrap.rs index 27ffdd8..5e9ac06 100644 --- a/src/swarm/bootstrap.rs +++ b/src/swarm/bootstrap.rs @@ -1,55 +1,116 @@ //! Joining a swarm (or creating one as the genesis node). +//! +//! A node starts a new swarm only with `--genesis`. Otherwise it joins through its +//! seeds and keeps trying until it is ready: seeds that don't resolve are resolved +//! again with backoff, and HELLOs to resolved ones are retried by the peer worker. A +//! node without seeds waits (nothing to join yet) and says so now and then. +//! +//! The genesis node records the swarm ID in the document (`_continuum/swarm`), so +//! snapshots carry it forward and every backup names its swarm. +//! +//! A node recovering from a backup (`--recover-from`) starts on the backup's document, +//! swarm and epoch instead, and is ready at once, like a genesis node with content. use std::sync::atomic::Ordering; use std::time::Duration; -use anyhow::{Result, bail}; -use tracing::{debug, error, info, warn}; +use std::collections::HashSet; +use std::path::Path; + +use anyhow::{Context, Result, bail}; +use automerge::{ActorId, AutoCommit}; +use serde_json::Value as JsonValue; +use tokio::time::Instant; +use tracing::{debug, info, warn}; use uuid::Uuid; +use crate::document; +use crate::document::replica::Epoch; +use crate::document::store; +use crate::files::{FileKind, Files}; +use crate::force::Force; +use crate::remake::snapshot::{self, META_ROOT, SWARM}; use crate::state::AppState; use crate::swarm::peer::{remember_peer, send_hello_once}; -const BOOTSTRAP_TIMEOUT: Duration = Duration::from_secs(300); -const BOOTSTRAP_RETRY: Duration = Duration::from_millis(500); +const BOOTSTRAP_POLL: Duration = Duration::from_millis(500); +/// How often a node that is still joining says why. +const BOOTSTRAP_REPORT: Duration = Duration::from_secs(300); +/// Retrying seeds that didn't resolve: first after this, doubling up to the maximum. +const RESOLVE_RETRY: Duration = Duration::from_secs(5); +const RESOLVE_RETRY_MAX: Duration = Duration::from_secs(300); -pub(crate) async fn bootstrap(state: &AppState, seeds: &[String]) -> Result<()> { - if seeds.is_empty() { - info!("Initializing genesis node (no seeds configured)"); +/// How this node starts. +#[derive(Debug, Clone, Copy, PartialEq, Eq)] +pub(crate) enum Start { + /// Join a swarm through seeds (or remembered members). + Join, + /// Start a swarm, empty (`--genesis`). + Genesis, + /// Start on a document already loaded as the replica (`StartingDocument`). + FromDocument(Origin), +} + +pub(crate) async fn bootstrap(state: &AppState, seeds: &[String], start: Start) -> Result<()> { + if let Start::FromDocument(origin) = start { + let Some(swarm_id) = state.swarm_id().await else { + bail!("starting on a document without a swarm ID"); + }; + info!( + swarm_id = %swarm_id, + epoch = %state.local_epoch(), + "Starting on {}, ready at once", + origin.describe() + ); state.initial_sync_done.store(true, Ordering::Release); - state.mark_ready(Uuid::new_v4().to_string()).await?; + state.mark_ready(swarm_id).await?; + // Members of the swarm still running, if any: they merge with this node, or it + // rejoins them if they moved on to a newer epoch. + resolve_seeds(state, &mut seeds.to_vec(), &mut HashSet::new()).await; return Ok(()); } - - info!( - timeout_seconds = BOOTSTRAP_TIMEOUT.as_secs(), - "Joining existing swarm" - ); - - let mut peers = Vec::new(); - - for seed in seeds { - match state.transport.resolve_seed(seed).await { - Ok(peer) => { - debug!(seed = %seed, peer = %peer, "Resolved seed"); - peers.push(peer) + if start == Start::Genesis { + // A remembered swarm starts again under its ID: nodes of it still running then + // merge with this one instead of ending up in another swarm. + let swarm_id = match state.swarm_id().await { + Some(remembered) => { + warn!( + swarm_id = %remembered, + "--genesis with a remembered swarm: starting it again, empty; nodes of it still running merge with this one" + ); + remembered } - Err(e) => warn!(seed = %seed, error = %e, "Ignoring seed argument"), - } - } - - if peers.is_empty() { - bail!("None of the configured seeds could be resolved"); + None => { + let swarm_id = Uuid::new_v4().to_string(); + info!(swarm_id = %swarm_id, "Starting a new swarm as its genesis node (--genesis)"); + swarm_id + } + }; + let recorded = JsonValue::String(swarm_id.clone()); + state + .mutate_doc(move |doc| store::write_path(doc, &swarm_id_path(), &recorded)) + .await + .context("Recording the swarm ID")?; + state.initial_sync_done.store(true, Ordering::Release); + state.mark_ready(swarm_id).await?; + // Remembered members, if any, to find those still running. + resolve_seeds(state, &mut seeds.to_vec(), &mut HashSet::new()).await; + return Ok(()); } - for peer in peers { - remember_peer(state, &peer).await; - if let Err(error) = send_hello_once(state, &peer).await { - warn!(peer = %peer, error = %format!("{error:#}"), "Initial HELLO failed; will retry"); - } + if seeds.is_empty() { + warn!( + "No seeds: nothing to join. Add --seeds to join a swarm, or --genesis to start a new one" + ); + } else { + info!(seeds = seeds.len(), "Joining existing swarm"); } - let deadline = tokio::time::Instant::now() + BOOTSTRAP_TIMEOUT; + let mut unresolved: Vec<String> = seeds.to_vec(); + let mut failing = HashSet::new(); + let mut resolve_retry = RESOLVE_RETRY; + let mut next_resolve = Instant::now(); + let mut next_report = Instant::now() + BOOTSTRAP_REPORT; loop { if let Some(swarm_id) = state.swarm_id().await { // A node joining a compacted swarm rejoins the newer epoch first; it is @@ -61,39 +122,244 @@ pub(crate) async fn bootstrap(state: &AppState, seeds: &[String]) -> Result<()> } } - if tokio::time::Instant::now() >= deadline { - let mut peer_diagnostics: Vec<String> = Vec::new(); + let now = Instant::now(); + if !unresolved.is_empty() && now >= next_resolve { + resolve_seeds(state, &mut unresolved, &mut failing).await; + if !unresolved.is_empty() { + next_resolve = now + resolve_retry; + resolve_retry = (resolve_retry * 2).min(RESOLVE_RETRY_MAX); + } + } + if now >= next_report { + next_report = now + BOOTSTRAP_REPORT; + report_joining(state, seeds, &unresolved).await; + } + tokio::time::sleep(BOOTSTRAP_POLL).await; + } +} - let peer_states = state.peer_states.read().await.clone(); +fn swarm_id_path() -> Vec<String> { + vec![META_ROOT.to_owned(), SWARM.to_owned()] +} - for (peer, peer_state_mutex) in &peer_states { - let peer_state = peer_state_mutex.lock().await; - let validated_swarm = peer_state.standing.swarm_id(); - peer_diagnostics.push(format!( - "{peer}: validated_swarm={:?}, epoch={:?}", - validated_swarm, peer_state.epoch, - )) +/// The swarm ID the document names, if any. +fn read_swarm_id(doc: &AutoCommit) -> Option<String> { + match store::read_path(doc, &swarm_id_path()) { + Ok(Some(JsonValue::String(id))) if !id.is_empty() => Some(id), + _ => None, + } +} + +/// Where a document to start from came from. +#[derive(Debug, Clone, Copy, PartialEq, Eq)] +pub(crate) enum Origin { + /// A backup to recover from (`--recover-from`). + Backup, + /// This node's document file (`--document-file`). + DocumentFile, +} + +impl Origin { + fn describe(self) -> &'static str { + match self { + Origin::Backup => "the backup", + Origin::DocumentFile => "the document file", + } + } +} + +/// A document the node starts from, ready at once: on the swarm and epoch it names. +pub(crate) struct StartingDocument { + /// Under a fresh actor. + pub(crate) doc: AutoCommit, + pub(crate) epoch: Epoch, + pub(crate) swarm_id: String, + pub(crate) origin: Origin, + /// The actors this node wrote with before (from the document file). + pub(crate) earlier_actors: Vec<ActorId>, +} + +/// What `--recover-from` and `--document-file` say about where to start. +pub(crate) struct StartOptions<'a> { + pub(crate) recover_from: Option<&'a Path>, + pub(crate) document_file: Option<&'a Path>, + pub(crate) genesis: bool, + /// The swarm the state file remembers. + pub(crate) remembered: Option<&'a str>, +} + +/// The document to start from, if any, checked against the other options. Conflicts +/// are errors unless forced (`force`): +/// * an unreadable document file (`badDocumentFile`: moved aside, start without it); +/// * a backup to recover from while the document file holds a document +/// (`replaceDocumentFile`: the backup wins, and the file is rewritten from it); +/// * a document of another swarm than the state file remembers (`swarmId`). +/// +/// A document file holding a document with `--genesis` is an error: move it away. +pub(crate) fn starting_document( + options: &StartOptions<'_>, + files: &Files, + force: &Force, +) -> Result<Option<StartingDocument>> { + let from_file = match options.document_file { + None => None, + Some(path) => match document::file::load(path, files) { + Ok(loaded) => loaded, + Err(error) if force.allows("badDocumentFile") => { + warn!(error = %format!("{error:#}"), "Unusable document file; moving it aside and starting without it (--force badDocumentFile)"); + let aside = files.move_aside(path, ".bad").with_context(|| { + format!("Moving the document file {} aside", path.display()) + })?; + warn!(moved_to = %aside.display(), "Moved the document file aside"); + None + } + Err(error) => { + bail!( + "{error:#}; add --force badDocumentFile to move it aside and start without it" + ) } - let local_swarm = state.swarm_id().await; - let local_epoch = state.local_epoch(); - let initial_sync_done = state.initial_sync_done.load(Ordering::Acquire); - error!( - local_swarm = ?local_swarm, - local_epoch = %local_epoch, - initial_sync_done, - peers = %peer_diagnostics.join("; "), - timeout_seconds = BOOTSTRAP_TIMEOUT.as_secs(), - "Bootstrap timed out" + }, + }; + let backup = match options.recover_from { + None => None, + Some(path) => Some(load_backup(path, files)?), + }; + + let (doc, origin, earlier_actors) = match (backup, from_file) { + (Some(_), Some(_)) if !force.allows("replaceDocumentFile") => { + bail!( + "recover-from would replace the document in the document file; add --force replaceDocumentFile to recover the backup anyway" ); + } + (Some(backup), from_file) => { + if from_file.is_some() { + warn!( + "Recovering the backup instead of the document file's document (--force replaceDocumentFile)" + ); + } + (backup, Origin::Backup, Vec::new()) + } + (None, Some(loaded)) => (loaded.doc, Origin::DocumentFile, loaded.actors), + (None, None) => return Ok(None), + }; + if options.genesis && origin == Origin::DocumentFile { + bail!( + "The document file holds a document of a swarm, and genesis would start a new one; move the file away to start a new swarm" + ); + } + let starting = identify(doc, origin, earlier_actors)?; + + if let Some(remembered) = options.remembered.filter(|id| *id != starting.swarm_id) { + let what = origin.describe(); + if !force.allows("swarmId") { bail!( - "Bootstrap timed out after {:?}; local_swarm={:?}; local_epoch={}; initial_sync_done={}; peers=[{}]. Refusing to create a genesis document.", - BOOTSTRAP_TIMEOUT, - local_swarm, - local_epoch, - initial_sync_done, - peer_diagnostics.join("; ") + "{what} is of swarm {}, but the state file remembers swarm {remembered}; add --force swarmId to start on {what}'s swarm anyway", + starting.swarm_id ); } - tokio::time::sleep(BOOTSTRAP_RETRY).await; + warn!( + document = %starting.swarm_id, + remembered = %remembered, + "Starting on {what}'s swarm, not the remembered one (--force swarmId)" + ); + } + Ok(Some(starting)) +} + +/// The swarm and epoch `doc` is on. +fn identify( + mut doc: AutoCommit, + origin: Origin, + earlier_actors: Vec<ActorId>, +) -> Result<StartingDocument> { + let what = origin.describe(); + let Some(swarm_id) = read_swarm_id(&doc) else { + bail!("{what} names no swarm (no _continuum/swarm); it predates swarm IDs in documents"); + }; + let epoch = + snapshot::epoch_of(&mut doc).map_err(|reason| anyhow::anyhow!("{what}: {reason}"))?; + Ok(StartingDocument { + doc, + epoch, + swarm_id, + origin, + earlier_actors, + }) +} + +/// A backup to recover from, as `GET /admin/document` saves it, under a fresh actor. +fn load_backup(path: &Path, files: &Files) -> Result<AutoCommit> { + let bytes = files + .read(FileKind::Backup, path) + .with_context(|| format!("Reading the backup {}", path.display()))? + .with_context(|| format!("The backup {} doesn't exist", path.display()))?; + Ok(AutoCommit::load(&bytes) + .with_context(|| format!("Loading the backup {}", path.display()))? + .with_actor(ActorId::random())) +} + +/// Resolves the seeds in `unresolved`, says HELLO to each that resolves, and keeps the +/// rest for later. Warns once per seed that fails (`failing` remembers which), and says +/// when such a seed resolves after all. +async fn resolve_seeds( + state: &AppState, + unresolved: &mut Vec<String>, + failing: &mut HashSet<String>, +) { + let mut still = Vec::new(); + for seed in unresolved.drain(..) { + match state.transport.resolve_seed(&seed).await { + Ok(peer) => { + if failing.remove(&seed) { + info!(seed = %seed, peer = %peer, "Seed resolves now"); + } else { + debug!(seed = %seed, peer = %peer, "Resolved seed"); + } + remember_peer(state, &peer).await; + if let Err(error) = send_hello_once(state, &peer).await { + warn!(peer = %peer, error = %format!("{error:#}"), "Initial HELLO failed; will retry"); + } + } + Err(error) => { + if failing.insert(seed.clone()) { + warn!(seed = %seed, error = %error, "Seed doesn't resolve; trying again until it does"); + } else { + debug!(seed = %seed, error = %error, "Seed still doesn't resolve"); + } + still.push(seed); + } + } } + *unresolved = still; +} + +/// Why the node is still joining, for the log. +async fn report_joining(state: &AppState, seeds: &[String], unresolved: &[String]) { + if seeds.is_empty() { + warn!( + "Still no seeds: nothing to join. Add --seeds to join a swarm, or --genesis to start a new one" + ); + return; + } + let mut peer_diagnostics: Vec<String> = Vec::new(); + let peer_states = state.peer_states.read().await.clone(); + for (peer, peer_state_mutex) in &peer_states { + let peer_state = peer_state_mutex.lock().await; + peer_diagnostics.push(format!( + "{peer}: validated_swarm={:?}, epoch={:?}", + peer_state.standing.swarm_id(), + peer_state.epoch, + )); + } + // Awaited outside the macro: its arguments aren't Send, and a spawned bootstrap + // (the simulator's) must be. + let local_swarm = state.swarm_id().await; + warn!( + local_swarm = ?local_swarm, + local_epoch = %state.local_epoch(), + initial_sync_done = state.initial_sync_done.load(Ordering::Acquire), + unresolved_seeds = ?unresolved, + peers = %peer_diagnostics.join("; "), + "Still joining the swarm; trying on" + ); } diff --git a/src/swarm/members.rs b/src/swarm/members.rs index 221437e..9b0de18 100644 --- a/src/swarm/members.rs +++ b/src/swarm/members.rs @@ -21,11 +21,11 @@ use serde_json::{Value as JsonValue, json}; use tracing::{debug, info, warn}; use crate::document::store::{self, StoreError}; -use crate::state::{AppState, unix_millis}; +use crate::state::AppState; const META_ROOT: &str = "_continuum"; -fn members_dir() -> Vec<String> { +pub(crate) fn members_dir() -> Vec<String> { vec![META_ROOT.to_owned(), "members".to_owned()] } @@ -168,7 +168,7 @@ pub(crate) async fn ensure_registered(state: &AppState) { ) }; let wanted = state.alias.clone(); - let now = unix_millis(); + let now = state.clock.now_millis(); if !registered { let mut entry = json!({ "since": now, "alive": now }); diff --git a/src/swarm/mod.rs b/src/swarm/mod.rs index adb5d1c..d424529 100644 --- a/src/swarm/mod.rs +++ b/src/swarm/mod.rs @@ -11,6 +11,7 @@ pub(crate) mod relay; pub(crate) mod rtt; pub(crate) mod source; pub(crate) mod stability; +pub(crate) mod state_file; pub(crate) mod stats; pub(crate) mod sync; pub(crate) mod throughput; diff --git a/src/swarm/peer.rs b/src/swarm/peer.rs index 3d7da32..e8248bf 100644 --- a/src/swarm/peer.rs +++ b/src/swarm/peer.rs @@ -110,11 +110,12 @@ pub(crate) async fn send_ping(state: &AppState, peer: &str) -> Result<()> { pub(crate) async fn local_keepalive(state: &AppState, peer: &str) -> transport::Keepalive { // Heads and epoch come from the same locked replica, so they always match. The heads // are cached (`Replica`), so the read lock is enough. - let (heads, epoch) = { + let (heads, epoch, epoch_seq) = { let replica = state.doc.read().await; ( transport::HeadsSummary::of(replica.heads().to_vec()), replica.epoch, + replica.epoch_seq, ) }; let syncs_with_you = match state.get_peer_state(peer).await { @@ -129,6 +130,7 @@ pub(crate) async fn local_keepalive(state: &AppState, peer: &str) -> transport:: transport::Keepalive { heads, epoch, + epoch_seq, ready: state.is_ready().await, syncs_with_you, } @@ -276,6 +278,7 @@ async fn transmit_hello(state: &AppState, peer: &str) -> Result<()> { .as_ref() .map(|key| key.hello_tag(swarm_id.as_deref().unwrap_or(""), state.local_id(), peer)) .unwrap_or_default(); + let (epoch, epoch_seq) = state.epoch_stamp().await; state.lock_rtt().hello_sent(peer, Instant::now()); let result = state .transport @@ -283,7 +286,8 @@ async fn transmit_hello(state: &AppState, peer: &str) -> Result<()> { peer, &ParsedMessage::Hello(transport::HelloRequest { swarm_id, - epoch: state.local_epoch(), + epoch, + epoch_seq, auth, ready: state.is_ready().await, run_id: state.run_id, @@ -295,7 +299,11 @@ async fn transmit_hello(state: &AppState, peer: &str) -> Result<()> { let mut peer_state = peer_state_mutex.lock().await; if peer_state.hello_state == HelloState::Sending { + let was_unreachable = peer_state.unreachable_since.is_some(); peer_state.hello_sent(Instant::now()); + if !was_unreachable && peer_state.unreachable_since.is_some() { + state.topology_changed.notify_one(); + } let wait = match peer_state.liveness { Liveness::Dormant { .. } => state.timing.dormant_probe, Liveness::Active => peer_state.hello_backoff.next(), @@ -354,14 +362,17 @@ async fn send_hello_reply( .map(|key| key.hello_reply_tag(&swarm_id, state.local_id(), peer)) .unwrap_or_default(); let ready = state.is_ready().await; + let (epoch, epoch_seq) = state.epoch_stamp().await; let reply = transport::HelloReply { swarm_id, peers, - epoch: state.local_epoch(), + epoch, + epoch_seq, auth, ready, run_id, away_ms: away.map(|away| away.as_millis() as u64), + sender_run_id: state.run_id, }; debug!(to = %peer, ready, "Sending HELLO_REPLY"); @@ -399,15 +410,24 @@ async fn discover_peers(state: &AppState, from_peer: &str, advertised_peers: &[S } /// New neighbours get our HELLO (we may not have known them before: the member list is -/// how a partial mesh discovers them). Former neighbours get no more HELLOs, keepalives -/// or sync openings from us; their session is reset, so we don't keep pushing changes -/// to them either. They are still answered if they contact us. -async fn apply_neighbour_change(state: &AppState, change: topology::NeighbourChange) { +/// how a partial mesh discovers them) and a sync opening. The opening matters for a +/// neighbour whose handshake is already complete, e.g. one that returned from dormancy +/// while it wasn't our neighbour: it was only answered then, and with no HELLO to send, +/// nothing else would open sync until the keepalive heal (`resync_after`). Former +/// neighbours get no more HELLOs, keepalives or sync openings from us; their session is +/// reset, so we don't keep pushing changes to them either. They are still answered if +/// they contact us. +pub(crate) async fn apply_neighbour_change(state: &AppState, change: topology::NeighbourChange) { for peer in &change.added { remember_peer(state, peer).await; if let Err(error) = send_hello_once(state, peer).await { warn!(peer = %peer, error = %format!("{error:#}"), "HELLO to new neighbour failed; will retry"); } + // Skipped there unless the handshake is complete; otherwise its reply restarts + // sync anyway. + let _ = state + .sync_events + .send(SyncEvent::PeerNeedsSync(peer.clone())); } for peer in &change.removed { if let Some(peer_state) = state.get_peer_state(peer).await { @@ -427,12 +447,20 @@ pub(crate) async fn run_peer_worker(state: AppState) { info!("Starting peer maintenance worker"); let mut last_forget_check: Option<Instant> = None; + let mut was_ready = false; loop { tick.tick().await; // While joining, keep handshakes fast; dormancy would slow probes to minutes. let ready = state.is_ready().await; + if ready && !was_ready { + // Just became ready: register in the member list (and do the other periodic + // checks) now rather than at the next check. Until it is registered, other + // members don't count this node as a neighbour. + last_forget_check = None; + } + was_ready = ready; let peers: Vec<(String, Arc<Mutex<PeerState>>)> = state .peer_states @@ -461,12 +489,14 @@ pub(crate) async fn run_peer_worker(state: AppState) { if let Some(changes) = unanswered { debug!(peer = %peer, changes, "No answer to our sync message; sending again"); + crate::swarm::source::reply_timed_out(&state, peer); let _ = state .sync_events .send(SyncEvent::PeerNeedsSync(peer.clone())); } if became_dormant { + state.topology_changed.notify_one(); warn!( peer = %peer, silent_secs = state.timing.dormant_after.as_secs(), @@ -493,6 +523,8 @@ pub(crate) async fn run_peer_worker(state: AppState) { crate::swarm::members::ensure_registered(&state).await; crate::swarm::stability::tick(&state).await; crate::swarm::stats::measure_saved_size(&state); + crate::swarm::source::warn_if_stalled(&state); + crate::swarm::state_file::update(&state, false).await; { let known = state.peer_states.read().await; state.traffic.retain(|peer| known.contains_key(peer)); @@ -502,9 +534,8 @@ pub(crate) async fn run_peer_worker(state: AppState) { .unwrap_or_else(|poisoned| poisoned.into_inner()) .retain(|peer| known.contains_key(peer)); } - if let Some(change) = topology::refresh(&state).await { - apply_neighbour_change(&state, change).await; - } + // Fallback for what doesn't notify (`topology::run_worker`). + state.topology_changed.notify_one(); last_forget_check = Some(Instant::now()); } } @@ -581,9 +612,20 @@ pub(crate) async fn peer_swarm(state: &AppState, peer: &str) -> Option<String> { // --- epochs ------------------------------------------------------------------ -/// Records the epoch a peer advertised and compares it with ours. Returns true when both -/// are on the same epoch, the only case in which the documents may be synced. -async fn observe_peer_epoch(state: &AppState, peer: &str, remote: Epoch) -> bool { +/// Records the epoch a peer advertised and compares it with ours. Returns true when the +/// message's epoch is ours, the only case in which its documents may be synced. +/// +/// `seq` is the peer's `epoch_seq` with it. A message with a lower one than we have was +/// sent before the peer's last epoch change and delivered late: its epoch isn't +/// recorded, so it can't flip our view of the peer back. The count starts over when the +/// peer restarts, which its HELLO or HELLO_REPLY tells by a new run (`run`). +async fn observe_peer_epoch( + state: &AppState, + peer: &str, + remote: Epoch, + seq: u32, + run: Option<transport::RunId>, +) -> bool { let local = state.local_epoch(); let relation = local.relation_to(&remote); @@ -593,6 +635,21 @@ async fn observe_peer_epoch(state: &AppState, peer: &str, remote: Epoch) -> bool let previous = { let mut peer_state = peer_state_mutex.lock().await; + if let Some(run) = run.filter(|run| peer_state.run_id != Some(*run)) { + peer_state.run_id = Some(run); + peer_state.epoch_seq = None; + } + if peer_state.epoch_seq.is_some_and(|newest| seq < newest) { + debug!( + peer = %peer, + epoch = %remote, + seq, + newest = ?peer_state.epoch_seq, + "Ignoring a late message's epoch" + ); + return relation == EpochRelation::Same; + } + peer_state.epoch_seq = Some(seq); let previous = peer_state.epoch.replace(remote); if previous.is_some_and(|previous| previous != remote) { // The peer switched documents; our sync state describes its old one. Count @@ -749,6 +806,10 @@ pub(crate) async fn handle_message( let peer_state_mutex = state.get_or_create_peer_state(&peer).await; let mut peer_state = peer_state_mutex.lock().await; let previously_seen = peer_state.last_seen; + if peer_state.unreachable_since.is_some() { + // Reachable again (`mark_seen` clears it). + state.topology_changed.notify_one(); + } (peer_state.mark_seen(Instant::now()), previously_seen) }; crate::swarm::source::heard_from(state, &peer); @@ -757,7 +818,13 @@ pub(crate) async fn handle_message( info!(peer = %peer, "Dormant peer is active again; re-handshaking"); } - let same_epoch = observe_peer_epoch(state, &peer, message.epoch()).await; + let run = match &message { + transport::ParsedMessage::Hello(hello) => Some(hello.run_id), + transport::ParsedMessage::HelloReply(reply) => Some(reply.sender_run_id), + _ => None, + }; + let same_epoch = + observe_peer_epoch(state, &peer, message.epoch(), message.epoch_seq(), run).await; //debug!(from = %peer, message = ?message, "Received message"); @@ -807,6 +874,7 @@ pub(crate) async fn handle_message( debug!(peer = %peer, epoch = %envelope.epoch, "Dropping sync message from another epoch"); return Ok(()); } + crate::swarm::source::synced_from(state, &peer); handle_sync(state, &peer, envelope).await?; } transport::ParsedMessage::Ping(remote) => { diff --git a/src/swarm/source.rs b/src/swarm/source.rs index 378c40a..9eac445 100644 --- a/src/swarm/source.rs +++ b/src/swarm/source.rs @@ -15,12 +15,22 @@ //! 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. +//! +//! Alive isn't enough: a source that answers keepalives but not our sync messages (a +//! big answer lost on the way, a stuck sync state) would keep us catching up forever. +//! So it is also given up after three of our sync messages in a row went unanswered, +//! once nothing from it can still be on the way: no sync message from it for +//! `Timing::delivery_limit`, after which Mycelium has delivered or dropped anything it +//! sent before. It isn't chosen again for another `delivery_limit`, so another peer +//! takes over rather than the same one stalling again; if it is the only candidate, +//! catching up waits that long before trying it again. Meanwhile, every minute without a sync message from it is logged as a +//! warning. use std::sync::atomic::Ordering; use std::time::Duration; use tokio::time::Instant; -use tracing::{debug, info}; +use tracing::{debug, info, warn}; use crate::state::{AppState, CatchUpSource, PeerState}; use crate::swarm::sync::SyncEvent; @@ -35,6 +45,12 @@ 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; +/// Unanswered sync messages in a row before the source may be given up. +const REPLY_TIMEOUTS: u32 = 3; + +/// How often a stalled catch-up is logged. +const STALL_WARNING: Duration = Duration::from_secs(60); + fn lock_preference(state: &AppState) -> std::sync::MutexGuard<'_, Option<Instant>> { state .catch_up_preference @@ -42,6 +58,13 @@ fn lock_preference(state: &AppState) -> std::sync::MutexGuard<'_, Option<Instant .unwrap_or_else(|poisoned| poisoned.into_inner()) } +fn lock_avoided(state: &AppState) -> std::sync::MutexGuard<'_, Option<(String, Instant)>> { + state + .catch_up_avoided + .lock() + .unwrap_or_else(|poisoned| poisoned.into_inner()) +} + fn catching_up(state: &AppState) -> bool { !state.initial_sync_done.load(Ordering::Acquire) } @@ -60,12 +83,20 @@ pub(crate) fn defer(state: &AppState, peer: &str, peer_state: &PeerState) -> Opt return None; } let stall = state.timing.catch_up_probe * UNANSWERED_PROBES; + let now = Instant::now(); + let avoided_until = lock_avoided(state) + .as_ref() + .filter(|(avoided, until)| avoided == peer && now < *until) + .map(|(_, until)| *until); + if let Some(until) = avoided_until { + // Given up for not answering; the timer brings it back once it may be chosen. + return Some(until - now); + } 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, @@ -115,6 +146,9 @@ pub(crate) fn defer(state: &AppState, peer: &str, peer_state: &PeerState) -> Opt *source = Some(CatchUpSource { peer: peer.to_owned(), last_heard: now, + last_sync: now, + reply_timeouts: 0, + last_warned: None, last_probe: None, refusals: 0, }); @@ -128,6 +162,78 @@ pub(crate) fn heard_from(state: &AppState, peer: &str) { } } +/// A sync message from `peer` arrived: progress, if it is the source. +pub(crate) fn synced_from(state: &AppState, peer: &str) { + if let Some(current) = lock(state).as_mut().filter(|current| current.peer == peer) { + current.last_sync = Instant::now(); + current.reply_timeouts = 0; + } +} + +/// Our sync message to `peer` went unanswered (`SyncSession` reply deadline). Gives up +/// the source after `REPLY_TIMEOUTS` in a row, once nothing from it can be on the way. +pub(crate) fn reply_timed_out(state: &AppState, peer: &str) { + if !catching_up(state) { + return; + } + let given_up = { + let mut source = lock(state); + let Some(current) = source.as_mut().filter(|current| current.peer == peer) else { + return; + }; + current.reply_timeouts += 1; + let silent = current.last_sync.elapsed(); + let give_up = + current.reply_timeouts >= REPLY_TIMEOUTS && silent >= state.timing.delivery_limit; + let timeouts = current.reply_timeouts; + if give_up { + *source = None; + *lock_avoided(state) = Some(( + peer.to_owned(), + Instant::now() + state.timing.delivery_limit, + )); + } + give_up.then_some((timeouts, silent)) + }; + if let Some((timeouts, silent)) = given_up { + warn!( + peer = %peer, + reply_timeouts = timeouts, + silent_secs = silent.as_secs(), + "Catch-up source doesn't answer our sync messages; catching up from another peer" + ); + resync_everyone(state); + } +} + +/// Warns, at most every `STALL_WARNING`, while the source sends no sync messages. Called +/// by the peer worker on its maintenance tick. +pub(crate) fn warn_if_stalled(state: &AppState) { + if !catching_up(state) { + return; + } + let mut source = lock(state); + let Some(current) = source.as_mut() else { + return; + }; + let silent = current.last_sync.elapsed(); + if silent < STALL_WARNING + || current + .last_warned + .is_some_and(|at| at.elapsed() < STALL_WARNING) + { + return; + } + current.last_warned = Some(Instant::now()); + warn!( + peer = %current.peer, + silent_secs = silent.as_secs(), + reply_timeouts = current.reply_timeouts, + heard_secs_ago = current.last_heard.elapsed().as_secs(), + "Catching up stalls: no sync message from the catch-up source" + ); +} + /// 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>) { @@ -188,6 +294,7 @@ pub(crate) fn probe_due(state: &AppState, peer: &str, now: Instant) -> bool { pub(crate) fn finished(state: &AppState, from: &str) { *lock(state) = None; *lock_preference(state) = None; + *lock_avoided(state) = None; let state = state.clone(); let from = from.to_owned(); tokio::spawn(async move { diff --git a/src/swarm/state_file.rs b/src/swarm/state_file.rs new file mode 100644 index 0000000..8b883fc --- /dev/null +++ b/src/swarm/state_file.rs @@ -0,0 +1,189 @@ +//! The state file (`--state-file`): what a node remembers across restarts, so it +//! rejoins its swarm on its own: the swarm ID and some members to reach it through. +//! Not the document: a restarted node still starts empty and syncs. +//! +//! Written rarely: once the node is ready, then when the remembered members change +//! (checked every few minutes, written at most once an hour), and at shutdown if they +//! changed since, atomically (`files`). + +use std::path::PathBuf; +use std::sync::{Arc, Mutex, MutexGuard}; +use std::time::Duration; + +use serde::{Deserialize, Serialize}; +use tokio::time::Instant; +use tracing::{error, info, warn}; + +use crate::files::{FileKind, Files}; +use crate::state::AppState; +use crate::swarm::members; + +/// Members remembered as seeds. +const MAX_SEEDS: usize = 16; +/// How often the remembered members are compared with the registry. +const CHECK_INTERVAL: Duration = Duration::from_secs(300); +/// Writes at least this far apart, but for the first one and at shutdown. +const MIN_WRITE_INTERVAL: Duration = Duration::from_secs(3600); + +/// The file's content. +#[derive(Clone, Debug, Default, PartialEq, Eq, Serialize, Deserialize)] +#[serde(rename_all = "camelCase")] +pub(crate) struct Remembered { + pub(crate) swarm_id: String, + /// Public keys of members to join through (sorted). + #[serde(default)] + pub(crate) seeds: Vec<String>, +} + +pub(crate) struct StateFile { + path: PathBuf, + files: Arc<Files>, + // A std mutex: never held across an await. + inner: Mutex<Inner>, +} + +#[derive(Default)] +struct Inner { + written: Option<Remembered>, + written_at: Option<Instant>, + checked_at: Option<Instant>, +} + +impl StateFile { + /// Opens `path`, returning what it remembers. A missing file remembers nothing; an + /// unreadable one is moved aside (`.bad`) and logged, and remembers nothing either. + pub(crate) fn open(path: PathBuf, files: Arc<Files>) -> (Self, Option<Remembered>) { + let remembered = match files.read(FileKind::State, &path) { + Ok(Some(bytes)) => match serde_json::from_slice::<Remembered>(&bytes) { + Ok(remembered) if !remembered.swarm_id.is_empty() => Some(remembered), + Ok(_) => None, + Err(error) => { + error!( + path = %path.display(), + error = %error, + "Unreadable state file; moving it aside and remembering nothing" + ); + match files.move_aside(&path, ".bad") { + Ok(aside) => { + warn!(moved_to = %aside.display(), "Moved the state file aside") + } + Err(error) => warn!(error = %error, "Couldn't move the state file aside"), + } + None + } + }, + Ok(None) => None, + Err(error) => { + error!(path = %path.display(), error = %error, "Couldn't read the state file; remembering nothing"); + None + } + }; + let file = Self { + path, + files, + inner: Mutex::new(Inner { + written: remembered.clone(), + ..Inner::default() + }), + }; + (file, remembered) + } + + fn lock(&self) -> MutexGuard<'_, Inner> { + self.inner + .lock() + .unwrap_or_else(|poisoned| poisoned.into_inner()) + } + + fn write(&self, remembered: &Remembered) -> std::io::Result<()> { + let mut bytes = serde_json::to_vec_pretty(remembered)?; + bytes.push(b'\n'); + self.files.write_atomic(FileKind::State, &self.path, &bytes) + } +} + +/// Brings the state file up to date if it is due: `now` for the first write and at +/// shutdown, otherwise only every few minutes and at most once an hour. Called by the +/// peer worker on its maintenance tick. +pub(crate) async fn update(state: &AppState, now: bool) { + let Some(file) = state.state_file.clone() else { + return; + }; + if !state.is_ready().await { + return; + } + let Some(swarm_id) = state.swarm_id().await else { + return; + }; + { + let mut inner = file.lock(); + if !now + && inner + .checked_at + .is_some_and(|at| at.elapsed() < CHECK_INTERVAL) + { + return; + } + inner.checked_at = Some(Instant::now()); + } + + let seeds = { + let replica = state.doc.read().await; + seeds( + &replica.doc, + state.local_id(), + state.electorate_window, + state.clock.now_millis(), + ) + }; + let remembered = Remembered { swarm_id, seeds }; + + { + let inner = file.lock(); + if inner.written.as_ref() == Some(&remembered) { + return; + } + let first = inner.written_at.is_none(); + if !now + && !first + && inner + .written_at + .is_some_and(|at| at.elapsed() < MIN_WRITE_INTERVAL) + { + return; + } + } + + let writer = Arc::clone(&file); + let content = remembered.clone(); + match tokio::task::spawn_blocking(move || writer.write(&content)).await { + Ok(Ok(())) => { + info!( + path = %file.path.display(), + seeds = remembered.seeds.len(), + "Wrote the state file" + ); + let mut inner = file.lock(); + inner.written = Some(remembered); + inner.written_at = Some(Instant::now()); + } + Ok(Err(error)) => { + error!(path = %file.path.display(), error = %error, "Couldn't write the state file"); + } + Err(error) => error!(error = %error, "State file write task failed"), + } +} + +/// Members to rejoin through: those alive within the electorate window, stable ones +/// first, at most `MAX_SEEDS`, sorted. Changes only when members join, leave or change +/// tier, so the file rarely does. +fn seeds(doc: &automerge::AutoCommit, me: &str, window: Duration, now: u64) -> Vec<String> { + let (mut stable, transient): (Vec<String>, Vec<String>) = members::electorate(doc, now, window) + .into_iter() + .filter(|pk| pk != me) + .partition(|pk| members::tier(doc, pk) == members::Tier::Stable); + stable.extend(transient); + stable.truncate(MAX_SEEDS); + stable.sort(); + stable +} diff --git a/src/swarm/sync.rs b/src/swarm/sync.rs index 80d6d56..a05e346 100644 --- a/src/swarm/sync.rs +++ b/src/swarm/sync.rs @@ -141,10 +141,10 @@ async fn deliver_sync_to_peer(state: &AppState, peer: &str) -> Result<()> { tokio::task::spawn_blocking(move || -> Result<Option<transport::EncodedMessage>> { // Label the message with the epoch of the document it was generated // from, read under the same lock. - let (epoch, message) = { + let (epoch, epoch_seq, message) = { let mut replica = doc.blocking_write(); let mut peer_state = psm.blocking_lock(); - let epoch = replica.epoch; + let (epoch, epoch_seq) = (replica.epoch, replica.epoch_seq); let sent_before = peer_state.sync.state.sent_hashes.clone(); let message = { let sync_doc = replica.doc.sync(); @@ -178,7 +178,7 @@ async fn deliver_sync_to_peer(state: &AppState, peer: &str) -> Result<()> { regenerate: false, }); } - (epoch, message) + (epoch, epoch_seq, message) }; let Some(message) = message else { return Ok(None); @@ -186,6 +186,7 @@ async fn deliver_sync_to_peer(state: &AppState, peer: &str) -> Result<()> { let envelope = transport::SyncEnvelope { swarm_id, epoch, + epoch_seq, message, ready, }; @@ -329,6 +330,7 @@ pub(crate) async fn handle_sync( let epoch = envelope.epoch; let message = envelope.message; let peer = peer.to_owned(); + let clock = Arc::clone(&state.clock); // Some((caught_up, heads changed)) when applied, None when the message belongs // to another epoch. @@ -365,7 +367,7 @@ pub(crate) async fn handle_sync( stats.heads_seen(heads_after.len()); let changed = heads_after != heads_before; if changed && measure_propagation { - let now_ms = crate::state::unix_millis() as i64; + let now_ms = clock.now_millis() as i64; // Time 0: not stamped (snapshots, older nodes). let delays: Vec<i64> = replica .doc @@ -607,6 +609,13 @@ pub(crate) async fn run_change_publisher(state: AppState) { warn!(path = ?path, "Concurrent writes produced a conflict"); } + if paths + .iter() + .any(|path| crate::swarm::topology::affects_neighbours(path)) + { + state.topology_changed.notify_one(); + } + // Fails only when nobody is subscribed. let _ = state.change_events.send(Arc::new(ChangeSet { paths })); } diff --git a/src/swarm/topology.rs b/src/swarm/topology.rs index 57856b6..0dae7e9 100644 --- a/src/swarm/topology.rs +++ b/src/swarm/topology.rs @@ -107,6 +107,26 @@ pub(crate) struct NeighbourChange { pub(crate) removed: Vec<String>, } +/// Recomputes the neighbours whenever something they depend on may have changed +/// (`AppState::topology_changed`): the member list (`swarm::sync::run_change_publisher`), +/// a peer becoming unreachable or reachable again (`swarm::peer`), and every 5 s as a +/// fallback, which also covers unreachability expiring after `dormant_probe`. A burst of +/// notifications coalesces into one recomputation. +pub(crate) async fn run_worker(state: AppState) { + loop { + state.topology_changed.notified().await; + if let Some(change) = refresh(&state).await { + crate::swarm::peer::apply_neighbour_change(&state, change).await; + } + } +} + +/// Whether a change at `path` may change the neighbours: one in the member list, or the +/// root (the whole document replaced, as after an epoch switch). +pub(crate) fn affects_neighbours(path: &[String]) -> bool { + path.is_empty() || path.starts_with(&members::members_dir()) +} + /// Recomputes the neighbours once ready, if the member list, the reachable members or /// `k` changed. pub(crate) async fn refresh(state: &AppState) -> Option<NeighbourChange> { @@ -397,6 +417,68 @@ mod tests { .collect() } + fn hex(bytes: [u8; 32]) -> String { + bytes.iter().map(|byte| format!("{byte:02x}")).collect() + } + + /// Over random swarms (size, `k`, who is reachable, who is stable): links only + /// between reachable members, never to oneself, symmetric; a full mesh while small, + /// otherwise at least `k` links each and connected; independent of the order members + /// are listed in; and each member's own view (`neighbours`) agrees with the graph. + #[test] + fn random_swarms() { + crate::testrand::for_seeds(300, |rng| { + let n = rng.between(1, 60); + let k = rng.between(2, 14); + let mut all: Vec<String> = (0..n).map(|_| hex(rng.bytes32())).collect(); + let unreachable: BTreeSet<String> = + all.iter().filter(|_| rng.chance(0.3)).cloned().collect(); + let stable: BTreeSet<String> = + all.iter().filter(|_| rng.chance(0.3)).cloned().collect(); + let is_unreachable = |m: &str| unreachable.contains(m); + let is_stable = |m: &str| stable.contains(m); + + let g = super::graph(&all, k, &is_unreachable, &is_stable); + // Owned: `all` is shuffled below. + let live: BTreeSet<String> = all + .iter() + .filter(|m| !unreachable.contains(*m)) + .cloned() + .collect(); + assert_eq!(g.keys().cloned().collect::<BTreeSet<_>>(), live); + for (a, ns) in &g { + assert!(!ns.contains(a), "{a} links to itself"); + for b in ns { + assert!(live.contains(b), "{a} links to unreachable {b}"); + assert!(g[b].contains(a), "{a} -> {b} but not back"); + } + } + if live.len() <= k + 1 { + assert!( + g.values().all(|ns| ns.len() + 1 == live.len()), + "not a full mesh" + ); + } else { + assert!( + g.values().all(|ns| ns.len() >= k), + "a member with fewer than k links" + ); + assert!(connected(&g), "not connected"); + } + + rng.shuffle(&mut all); + assert_eq!(super::graph(&all, k, &is_unreachable, &is_stable), g); + + if let Some(local) = live.iter().nth(rng.below(live.len())) { + let others: Vec<String> = all.iter().filter(|m| *m != local).cloned().collect(); + assert_eq!( + neighbours(&others, local, k, &is_unreachable, &is_stable), + g[local] + ); + } + }); + } + #[test] fn small_swarms_are_a_full_mesh() { let ms = members(13); |