diff options
Diffstat (limited to 'src')
| -rw-r--r-- | src/config/mod.rs | 10 | ||||
| -rw-r--r-- | src/main.rs | 48 | ||||
| -rw-r--r-- | src/remake/snapshot.rs | 53 | ||||
| -rw-r--r-- | src/swarm/bootstrap.rs | 65 |
4 files changed, 163 insertions, 13 deletions
diff --git a/src/config/mod.rs b/src/config/mod.rs index b4ea0aa..24791d3 100644 --- a/src/config/mod.rs +++ b/src/config/mod.rs @@ -110,6 +110,13 @@ pub(crate) const OPTIONS: &[OptionSpec] = &[ "Start a new swarm, as its genesis node, instead of joining one through seeds. Every start with it starts another new swarm: drop it once the swarm exists, and give the node seeds instead. With a --state-file that remembers a swarm, it starts that swarm again instead (empty; nodes of it still running merge with this one).", ), option( + "recover-from", + Kind::Text, + "PATH", + None, + "Start the swarm again from a backup (as GET /admin/document saves it): this node loads it, on the swarm and epoch it names, and is ready at once; other nodes join it. Members of the swarm still running merge with it, or it rejoins them if they moved on to a newer epoch (the backup is then dropped).", + ), + option( "state-file", Kind::Text, "PATH", @@ -230,6 +237,7 @@ pub(crate) struct Settings { #[serde(default)] pub(crate) genesis: bool, pub(crate) state_file: Option<PathBuf>, + pub(crate) recover_from: Option<PathBuf>, pub(crate) swarm_key_file: Option<PathBuf>, pub(crate) alias: Option<String>, #[serde(default = "default_keepalive_secs")] @@ -429,6 +437,8 @@ const CONFLICTS: &[(&[&str], &[&str])] = &[ ), // A new swarm has nobody to join. (&["genesis"], &["seeds"]), + // Recovering starts the swarm too. + (&["recover-from"], &["genesis"]), ]; /// The options `name` can't be combined with. diff --git a/src/main.rs b/src/main.rs index 4cfae66..d6773eb 100644 --- a/src/main.rs +++ b/src/main.rs @@ -116,12 +116,21 @@ async fn main() -> Result<()> { let (change_tx, _) = watch::channel(0_u64); let (change_events, _) = broadcast::channel(256); - let initial_epoch = match args.debug_epoch { - Some(generation) => Epoch { + let recovered = match &args.recover_from { + Some(path) => Some(bootstrap::load_backup(path)?), + None => None, + }; + let (initial_doc, recovered_epoch, recovered_swarm) = match recovered { + Some((doc, epoch, swarm_id)) => (doc, Some(epoch), Some(swarm_id)), + None => (AutoCommit::new(), None, None), + }; + let initial_epoch = match (recovered_epoch, args.debug_epoch) { + (Some(epoch), _) => epoch, + (None, Some(generation)) => Epoch { generation, id: [0xee; 32], }, - None => Epoch::GENESIS, + (None, None) => Epoch::GENESIS, }; let (epoch_tx, _) = watch::channel(initial_epoch); debug!(epoch = %initial_epoch, "Local epoch"); @@ -133,7 +142,7 @@ async fn main() -> Result<()> { debug!(local_id = mycelium_transport.local_id(), "Transport ready"); let state = AppState { - doc: Arc::new(RwLock::new(Replica::new(initial_epoch, AutoCommit::new()))), + doc: Arc::new(RwLock::new(Replica::new(initial_epoch, initial_doc))), epoch_tx, lifecycle: Arc::new(RwLock::new(NodeLifecycle::Joining { discovered_swarm_id: None, @@ -179,10 +188,31 @@ async fn main() -> Result<()> { trace!("Application state initialized"); - // Join the remembered swarm and no other; HELLOs claim it. - if let Some(remembered) = remembered { - state.set_expected_swarm_id(remembered.swarm_id).await?; - } + // Join the remembered swarm and no other; HELLOs claim it. A backup to recover from + // names its swarm itself, which wins. + let start = match recovered_swarm { + Some(swarm_id) => { + if let Some(remembered) = remembered.filter(|r| r.swarm_id != swarm_id) { + warn!( + backup = %swarm_id, + remembered = %remembered.swarm_id, + "The backup is of another swarm than the remembered one; recovering the backup's" + ); + } + state.set_expected_swarm_id(swarm_id).await?; + bootstrap::Start::Recovered + } + None => { + if let Some(remembered) = remembered { + state.set_expected_swarm_id(remembered.swarm_id).await?; + } + if args.genesis { + bootstrap::Start::Genesis + } else { + bootstrap::Start::Join + } + } + }; let consumer_state = state.clone(); tokio::spawn(async move { @@ -215,7 +245,7 @@ async fn main() -> Result<()> { tokio::spawn(remake::run_worker(state.clone())); trace!("Remake worker task spawned"); - let joining = bootstrap::bootstrap(&state, &seeds, args.genesis); + let joining = bootstrap::bootstrap(&state, &seeds, start); tokio::select! { joined = joining => joined.context("Joining swarm")?, _ = stopped(stop.clone()) => { diff --git a/src/remake/snapshot.rs b/src/remake/snapshot.rs index ca50c86..3c180ba 100644 --- a/src/remake/snapshot.rs +++ b/src/remake/snapshot.rs @@ -226,6 +226,38 @@ pub(crate) fn verify_document(doc: &mut AutoCommit, epoch: Epoch) -> Result<Epoc parse_metadata(&at_snapshot, epoch.generation) } +/// The epoch `doc` is on: genesis if it holds no epoch metadata; otherwise the newest +/// generation it has metadata for, whose snapshot change is the document's only change +/// without dependencies (verified). For a document loaded from a backup. +pub(crate) fn epoch_of(doc: &mut AutoCommit) -> Result<Epoch, String> { + let generation = store::keys_at(doc, &[META_ROOT.to_owned(), EPOCHS.to_owned()]) + .map_err(|error| error.to_string())? + .iter() + .filter_map(|key| key.parse::<u64>().ok()) + .max(); + let Some(generation) = generation.filter(|generation| *generation > 0) else { + return Ok(Epoch::GENESIS); + }; + let roots: Vec<ChangeHash> = doc + .get_changes(&[]) + .iter() + .filter(|change| change.deps().is_empty()) + .map(|change| change.hash()) + .collect(); + let [root] = roots.as_slice() else { + return Err(format!( + "generation {generation} needs exactly one change without dependencies (its snapshot), found {}", + roots.len() + )); + }; + let epoch = Epoch { + generation, + id: root.0, + }; + verify_document(doc, epoch)?; + Ok(epoch) +} + /// Metadata of an earlier generation, as the current document holds it. Snapshots /// carry the metadata of all earlier epochs forward (only remake records stay /// behind), so a node many epochs behind can still find the epoch that followed its own. @@ -523,6 +555,27 @@ mod tests { } #[test] + fn epoch_of_a_backup() { + let mut a = doc(b"actor-a"); + store::write_path(&mut a, &["k".into()], &json!(1)).unwrap(); + a.commit(); + assert_eq!(epoch_of(&mut a), Ok(Epoch::GENESIS)); + + let heads = a.get_heads(); + let policy = json!({}); + let mut snap = build(&mut a, ¶ms(&heads, &policy)).unwrap(); + // Saved and loaded, with a write since, as a backup would be. + let mut later = snap + .doc + .fork() + .with_actor(ActorId::from(b"actor-b".as_slice())); + store::write_path(&mut later, &["k".into()], &json!(2)).unwrap(); + later.commit(); + let mut backup = AutoCommit::load(&later.save()).unwrap(); + assert_eq!(epoch_of(&mut backup), Ok(snap.epoch)); + } + + #[test] fn metadata_written_and_remake_records_dropped() { let mut a = doc(b"actor-a"); store::write_path( diff --git a/src/swarm/bootstrap.rs b/src/swarm/bootstrap.rs index 84ff156..125321b 100644 --- a/src/swarm/bootstrap.rs +++ b/src/swarm/bootstrap.rs @@ -7,18 +7,25 @@ //! //! 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::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::{META_ROOT, SWARM}; +use crate::remake::snapshot::{self, META_ROOT, SWARM}; use crate::state::AppState; use crate::swarm::peer::{remember_peer, send_hello_once}; @@ -29,8 +36,31 @@ 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 { +/// 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 { @@ -102,6 +132,33 @@ 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`, as `GET /admin/document` saves it): the +/// document under a fresh actor, the epoch it is on and the swarm it names. +pub(crate) fn load_backup(path: &Path) -> Result<(AutoCommit, Epoch, String)> { + 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((doc, epoch, swarm_id)) +} + /// 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>) { |