//! 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 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_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); /// 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(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(()); } 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 } 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(()); } 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 mut unresolved: Vec = 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 // ready only once that rejoin verified the epoch. if state.initial_sync_done.load(Ordering::Acquire) && !state.remake.is_rejoining() { state.mark_ready(swarm_id.clone()).await?; info!(swarm_id = %swarm_id, "Node is ready"); return Ok(()); } } 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; } } fn swarm_id_path() -> Vec { vec![META_ROOT.to_owned(), SWARM.to_owned()] } /// The swarm ID the document names, if any. fn read_swarm_id(doc: &AutoCommit) -> Option { 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, } /// 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> { 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 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!( "{what} is of swarm {}, but the state file remembers swarm {remembered}; add --force swarmId to start on {what}'s swarm anyway", starting.swarm_id ); } 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, ) -> Result { 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 { 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, failing: &mut HashSet, ) { 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 = 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" ); }