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::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::replica::Epoch;
use crate::document::store;
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 a swarm from a backup already loaded as the document (`--recover-from`).
    Recovered,
}

pub(crate) async fn bootstrap(state: &AppState, seeds: &[String], start: Start) -> Result<()> {
    if start == Start::Recovered {
        let Some(swarm_id) = state.swarm_id().await else {
            bail!("recovering without a swarm ID");
        };
        info!(swarm_id = %swarm_id, epoch = %state.local_epoch(), "Recovered the swarm from a backup");
        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()).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()).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 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).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,
    }
}

/// A backup to recover from (`--recover-from`).
pub(crate) struct Backup {
    /// Under a fresh actor.
    pub(crate) doc: AutoCommit,
    pub(crate) epoch: Epoch,
    pub(crate) swarm_id: String,
    /// The file's size (`swarm::stats`, disk).
    pub(crate) bytes: u64,
}

/// Loads a backup to recover from, as `GET /admin/document` saves it.
pub(crate) fn load_backup(path: &Path) -> Result<Backup> {
    let bytes =
        std::fs::read(path).with_context(|| format!("Reading the backup {}", path.display()))?;
    let mut doc = AutoCommit::load(&bytes)
        .with_context(|| format!("Loading the backup {}", path.display()))?
        .with_actor(ActorId::random());
    let Some(swarm_id) = read_swarm_id(&doc) else {
        bail!(
            "The backup {} names no swarm (no _continuum/swarm); it predates swarm IDs in documents",
            path.display()
        );
    };
    let epoch = snapshot::epoch_of(&mut doc)
        .map_err(|reason| anyhow::anyhow!("The backup {}: {reason}", path.display()))?;
    Ok(Backup {
        doc,
        epoch,
        swarm_id,
        bytes: bytes.len() as u64,
    })
}

/// 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>) {
    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");
                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");
                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"
    );
}