From 72472131a4bb3798759884da597cfb4ef7793782 Mon Sep 17 00:00:00 2001 From: tv Date: Sun, 25 Oct 2026 19:59:14 +0100 Subject: allow to persist document --- src/api/stats.rs | 4 + src/api/stats/prometheus.rs | 72 +++++---- src/config/mod.rs | 39 ++++- src/document/file.rs | 362 ++++++++++++++++++++++++++++++++++++++++++++ src/document/mod.rs | 1 + src/files.rs | 41 ++++- src/main.rs | 66 ++++---- src/state.rs | 2 + src/swarm/bootstrap.rs | 167 +++++++++++++++++--- 9 files changed, 670 insertions(+), 84 deletions(-) create mode 100644 src/document/file.rs diff --git a/src/api/stats.rs b/src/api/stats.rs index 8899421..fa11b45 100644 --- a/src/api/stats.rs +++ b/src/api/stats.rs @@ -49,6 +49,8 @@ struct DiskStats { read_bytes: u64, writes: u64, written_bytes: u64, + appends: u64, + appended_bytes: u64, errors: u64, } @@ -220,6 +222,8 @@ async fn collect(state: &AppState) -> StatsResponse { read_bytes: counts.read_bytes, writes: counts.writes, written_bytes: counts.written_bytes, + appends: counts.appends, + appended_bytes: counts.appended_bytes, errors: counts.errors, }; (kind.as_str(), stats) diff --git a/src/api/stats/prometheus.rs b/src/api/stats/prometheus.rs index e267042..74c18e9 100644 --- a/src/api/stats/prometheus.rs +++ b/src/api/stats/prometheus.rs @@ -54,6 +54,45 @@ impl Out { } } +const DISK_COUNTERS: usize = 7; + +/// Name, help and value of each counter of one kind of file. +fn disk_counters(c: &super::DiskStats) -> [(&'static str, &'static str, u64); DISK_COUNTERS] { + [ + ( + "disk_reads_total", + "Reads of this node's own files.", + c.reads, + ), + ( + "disk_read_bytes_total", + "Bytes read from them.", + c.read_bytes, + ), + ("disk_writes_total", "Whole files written.", c.writes), + ( + "disk_written_bytes_total", + "Bytes of them.", + c.written_bytes, + ), + ( + "disk_appends_total", + "Appends to this node's own files.", + c.appends, + ), + ( + "disk_appended_bytes_total", + "Bytes appended.", + c.appended_bytes, + ), + ( + "disk_errors_total", + "Failed reads, writes and appends.", + c.errors, + ), + ] +} + fn escape(value: &str) -> String { value .replace('\\', "\\\\") @@ -297,32 +336,13 @@ pub(super) async fn render(state: &AppState, stats: &StatsResponse) -> String { stats.stale_hello_replies as f64, ); - type Field = fn(&super::DiskStats) -> u64; - let disk: [(&str, &str, Field); 5] = [ - ("disk_reads_total", "Reads of this node's own files.", |c| { - c.reads - }), - ("disk_read_bytes_total", "Bytes read from them.", |c| { - c.read_bytes - }), - ( - "disk_writes_total", - "Writes of this node's own files.", - |c| c.writes, - ), - ("disk_written_bytes_total", "Bytes written to them.", |c| { - c.written_bytes - }), - ( - "disk_errors_total", - "Failed reads and writes of them.", - |c| c.errors, - ), - ]; - for (name, help, field) in disk { - out.family(name, "counter", help); - for (kind, counts) in &stats.disk { - out.sample(name, &[("file", *kind)], field(counts) as f64); + for counter in 0..DISK_COUNTERS { + for (i, (kind, counts)) in stats.disk.iter().enumerate() { + let (name, help, value) = disk_counters(counts)[counter]; + if i == 0 { + out.family(name, "counter", help); + } + out.sample(name, &[("file", *kind)], value as f64); } } diff --git a/src/config/mod.rs b/src/config/mod.rs index c83ca70..c56afa6 100644 --- a/src/config/mod.rs +++ b/src/config/mod.rs @@ -66,7 +66,7 @@ const fn option( } /// What `--force` may override at start (`force`). -pub(crate) const FORCE: &[&str] = &["swarmId"]; +pub(crate) const FORCE: &[&str] = &["swarmId", "badDocumentFile", "replaceDocumentFile"]; /// Every option. Names are the long option names and the configuration file keys. pub(crate) const OPTIONS: &[OptionSpec] = &[ @@ -117,7 +117,7 @@ pub(crate) const OPTIONS: &[OptionSpec] = &[ Kind::List, "NAME,[NAME...]", None, - "Override checks that otherwise refuse to start: true for all, or names: swarmId (a backup of another swarm than the state file remembers).", + "Override checks that otherwise refuse to start: true for all, or names: swarmId (a document to start on is of another swarm than the state file remembers), badDocumentFile (an unusable document file: moved aside, start without it), replaceDocumentFile (recover-from while the document file holds a document).", ), option( "recover-from", @@ -126,6 +126,27 @@ pub(crate) const OPTIONS: &[OptionSpec] = &[ 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( + "document-file", + Kind::Text, + "PATH", + None, + "This node's document on disk. Read at start: the node resumes with it, ready at once. Kept up to date while ready: changes are appended every --document-append-secs, and the file is rewritten after a switch, a rejoin or a restart, and when the appended changes outgrew it.", + ), + option( + "document-append-secs", + Kind::Integer, + "SECONDS", + Some("5"), + "How often changes are appended to the document file.", + ), + option( + "document-rewrite-secs", + Kind::Integer, + "SECONDS", + Some("3600"), + "Shortest time between rewrites of the document file because its appended changes outgrew its full save (a new epoch or a restart rewrites it at once).", + ), option( "state-file", Kind::Text, @@ -247,6 +268,11 @@ pub(crate) struct Settings { #[serde(default)] pub(crate) genesis: bool, pub(crate) state_file: Option, + pub(crate) document_file: Option, + #[serde(default = "default_document_append_secs")] + pub(crate) document_append_secs: u64, + #[serde(default = "default_document_rewrite_secs")] + pub(crate) document_rewrite_secs: u64, pub(crate) recover_from: Option, #[serde(default)] pub(crate) force: Vec, @@ -273,6 +299,12 @@ pub(crate) struct Settings { fn default_bind_addr() -> SocketAddr { SocketAddr::from(([127, 0, 0, 1], 3000)) } +fn default_document_append_secs() -> u64 { + 5 +} +fn default_document_rewrite_secs() -> u64 { + 3600 +} fn default_keepalive_secs() -> u64 { 60 } @@ -340,6 +372,9 @@ impl Settings { if self.neighbours < 2 { bail!("neighbours must be at least 2"); } + if self.document_append_secs == 0 || self.document_rewrite_secs == 0 { + bail!("document-append-secs and document-rewrite-secs must be at least 1"); + } if self.keepalive_secs == 0 { bail!("keepalive-secs must be at least 1"); } diff --git a/src/document/file.rs b/src/document/file.rs new file mode 100644 index 0000000..a773502 --- /dev/null +++ b/src/document/file.rs @@ -0,0 +1,362 @@ +//! The document file (`--document-file`): this node's document on disk, so a restart +//! resumes with it instead of starting empty. +//! +//! Format: the line `continuum-document/1`, a header line (JSON: the actor IDs this +//! node wrote with, so a later rejoin replays those writes as its own), then records, +//! each a little-endian `u32` length and that many bytes: first a full save of the +//! document, then increments (the changes after the heads written before). A record cut +//! short by a crash is dropped when reading. +//! +//! Written by a worker (`run`), only while the node is ready and not rejoining (a rejoin +//! keeps the old file until it is done): +//! * every `--document-append-secs`, what changed is appended; +//! * the whole file is rewritten, atomically, when the epoch or this node's actors +//! changed (a switch, a rejoin, a restart), and when the appended increments outgrew +//! the full save, at most every `--document-rewrite-secs`; +//! * at shutdown, what changed is appended. + +use std::collections::HashSet; +use std::path::{Path, PathBuf}; +use std::sync::{Arc, Mutex, MutexGuard}; +use std::time::Duration; + +use anyhow::{Context, Result, anyhow, bail}; +use automerge::{ActorId, AutoCommit, ChangeHash}; +use serde::{Deserialize, Serialize}; +use tokio::time::Instant; +use tracing::{debug, error, warn}; + +use crate::document::replica::Epoch; +use crate::files::{FileKind, Files}; +use crate::state::AppState; + +const MAGIC: &[u8] = b"continuum-document/1\n"; + +#[derive(Debug, Default, Serialize, Deserialize)] +struct Header { + /// Hex actor IDs. + #[serde(default)] + actors: Vec, +} + +/// A document file as read at start. +pub(crate) struct Loaded { + /// Under a fresh actor. + pub(crate) doc: AutoCommit, + /// The actors this node wrote with before. + pub(crate) actors: Vec, +} + +/// Reads the document file; None if there is none. An error if it can't be read or +/// isn't one: the caller decides (`--force badDocumentFile`). +pub(crate) fn load(path: &Path, files: &Files) -> Result> { + let Some(bytes) = files + .read(FileKind::Document, path) + .with_context(|| format!("Reading the document file {}", path.display()))? + else { + return Ok(None); + }; + let (loaded, dropped) = + parse(&bytes).with_context(|| format!("The document file {}", path.display()))?; + if dropped > 0 { + warn!( + path = %path.display(), + bytes = dropped, + "Dropped a record cut short at the end of the document file (a crash while appending)" + ); + } + Ok(Some(loaded)) +} + +/// The file's content, and how many bytes at its end were cut short. +fn parse(bytes: &[u8]) -> Result<(Loaded, usize)> { + let rest = bytes + .strip_prefix(MAGIC) + .ok_or_else(|| anyhow!("not a continuum document file"))?; + let line_end = rest + .iter() + .position(|byte| *byte == b'\n') + .ok_or_else(|| anyhow!("no header"))?; + let header: Header = serde_json::from_slice(&rest[..line_end]).context("header")?; + let actors = header + .actors + .iter() + .map(|hex| { + ActorId::try_from(hex.as_str()) + .map_err(|_| anyhow!("invalid actor {hex:?} in the header")) + }) + .collect::>>()?; + + let mut records = &rest[line_end + 1..]; + let mut doc: Option = None; + while !records.is_empty() { + let Some((length, after)) = records.split_first_chunk::<4>() else { + break; + }; + let length = u32::from_le_bytes(*length) as usize; + if after.len() < length { + break; + } + let (record, after) = after.split_at(length); + match &mut doc { + None => doc = Some(AutoCommit::load(record).context("the full save")?), + Some(doc) => { + doc.load_incremental(record).context("an increment")?; + } + } + records = after; + } + let Some(doc) = doc else { + bail!("no full save"); + }; + let loaded = Loaded { + doc: doc.with_actor(ActorId::random()), + actors, + }; + Ok((loaded, records.len())) +} + +fn record(bytes: &[u8]) -> Vec { + let mut out = Vec::with_capacity(bytes.len() + 4); + out.extend_from_slice(&(bytes.len() as u32).to_le_bytes()); + out.extend_from_slice(bytes); + out +} + +/// A whole file: header and one full save. +fn whole(actors: &[ActorId], save: &[u8]) -> Result> { + let header = Header { + actors: actors.iter().map(ActorId::to_hex_string).collect(), + }; + let mut out = MAGIC.to_vec(); + out.extend(serde_json::to_vec(&header)?); + out.push(b'\n'); + out.extend(record(save)); + Ok(out) +} + +/// The document file and what was last written to it. +pub(crate) struct DocumentFile { + path: PathBuf, + files: Arc, + append_every: Duration, + rewrite_every: Duration, + // A std mutex: never held across an await. + written: Mutex>, +} + +#[derive(Debug, Clone)] +struct Written { + epoch: Epoch, + /// This node's actors when the file was last rewritten (`RemakeState::own_actors`). + actors: usize, + heads: Vec, + /// The full save. + base_bytes: u64, + /// Appended since. + tail_bytes: u64, + rewritten_at: Instant, +} + +impl DocumentFile { + pub(crate) fn new( + path: PathBuf, + files: Arc, + append_every: Duration, + rewrite_every: Duration, + ) -> Self { + Self { + path, + files, + append_every, + rewrite_every, + written: Mutex::new(None), + } + } + + fn lock(&self) -> MutexGuard<'_, Option> { + self.written + .lock() + .unwrap_or_else(|poisoned| poisoned.into_inner()) + } +} + +/// Keeps the document file up to date. Runs for as long as the node does. +pub(crate) async fn run(state: AppState) { + let Some(file) = state.document_file.clone() else { + return; + }; + let mut tick = tokio::time::interval(file.append_every); + tick.set_missed_tick_behavior(tokio::time::MissedTickBehavior::Delay); + loop { + tick.tick().await; + flush(&state).await; + } +} + +/// Writes what is due now (see the module docs). Also called at shutdown. +pub(crate) async fn flush(state: &AppState) { + let Some(file) = state.document_file.clone() else { + return; + }; + if !state.is_ready().await || state.remake.is_rejoining() { + return; + } + let own = state.remake.own_actors(); + let (epoch, heads) = { + let replica = state.doc.read().await; + (replica.epoch, replica.heads().to_vec()) + }; + let written = file.lock().clone(); + let reason = match &written { + None => Some("first write"), + Some(w) if w.epoch != epoch => Some("new epoch"), + Some(w) if w.actors != own.len() => Some("new actor"), + Some(w) + if w.tail_bytes > w.base_bytes && w.rewritten_at.elapsed() >= file.rewrite_every => + { + Some("increments outgrew the full save") + } + Some(_) => None, + }; + match (reason, written) { + (Some(reason), _) => rewrite(state, &file, own, reason).await, + (None, Some(written)) if written.heads != heads => append(state, &file, written).await, + (None, _) => {} + } +} + +async fn rewrite(state: &AppState, file: &Arc, own: Vec, reason: &str) { + let start = Instant::now(); + let (epoch, mut copy) = { + let replica = state.doc.read().await; + (replica.epoch, replica.doc.clone()) + }; + let own_count = own.len(); + let writer = Arc::clone(file); + let written = tokio::task::spawn_blocking(move || -> Result<(Vec, u64)> { + let heads = copy.get_heads(); + // Only the actors with changes in this document matter for a later rejoin; + // the current one in any case. + let present: HashSet = copy + .get_changes(&[]) + .iter() + .map(|change| change.actor_id().clone()) + .collect(); + let current = copy.get_actor().clone(); + let actors: Vec = own + .into_iter() + .filter(|actor| *actor == current || present.contains(actor)) + .collect(); + let save = copy.save(); + let bytes = whole(&actors, &save)?; + writer + .files + .write_atomic(FileKind::Document, &writer.path, &bytes)?; + Ok((heads, save.len() as u64)) + }) + .await; + match written { + Ok(Ok((heads, base_bytes))) => { + debug!( + reason, + bytes = base_bytes, + took_ms = start.elapsed().as_millis() as u64, + "Rewrote the document file" + ); + *file.lock() = Some(Written { + epoch, + actors: own_count, + heads, + base_bytes, + tail_bytes: 0, + rewritten_at: Instant::now(), + }); + } + Ok(Err(error)) => { + error!(path = %file.path.display(), error = %format!("{error:#}"), "Couldn't rewrite the document file"); + } + Err(error) => error!(error = %error, "Document file rewrite task failed"), + } +} + +async fn append(state: &AppState, file: &Arc, written: Written) { + let (heads, increment) = { + let mut replica = state.doc.write().await; + // Switched since `flush` looked: the next one rewrites the file instead. An + // increment of another epoch's document must never be appended. + if replica.epoch != written.epoch { + return; + } + let heads = replica.heads().to_vec(); + (heads, replica.doc.save_after(&written.heads)) + }; + let bytes = record(&increment); + let writer = Arc::clone(file); + let appended = tokio::task::spawn_blocking(move || { + writer + .files + .append(FileKind::Document, &writer.path, &bytes) + }) + .await; + match appended { + Ok(Ok(())) => { + let mut guard = file.lock(); + if let Some(current) = guard + .as_mut() + .filter(|current| current.heads == written.heads) + { + current.heads = heads; + current.tail_bytes += increment.len() as u64 + 4; + debug!( + bytes = increment.len(), + tail_bytes = current.tail_bytes, + base_bytes = current.base_bytes, + "Appended to the document file" + ); + } + } + Ok(Err(error)) => { + // The file may end in a partial record now; the next rewrite replaces it. + error!(path = %file.path.display(), error = %error, "Couldn't append to the document file; rewriting it next"); + *file.lock() = None; + } + Err(error) => error!(error = %error, "Document file append task failed"), + } +} + +#[cfg(test)] +mod tests { + use super::*; + use automerge::{ROOT, ReadDoc, transaction::Transactable}; + + #[test] + fn whole_then_increments() { + let mut doc = AutoCommit::new(); + doc.put(ROOT, "a", 1_i64).unwrap(); + doc.commit(); + let actor = doc.get_actor().clone(); + let heads = doc.get_heads(); + let mut bytes = whole(std::slice::from_ref(&actor), &doc.save()).unwrap(); + + doc.put(ROOT, "b", 2_i64).unwrap(); + doc.commit(); + bytes.extend(record(&doc.save_after(&heads))); + + let (loaded, dropped) = parse(&bytes).unwrap(); + assert_eq!(dropped, 0); + assert_eq!(loaded.actors, [actor.clone()]); + assert_ne!(loaded.doc.get_actor(), &actor); + assert!(loaded.doc.get(ROOT, "b").unwrap().is_some()); + + // A record cut short by a crash is dropped. + let mut cut = bytes.clone(); + cut.extend(record(b"whatever")[..6].iter()); + let (loaded, dropped) = parse(&cut).unwrap(); + assert_eq!(dropped, 6); + assert!(loaded.doc.get(ROOT, "b").unwrap().is_some()); + + assert!(parse(b"something else").is_err()); + assert!(parse(&bytes[..MAGIC.len() + 3]).is_err()); + } +} diff --git a/src/document/mod.rs b/src/document/mod.rs index ce24815..0da1d67 100644 --- a/src/document/mod.rs +++ b/src/document/mod.rs @@ -1,5 +1,6 @@ //! The replicated document: the replica with its epoch, and the store semantics //! (merged-view paths over JSON) the HTTP API and remakes work with. +pub(crate) mod file; pub mod replica; pub mod store; diff --git a/src/files.rs b/src/files.rs index 81c8b1e..7bf9117 100644 --- a/src/files.rs +++ b/src/files.rs @@ -1,9 +1,12 @@ //! The files this node reads and writes for itself, and how much it did (`/admin/stats`, -//! `disk`): the state file (`swarm::state_file`) and a backup to recover from -//! (`swarm::bootstrap`). Not the Mycelium daemon's files, configuration or logs. +//! `disk`): the state file (`swarm::state_file`), the document file +//! (`document::file`) and a backup to recover from (`swarm::bootstrap`). Not the +//! Mycelium daemon's files, configuration or logs. //! //! Writes replace a file atomically: a temporary file next to it, flushed to disk, then -//! renamed over it, so a crash leaves the old content or the new one. +//! renamed over it, so a crash leaves the old content or the new one. Appends add to the +//! end and are flushed to disk before they count as done; a crash may leave the last one +//! cut short, which the file's reader has to expect. use std::io::Write; use std::path::{Path, PathBuf}; @@ -13,15 +16,17 @@ use std::sync::{Mutex, MutexGuard}; #[derive(Clone, Copy, Debug, PartialEq, Eq)] pub(crate) enum FileKind { State, + Document, Backup, } impl FileKind { - pub(crate) const ALL: [FileKind; 2] = [FileKind::State, FileKind::Backup]; + pub(crate) const ALL: [FileKind; 3] = [FileKind::State, FileKind::Document, FileKind::Backup]; pub(crate) fn as_str(self) -> &'static str { match self { FileKind::State => "state", + FileKind::Document => "document", FileKind::Backup => "backup", } } @@ -36,9 +41,12 @@ impl FileKind { pub(crate) struct Counts { pub(crate) reads: u64, pub(crate) read_bytes: u64, + /// Whole files written. pub(crate) writes: u64, pub(crate) written_bytes: u64, - /// Reads or writes that failed (a missing file read is no failure). + pub(crate) appends: u64, + pub(crate) appended_bytes: u64, + /// Reads, writes or appends that failed (a missing file read is no failure). pub(crate) errors: u64, } @@ -96,6 +104,23 @@ impl Files { result } + /// Adds `bytes` to the end of the file (which must exist), flushed to disk. + pub(crate) fn append(&self, kind: FileKind, path: &Path, bytes: &[u8]) -> std::io::Result<()> { + let result = (|| { + let mut file = std::fs::OpenOptions::new().append(true).open(path)?; + file.write_all(bytes)?; + file.sync_data() + })(); + self.count(kind, |counts| match &result { + Ok(()) => { + counts.appends += 1; + counts.appended_bytes += bytes.len() as u64; + } + Err(_) => counts.errors += 1, + }); + result + } + /// Moves an unusable file aside (``), returning where to. pub(crate) fn move_aside(&self, path: &Path, suffix: &str) -> std::io::Result { let aside = with_suffix(path, suffix); @@ -167,6 +192,12 @@ mod tests { assert_eq!(kind, FileKind::State); assert_eq!((state.writes, state.written_bytes), (2, 8)); assert_eq!((state.reads, state.read_bytes, state.errors), (1, 5, 0)); + files.append(FileKind::State, &path, b"!").unwrap(); + assert_eq!( + files.read(FileKind::State, &path).unwrap().as_deref(), + Some(&b"three!"[..]) + ); + assert_eq!(files.counts()[0].1.appended_bytes, 1); assert_eq!(files.counts()[1].1.reads, 0); std::fs::remove_dir_all(&dir).unwrap(); diff --git a/src/main.rs b/src/main.rs index 278a0f4..0a00f1b 100644 --- a/src/main.rs +++ b/src/main.rs @@ -9,7 +9,7 @@ use std::sync::Arc; use std::sync::atomic::AtomicBool; use std::time::Duration; -use anyhow::{Context, Result, bail}; +use anyhow::{Context, Result}; use automerge::AutoCommit; use tokio::sync::{RwLock, broadcast, mpsc, watch}; use tracing::{debug, error, info, trace, warn}; @@ -122,32 +122,27 @@ async fn main() -> Result<()> { let (change_tx, _) = watch::channel(0_u64); let (change_events, _) = broadcast::channel(256); - let recovered = match &args.recover_from { - Some(path) => Some(bootstrap::load_backup(path, &files)?), - None => None, - }; - if let (Some(backup), Some(remembered)) = (&recovered, &remembered) { - if backup.swarm_id != remembered.swarm_id { - if !force.allows("swarmId") { - bail!( - "The backup is of swarm {}, but the state file remembers swarm {}; add --force swarmId to recover the backup's swarm anyway", - backup.swarm_id, - remembered.swarm_id - ); - } - warn!( - backup = %backup.swarm_id, - remembered = %remembered.swarm_id, - "The backup is of another swarm than the remembered one; recovering the backup's (--force swarmId)" - ); - } - } - let (initial_doc, recovered_epoch, recovered_swarm) = match recovered { - Some(backup) => (backup.doc, Some(backup.epoch), Some(backup.swarm_id)), - None => (AutoCommit::new(), None, None), + let starting = bootstrap::starting_document( + &bootstrap::StartOptions { + recover_from: args.recover_from.as_deref(), + document_file: args.document_file.as_deref(), + genesis: args.genesis, + remembered: remembered.as_ref().map(|r| r.swarm_id.as_str()), + }, + &files, + &force, + )?; + let (initial_doc, starting_epoch, starting_swarm, earlier_actors) = match starting { + Some(starting) => ( + starting.doc, + Some(starting.epoch), + Some((starting.swarm_id, starting.origin)), + starting.earlier_actors, + ), + None => (AutoCommit::new(), None, None, Vec::new()), }; let initial_actor = initial_doc.get_actor().clone(); - let initial_epoch = match (recovered_epoch, args.debug_epoch) { + let initial_epoch = match (starting_epoch, args.debug_epoch) { (Some(epoch), _) => epoch, (None, Some(generation)) => Epoch { generation, @@ -197,6 +192,14 @@ async fn main() -> Result<()> { blobs: Arc::new(swarm::blob::Blobs::new(args.blob_chunk_kib * 1024)), protocol_writers: Arc::new(args.protocol_writers.iter().cloned().collect()), state_file, + document_file: args.document_file.clone().map(|path| { + Arc::new(document::file::DocumentFile::new( + path, + Arc::clone(&files), + Duration::from_secs(args.document_append_secs), + Duration::from_secs(args.document_rewrite_secs), + )) + }), files, config: Arc::new(config.report()), neighbours: Default::default(), @@ -211,14 +214,17 @@ async fn main() -> Result<()> { }; trace!("Application state initialized"); + for actor in earlier_actors { + state.remake.add_own_actor(actor); + } state.remake.add_own_actor(initial_actor); - // Join the remembered swarm and no other; HELLOs claim it. A backup to recover from + // Join the remembered swarm and no other; HELLOs claim it. A document to start on // names its swarm itself (checked against the remembered one above). - let start = match recovered_swarm { - Some(swarm_id) => { + let start = match starting_swarm { + Some((swarm_id, origin)) => { state.set_expected_swarm_id(swarm_id).await?; - bootstrap::Start::Recovered + bootstrap::Start::FromDocument(origin) } None => { if let Some(remembered) = remembered { @@ -283,6 +289,7 @@ async fn main() -> Result<()> { "Bootstrap completed" ); swarm::state_file::update(&state, true).await; + tokio::spawn(document::file::run(state.clone())); let auth_state = AuthState::new(&args.htpasswd); let app = api::router(state.clone(), auth_state, args.debug_api); @@ -328,6 +335,7 @@ async fn main() -> Result<()> { info!("HTTP API server stopped"); swarm::state_file::update(&stopping, true).await; + document::file::flush(&stopping).await; Ok(()) } diff --git a/src/state.rs b/src/state.rs index 15652ab..ce47673 100644 --- a/src/state.rs +++ b/src/state.rs @@ -604,6 +604,8 @@ pub(crate) struct AppState { pub(crate) protocol_writers: Arc>, /// This node's own files, and how much it read and wrote them. pub(crate) files: Arc, + /// This node's document on disk (`--document-file`), if configured. + pub(crate) document_file: Option>, /// What this node remembers across restarts (`--state-file`), if configured. pub(crate) state_file: Option>, /// The effective configuration with sources, imported values left out diff --git a/src/swarm/bootstrap.rs b/src/swarm/bootstrap.rs index f41a00b..bd15a34 100644 --- a/src/swarm/bootstrap.rs +++ b/src/swarm/bootstrap.rs @@ -24,9 +24,11 @@ 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}; @@ -45,16 +47,21 @@ pub(crate) enum Start { Join, /// Start a swarm, empty (`--genesis`). Genesis, - /// Start a swarm from a backup already loaded as the document (`--recover-from`). - Recovered, + /// 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 start == Start::Recovered { + if let Start::FromDocument(origin) = start { let Some(swarm_id) = state.swarm_id().await else { - bail!("recovering without a swarm ID"); + bail!("starting on a document without a swarm ID"); }; - info!(swarm_id = %swarm_id, epoch = %state.local_epoch(), "Recovered the swarm from a backup"); + 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 @@ -143,38 +150,154 @@ fn read_swarm_id(doc: &AutoCommit) -> Option { } } -/// A backup to recover from (`--recover-from`). -pub(crate) struct Backup { +/// 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, } -/// Loads a backup to recover from, as `GET /admin/document` saves it. -pub(crate) fn load_backup(path: &Path, files: &Files) -> Result { - 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()))?; - 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 { +/// 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> { + 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 backup {} names no swarm (no _continuum/swarm); it predates swarm IDs in documents", - path.display() + "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, +) -> Result { + 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!("The backup {}: {reason}", path.display()))?; - Ok(Backup { + 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 { + 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. -- cgit v1.3.1