summaryrefslogtreecommitdiffstats
path: root/src/swarm/bootstrap.rs
diff options
context:
space:
mode:
Diffstat (limited to 'src/swarm/bootstrap.rs')
-rw-r--r--src/swarm/bootstrap.rs386
1 files changed, 326 insertions, 60 deletions
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"
+ );
}
generated by cgit v1.3.1 (git 2.54.0) at 2026-10-26 00:29:54 +0000