diff options
Diffstat (limited to 'src/swarm/bootstrap.rs')
| -rw-r--r-- | src/swarm/bootstrap.rs | 252 |
1 files changed, 240 insertions, 12 deletions
diff --git a/src/swarm/bootstrap.rs b/src/swarm/bootstrap.rs index 1ae5173..5e9ac06 100644 --- a/src/swarm/bootstrap.rs +++ b/src/swarm/bootstrap.rs @@ -7,18 +7,29 @@ //! //! 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::{Context, Result}; +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::remake::snapshot::{META_ROOT, SWARM}; +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}; @@ -29,10 +40,52 @@ const BOOTSTRAP_REPORT: Duration = Duration::from_secs(300); 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], genesis: bool) -> Result<()> { - if genesis { - let swarm_id = Uuid::new_v4().to_string(); - info!(swarm_id = %swarm_id, "Starting a new swarm as its genesis node (--genesis)"); +/// 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)) @@ -40,6 +93,8 @@ pub(crate) async fn bootstrap(state: &AppState, seeds: &[String], genesis: bool) .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(()); } @@ -52,6 +107,7 @@ pub(crate) async fn bootstrap(state: &AppState, seeds: &[String], genesis: bool) } 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; @@ -68,7 +124,7 @@ pub(crate) async fn bootstrap(state: &AppState, seeds: &[String], genesis: bool) let now = Instant::now(); if !unresolved.is_empty() && now >= next_resolve { - resolve_seeds(state, &mut unresolved).await; + 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); @@ -86,21 +142,190 @@ fn swarm_id_path() -> Vec<String> { vec![META_ROOT.to_owned(), SWARM.to_owned()] } +/// 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 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<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. -async fn resolve_seeds(state: &AppState, unresolved: &mut Vec<String>) { +/// 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) => { - debug!(seed = %seed, peer = %peer, "Resolved seed"); + 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) => { - warn!(seed = %seed, error = %error, "Seed doesn't resolve; will retry"); + 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); } } @@ -126,8 +351,11 @@ async fn report_joining(state: &AppState, seeds: &[String], unresolved: &[String 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 = ?state.swarm_id().await, + local_swarm = ?local_swarm, local_epoch = %state.local_epoch(), initial_sync_done = state.initial_sync_done.load(Ordering::Acquire), unresolved_seeds = ?unresolved, |