path: root/src/swarm/bootstrap.rs

//! 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<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
            // 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<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. 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,
        ));
    }
    warn!(
        local_swarm = ?state.swarm_id().await,
        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"
    );
}