diff options
39 files changed, 3646 insertions, 364 deletions
diff --git a/src/api/mod.rs b/src/api/mod.rs index 1fba48b..affb9ef 100644 --- a/src/api/mod.rs +++ b/src/api/mod.rs @@ -120,8 +120,11 @@ fn parse_write_path( if path.first().map(String::as_str) != Some(crate::remake::snapshot::META_ROOT) { return Ok(path); } - let epochs = path.len() < 2 || path[1] == crate::remake::snapshot::EPOCHS; - if epochs || !state.protocol_writers.contains(&user.0) { + // `_continuum` itself, epoch metadata and the swarm ID: nobody. + let fixed = path.len() < 2 + || path[1] == crate::remake::snapshot::EPOCHS + || path[1] == crate::remake::snapshot::SWARM; + if fixed || !state.protocol_writers.contains(&user.0) { warn!(user = %user.0, method, path = %key, "Refused a write to protocol data"); return Err(StatusCode::FORBIDDEN); } diff --git a/src/api/peers.rs b/src/api/peers.rs index 919f187..dc4bd3b 100644 --- a/src/api/peers.rs +++ b/src/api/peers.rs @@ -100,8 +100,8 @@ pub(crate) struct PeerListResponse { #[derive(Debug, Deserialize)] pub(crate) struct DeletePeerQuery { - #[serde(default)] - force: bool, + /// The checks to override (`force`): `alive`. + force: Option<String>, } #[utoipa::path( @@ -197,12 +197,12 @@ pub(crate) async fn list_peers( tag = "Peers", params( ("peer" = String, Path, description = "Peer public key"), - ("force" = Option<bool>, Query, description = "Also forget a peer that is currently alive; it returns on its next message") + ("force" = Option<String>, Query, description = "`true` or `alive`: also forget a peer that is currently alive; it returns on its next message") ), responses( (status = 204, description = "Swarm-wide forget recorded"), (status = 400, description = "Not a public key, or the local node's own key"), - (status = 409, description = "Peer is alive; pass force=true"), + (status = 409, description = "Peer is alive; pass force=alive"), (status = 503, description = "Node is not ready") ) )] @@ -217,6 +217,8 @@ pub(crate) async fn delete_peer( if peer.is_empty() || peer == state.local_id() { return Err(StatusCode::BAD_REQUEST); } + let force = crate::force::Force::parse(query.force.as_deref(), &["alive"]) + .map_err(|_| StatusCode::BAD_REQUEST)?; // Unknown locally is fine: other nodes may still know the peer. if let Some(peer_state_mutex) = state.get_peer_state(&peer).await { @@ -224,7 +226,7 @@ pub(crate) async fn delete_peer( .lock() .await .is_alive(&state.timing, Instant::now()) - && !query.force + && !force.allows("alive") { return Err(StatusCode::CONFLICT); } diff --git a/src/api/stats.rs b/src/api/stats.rs index 838cc0f..fa11b45 100644 --- a/src/api/stats.rs +++ b/src/api/stats.rs @@ -9,6 +9,8 @@ use axum::{ http::{HeaderMap, header}, response::{IntoResponse, Response}, }; +use std::collections::BTreeMap; + use serde::Serialize; use super::peers::EpochInfo; @@ -35,6 +37,21 @@ pub(crate) struct StatsResponse { /// The peers listed in /peers together (a forgotten peer's are gone). incidents: Incidents, propagation: PropagationStats, + /// This node's own files (`files`), by kind. + disk: BTreeMap<&'static str, DiskStats>, +} + +/// Reads and writes of one kind of file, since the node started. +#[derive(Debug, Serialize)] +#[serde(rename_all = "camelCase")] +struct DiskStats { + reads: u64, + read_bytes: u64, + writes: u64, + written_bytes: u64, + appends: u64, + appended_bytes: u64, + errors: u64, } /// How long changes written elsewhere took from their commit there until applied here, @@ -195,5 +212,22 @@ async fn collect(state: &AppState) -> StatsResponse { .map(|(le_ms, count)| PropagationBucket { le_ms, count }) .collect(), }, + disk: state + .files + .counts() + .into_iter() + .map(|(kind, counts)| { + let stats = DiskStats { + reads: counts.reads, + 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) + }) + .collect(), } } diff --git a/src/api/stats/prometheus.rs b/src/api/stats/prometheus.rs index f68ca85..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,6 +336,16 @@ pub(super) async fn render(state: &AppState, stats: &StatsResponse) -> String { stats.stale_hello_replies 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); + } + } + let propagation = &stats.propagation; out.family( "propagation_delay_seconds", diff --git a/src/config/import.rs b/src/config/import.rs index 895204a..4602705 100644 --- a/src/config/import.rs +++ b/src/config/import.rs @@ -3,7 +3,10 @@ //! //! * `env:NAME`: an environment variable; //! * `file:PATH`: a file's contents; -//! * `cmd:COMMAND`: the standard output of a shell command (`sh -c`). +//! * `cmd:COMMAND`: the standard output of a shell command (`sh -c`); +//! * `template:TEXT`: text with importers in it, `${env:NAME}`, `${file:PATH}` or +//! `${cmd:COMMAND}`, each replaced by its content (`$$` is a `$`; a placeholder ends +//! at the first `}`), e.g. `template:node-${env:HOSTNAME}`. //! //! The content is text (trailing line breaks removed), converted like a value written //! on the command line; with `+json` (`env+json:NAME`, …) it is parsed as JSON instead. @@ -11,8 +14,9 @@ //! On the command line, a value with one of these prefixes is imported; `literal:` //! passes the rest as is. In configuration files, an importer is a table: //! `{ env = "NAME" }`, `{ file = "PATH" }`, `{ command = "COMMAND" }` (or an argument -//! list, run without a shell), each with an optional `format = "json"`. Relative paths -//! are relative to the file, and commands run in its directory. +//! list, run without a shell), `{ template = "TEXT" }`, each with an optional +//! `format = "json"`. Relative paths are relative to the file, and commands run in its +//! directory, also inside a template. use std::path::{Path, PathBuf}; use std::process::Command; @@ -38,6 +42,10 @@ pub(crate) enum Source { argv: Vec<String>, dir: Option<PathBuf>, }, + Template { + template: String, + dir: Option<PathBuf>, + }, } #[derive(Debug, Clone, PartialEq, Eq)] @@ -73,6 +81,10 @@ pub(crate) fn parse_argument(raw: &str) -> Raw<'_> { command: rest.to_owned(), dir: None, }, + "template" => Source::Template { + template: rest.to_owned(), + dir: None, + }, _ => return Raw::Literal(raw), }; Raw::Import(Importer { source, format }) @@ -83,7 +95,7 @@ pub(crate) fn parse_table( object: &JsonMap<String, JsonValue>, dir: &Path, ) -> Result<Option<Importer>> { - let sources = ["env", "file", "command"] + let sources = ["env", "file", "command", "template"] .into_iter() .filter(|key| object.contains_key(*key)) .count(); @@ -91,11 +103,11 @@ pub(crate) fn parse_table( return Ok(None); } if sources > 1 { - bail!("an importer takes exactly one of env, file and command"); + bail!("an importer takes exactly one of env, file, command and template"); } if let Some(unknown) = object .keys() - .find(|key| !["env", "file", "command", "format"].contains(&key.as_str())) + .find(|key| !["env", "file", "command", "template", "format"].contains(&key.as_str())) { bail!("unknown importer key {unknown:?}"); } @@ -115,6 +127,11 @@ pub(crate) fn parse_table( Source::Env(text("env")?) } else if object.contains_key("file") { Source::File(dir.join(text("file")?)) + } else if object.contains_key("template") { + Source::Template { + template: text("template")?, + dir: Some(dir.to_owned()), + } } else { match &object["command"] { JsonValue::String(command) => Source::Shell { @@ -153,6 +170,7 @@ impl Importer { Source::File(path) => format!("file{format}:{}", path.display()), Source::Shell { command, .. } => format!("cmd{format}:{command}"), Source::Argv { argv, .. } => format!("cmd{format}:{}", argv.join(" ")), + Source::Template { template, .. } => format!("template{format}:{template}"), } } @@ -174,6 +192,7 @@ impl Importer { process.args(&argv[1..]); run(process, dir.as_deref(), &argv.join(" "))? } + Source::Template { template, dir } => expand(template, dir.as_deref())?, }; let text = text.trim_end_matches(['\n', '\r']); Ok(match self.format { @@ -192,6 +211,55 @@ pub(crate) enum Imported { Json(JsonValue), } +/// A template with its placeholders replaced by their content. +fn expand(template: &str, dir: Option<&Path>) -> Result<String> { + let mut out = String::new(); + let mut rest = template; + while let Some(at) = rest.find('$') { + out.push_str(&rest[..at]); + let after = &rest[at + 1..]; + if let Some(after) = after.strip_prefix('$') { + out.push('$'); + rest = after; + } else if let Some(after) = after.strip_prefix('{') { + let end = after + .find('}') + .ok_or_else(|| anyhow!("unterminated placeholder in template {template:?}"))?; + let placeholder = &after[..end]; + let Raw::Import(mut importer) = parse_argument(placeholder) else { + bail!( + "placeholder {placeholder:?} in template {template:?} isn't env:, file: or cmd:" + ); + }; + match &mut importer.source { + Source::File(path) => { + if let Some(dir) = dir { + *path = dir.join(&*path); + } + } + Source::Shell { dir: run_in, .. } => *run_in = dir.map(Path::to_owned), + Source::Template { .. } => bail!("templates don't nest: {placeholder:?}"), + Source::Env(_) | Source::Argv { .. } => {} + } + if importer.format != Format::Text { + bail!( + "placeholder {placeholder:?} must be text; +json applies to the whole template" + ); + } + match importer.fetch()? { + Imported::Text(text) => out.push_str(&text), + Imported::Json(_) => unreachable!("text importer"), + } + rest = &after[end + 1..]; + } else { + out.push('$'); + rest = after; + } + } + out.push_str(rest); + Ok(out) +} + fn run(mut process: Command, dir: Option<&Path>, shown: &str) -> Result<String> { if let Some(dir) = dir { process.current_dir(dir); @@ -240,6 +308,39 @@ mod tests { }) ); assert_eq!(parse_argument("env+yaml:X"), Raw::Literal("env+yaml:X")); + assert_eq!( + parse_argument("template:node-${env:HOST}"), + Raw::Import(Importer { + source: Source::Template { + template: "node-${env:HOST}".into(), + dir: None + }, + format: Format::Text + }) + ); + } + + #[test] + fn templates() { + let expanded = |template: &str| expand(template, None); + assert_eq!(expanded("a-${cmd:echo b}-c $$5 $x").unwrap(), "a-b-c $5 $x"); + assert_eq!( + expanded("${cmd:printf one}${cmd:printf two}").unwrap(), + "onetwo" + ); + assert!(expanded("${cmd:echo b").is_err()); + assert!(expanded("${plain}").is_err()); + assert!(expanded("${env+json:X}").is_err()); + assert!(expanded("${template:x}").is_err()); + + let json = Importer { + source: Source::Template { + template: "[\"${cmd:printf a}\", 1]".into(), + dir: None, + }, + format: Format::Json, + }; + assert_eq!(json.fetch().unwrap(), Imported::Json(json!(["a", 1]))); } #[test] diff --git a/src/config/mod.rs b/src/config/mod.rs index ef28542..538689e 100644 --- a/src/config/mod.rs +++ b/src/config/mod.rs @@ -65,6 +65,9 @@ const fn option( } } +/// What `--force` may override at start (`force`). +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] = &[ option( @@ -86,7 +89,7 @@ pub(crate) const OPTIONS: &[OptionSpec] = &[ Kind::List, "USER,[USER...]", None, - "HTTP API users (from htpasswd) who may write protocol data under /store/_continuum (remake records, members, forget records), e.g. to repair a stuck remake. Such writes act on the whole swarm and are logged. Epoch metadata is never writable.", + "HTTP API users (from htpasswd) who may write protocol data under /store/_continuum (remake records, members, forget records), e.g. to repair a stuck remake. Such writes act on the whole swarm and are logged. Epoch metadata and the swarm ID are never writable.", ), option( "debug-api", @@ -100,14 +103,70 @@ pub(crate) const OPTIONS: &[OptionSpec] = &[ Kind::List, "STRING,[STRING...]", None, - "How to reach a node of the swarm to join: a Mycelium public key, a Mycelium IPv6 address, or a host name with an AAAA record for one. Without seeds, the node starts a new swarm (genesis).", + "How to reach a node of the swarm to join: a Mycelium public key, a Mycelium IPv6 address, or a host name with an AAAA record for one. A node keeps trying until it has joined.", + ), + option( + "genesis", + Kind::Flag, + "BOOL", + Some("false"), + "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. Can't be combined with seeds. 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( + "force", + Kind::List, + "NAME,[NAME...]", + None, + "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", + 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( + "remake", + Kind::Text, + "KIND[:PATH]", + None, + "With --genesis: make the new swarm's document as a remake of this kind would, instead of starting it empty: import:PATH (an Automerge document, e.g. a backup, without its history and protocol data) or replace:PATH (a JSON document, the store root); compact changes nothing.", + ), + 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, + "PATH", + None, + "File where this node remembers its swarm across restarts: the swarm ID and some members to rejoin through (not the document). Written once ready, then rarely. Without it, a restarted node knows only its configured seeds.", ), option( "swarm-key-file", Kind::Text, "PATH", None, - "Pre-shared swarm key: only nodes that know it may join (raw bytes, at least 16; e.g. `head -c 32 /dev/urandom`). Without it, anyone who knows the swarm ID can.", + "Pre-shared swarm key: only nodes that know it may join (raw bytes, at least 16; e.g. `head -c 32 /dev/urandom`). Without it, any node that can reach a member may join.", ), option( "alias", @@ -213,6 +272,18 @@ pub(crate) struct Settings { pub(crate) debug_api: bool, #[serde(default)] pub(crate) seeds: Vec<String>, + #[serde(default)] + pub(crate) genesis: bool, + pub(crate) state_file: Option<PathBuf>, + pub(crate) document_file: Option<PathBuf>, + #[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<PathBuf>, + pub(crate) remake: Option<String>, + #[serde(default)] + pub(crate) force: Vec<String>, pub(crate) swarm_key_file: Option<PathBuf>, pub(crate) alias: Option<String>, #[serde(default = "default_keepalive_secs")] @@ -236,6 +307,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 } @@ -286,12 +363,29 @@ impl Settings { ); } } + crate::force::Force::parse(self.force.iter().map(String::as_str), FORCE) + .map_err(|reason| anyhow!("force: {reason}"))?; + // Contradictions, not overrides: an error wherever the options come from. + if self.genesis && !self.seeds.is_empty() { + bail!( + "genesis starts a new swarm, so there are no seeds to join through; unset seeds (--unset seeds) or genesis" + ); + } + if self.remake.is_some() && !self.genesis { + bail!("remake makes a new swarm's document; it needs genesis"); + } + if self.genesis && self.recover_from.is_some() { + bail!("recover-from starts the swarm from a backup; it can't be combined with genesis"); + } if self.debug_epoch.is_some() && !self.debug_api { bail!("debug-epoch needs debug-api"); } 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<String>, +} + +/// 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<ActorId>, +} + +/// 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<Option<Loaded>> { + 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::<Result<Vec<_>>>()?; + + let mut records = &rest[line_end + 1..]; + let mut doc: Option<AutoCommit> = 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<u8> { + 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<Vec<u8>> { + 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<Files>, + append_every: Duration, + rewrite_every: Duration, + // A std mutex: never held across an await. + written: Mutex<Option<Written>>, +} + +#[derive(Debug, Clone)] +struct Written { + epoch: Epoch, + /// This node's actors when the file was last rewritten (`RemakeState::own_actors`). + actors: usize, + heads: Vec<ChangeHash>, + /// The full save. + base_bytes: u64, + /// Appended since. + tail_bytes: u64, + rewritten_at: Instant, +} + +impl DocumentFile { + pub(crate) fn new( + path: PathBuf, + files: Arc<Files>, + append_every: Duration, + rewrite_every: Duration, + ) -> Self { + Self { + path, + files, + append_every, + rewrite_every, + written: Mutex::new(None), + } + } + + fn lock(&self) -> MutexGuard<'_, Option<Written>> { + 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<DocumentFile>, own: Vec<ActorId>, 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<ChangeHash>, 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<ActorId> = copy + .get_changes(&[]) + .iter() + .map(|change| change.actor_id().clone()) + .collect(); + let current = copy.get_actor().clone(); + let actors: Vec<ActorId> = 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<DocumentFile>, 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/document/replica.rs b/src/document/replica.rs index e24f93c..db47223 100644 --- a/src/document/replica.rs +++ b/src/document/replica.rs @@ -86,6 +86,9 @@ impl fmt::Debug for Epoch { /// [`Replica::heads`]. pub struct Replica { pub epoch: Epoch, + /// How many times this node changed epoch (switch, rejoin, abandoned rejoin) in this + /// run: messages carry it, so a late one can't flip a peer's view of our epoch back. + pub epoch_seq: u32, pub doc: AutoCommit, heads: Vec<ChangeHash>, } @@ -94,6 +97,7 @@ impl Replica { pub fn new(epoch: Epoch, doc: AutoCommit) -> Self { let mut replica = Self { epoch, + epoch_seq: 0, doc, heads: Vec::new(), }; diff --git a/src/files.rs b/src/files.rs new file mode 100644 index 0000000..58fd043 --- /dev/null +++ b/src/files.rs @@ -0,0 +1,213 @@ +//! The files this node reads and writes for itself, and how much it did (`/admin/stats`, +//! `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. 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}; +use std::sync::{Mutex, MutexGuard}; + +/// What a file is, for the counts. +#[derive(Clone, Copy, Debug, PartialEq, Eq)] +pub(crate) enum FileKind { + State, + Document, + Backup, + /// The input of a genesis remake (`--genesis --remake`). + Remake, +} + +impl FileKind { + pub(crate) const ALL: [FileKind; 4] = [ + FileKind::State, + FileKind::Document, + FileKind::Backup, + FileKind::Remake, + ]; + + pub(crate) fn as_str(self) -> &'static str { + match self { + FileKind::State => "state", + FileKind::Document => "document", + FileKind::Backup => "backup", + FileKind::Remake => "remake", + } + } + + fn index(self) -> usize { + self as usize + } +} + +/// Counts for one kind of file, since the node started. +#[derive(Clone, Copy, Debug, Default)] +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, + 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, +} + +/// Reads and writes this node's files, counting per kind. +#[derive(Default)] +pub(crate) struct Files { + // A std mutex: every access is short and never spans an await. + counts: Mutex<[Counts; FileKind::ALL.len()]>, +} + +impl Files { + fn lock(&self) -> MutexGuard<'_, [Counts; FileKind::ALL.len()]> { + self.counts + .lock() + .unwrap_or_else(|poisoned| poisoned.into_inner()) + } + + fn count(&self, kind: FileKind, f: impl FnOnce(&mut Counts)) { + f(&mut self.lock()[kind.index()]); + } + + /// The file's content; None if it doesn't exist. + pub(crate) fn read(&self, kind: FileKind, path: &Path) -> std::io::Result<Option<Vec<u8>>> { + match std::fs::read(path) { + Ok(bytes) => { + self.count(kind, |counts| { + counts.reads += 1; + counts.read_bytes += bytes.len() as u64; + }); + Ok(Some(bytes)) + } + Err(error) if error.kind() == std::io::ErrorKind::NotFound => Ok(None), + Err(error) => { + self.count(kind, |counts| counts.errors += 1); + Err(error) + } + } + } + + /// Replaces the file's content atomically (see the module docs). + pub(crate) fn write_atomic( + &self, + kind: FileKind, + path: &Path, + bytes: &[u8], + ) -> std::io::Result<()> { + let result = write_atomic(path, bytes); + self.count(kind, |counts| match &result { + Ok(()) => { + counts.writes += 1; + counts.written_bytes += bytes.len() as u64; + } + Err(_) => counts.errors += 1, + }); + 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 (`<path><suffix>`), returning where to. + pub(crate) fn move_aside(&self, path: &Path, suffix: &str) -> std::io::Result<PathBuf> { + let aside = with_suffix(path, suffix); + std::fs::rename(path, &aside)?; + Ok(aside) + } + + /// The counts per kind. + pub(crate) fn counts(&self) -> Vec<(FileKind, Counts)> { + let counts = *self.lock(); + FileKind::ALL + .into_iter() + .map(|kind| (kind, counts[kind.index()])) + .collect() + } +} + +fn write_atomic(path: &Path, bytes: &[u8]) -> std::io::Result<()> { + let temporary = with_suffix(path, ".tmp"); + { + let mut file = std::fs::File::create(&temporary)?; + file.write_all(bytes)?; + file.sync_all()?; + } + std::fs::rename(&temporary, path)?; + sync_parent(path); + Ok(()) +} + +/// Makes a rename in `path`'s directory durable (best effort: not every platform syncs +/// directories). +fn sync_parent(path: &Path) { + if let Some(dir) = path.parent().filter(|dir| !dir.as_os_str().is_empty()) { + if let Ok(dir) = std::fs::File::open(dir) { + let _ = dir.sync_all(); + } + } +} + +fn with_suffix(path: &Path, suffix: &str) -> PathBuf { + let mut name = path.as_os_str().to_owned(); + name.push(suffix); + PathBuf::from(name) +} + +#[cfg(test)] +mod tests { + use super::*; + + #[test] + fn writes_reads_and_counts() { + let dir = std::env::temp_dir().join(format!("continuum-files-{}", std::process::id())); + std::fs::create_dir_all(&dir).unwrap(); + let path = dir.join("state.json"); + let files = Files::default(); + + assert_eq!(files.read(FileKind::State, &path).unwrap(), None); + files.write_atomic(FileKind::State, &path, b"one").unwrap(); + files + .write_atomic(FileKind::State, &path, b"three") + .unwrap(); + assert_eq!( + files.read(FileKind::State, &path).unwrap().as_deref(), + Some(&b"three"[..]) + ); + assert!(!with_suffix(&path, ".tmp").exists()); + + let (kind, state) = files.counts()[0]; + 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/force.rs b/src/force.rs new file mode 100644 index 0000000..40f0ffc --- /dev/null +++ b/src/force.rs @@ -0,0 +1,72 @@ +//! `force`: overriding checks that otherwise refuse. Conflicts are errors by default; +//! each check that can be overridden has a name, and `force` lists the names to +//! override, or is `true` for all of them (`false`, or nothing, for none). +//! +//! The same format everywhere: the `?force=` query parameter of the API and the +//! `--force` option. Names are camelCase, like the API's fields (`switchDelaySecs`), +//! on the command line too. An unknown name is an error, so a typo can't silently +//! override nothing. + +/// The checks a `force` value overrides. +#[derive(Debug, Clone, Default, PartialEq, Eq)] +pub(crate) struct Force { + all: bool, + names: Vec<&'static str>, +} + +impl Force { + /// Parses `values` (each `true`, `false`, or a comma-separated list of names), + /// allowing only the names in `known`. + pub(crate) fn parse<'a>( + values: impl IntoIterator<Item = &'a str>, + known: &[&'static str], + ) -> Result<Self, String> { + let mut force = Force::default(); + for value in values { + for name in value.split(',').map(str::trim) { + match name { + "" | "false" => {} + "true" => force.all = true, + name => match known.iter().find(|known| **known == name) { + Some(known) => force.names.push(known), + None => { + return Err(format!( + "unknown force {name:?}: use true, or a list of {}", + known.join(", ") + )); + } + }, + } + } + } + Ok(force) + } + + /// Whether the check `name` is overridden. + pub(crate) fn allows(&self, name: &str) -> bool { + self.all || self.names.contains(&name) + } +} + +#[cfg(test)] +mod tests { + use super::*; + + const KNOWN: &[&str] = &["switchDelaySecs", "supersede"]; + + #[test] + fn all_some_or_none() { + let all = Force::parse(["true"], KNOWN).unwrap(); + assert!(all.allows("supersede") && all.allows("switchDelaySecs")); + + let some = Force::parse(["supersede"], KNOWN).unwrap(); + assert!(some.allows("supersede") && !some.allows("switchDelaySecs")); + let list = Force::parse(["switchDelaySecs, supersede"], KNOWN).unwrap(); + assert!(list.allows("supersede") && list.allows("switchDelaySecs")); + + for none in [vec![], vec!["false"], vec![""]] { + assert!(!Force::parse(none, KNOWN).unwrap().allows("supersede")); + } + assert!(Force::parse(["swarm-id"], KNOWN).is_err()); + } +} diff --git a/src/main.rs b/src/main.rs index 217a9a4..e494497 100644 --- a/src/main.rs +++ b/src/main.rs @@ -1,37 +1,37 @@ //! continuum: a replicated Automerge key-value store synchronized over Mycelium. //! -//! main.rs only wires things together: arguments, state construction, background -//! tasks, bootstrap and the HTTP server. +//! main.rs only wires things together: arguments, the starting document, the node +//! (`node`), bootstrap and the HTTP server. -use std::collections::HashMap; use std::future::IntoFuture; use std::sync::Arc; -use std::sync::atomic::AtomicBool; use std::time::Duration; use anyhow::{Context, Result}; use automerge::AutoCommit; -use tokio::sync::{RwLock, broadcast, mpsc, watch}; -use tracing::{debug, error, info, trace, warn}; +use tokio::sync::watch; +use tracing::{debug, error, info, warn}; mod api; mod backoff; mod config; mod document; +mod files; +mod force; mod logging; +mod node; mod remake; +#[cfg(test)] +mod simtest; mod state; mod swarm; mod transport; use api::auth::AuthState; -use document::replica::{Epoch, Replica}; -use state::{AppState, NodeLifecycle, Timing}; -use swarm::{bootstrap, peer, sync}; +use document::replica::Epoch; +use swarm::bootstrap; use transport::Transport; -use transport::firewall::{Firewall, FirewallTransport}; use transport::mycelium::{self, MyceliumSetup, MyceliumSpawnConfig}; -use transport::traffic::{CountingTransport, Traffic}; #[tokio::main] async fn main() -> Result<()> { @@ -48,7 +48,7 @@ async fn main() -> Result<()> { let swarm_key = match &args.swarm_key_file { Some(path) => Some(Arc::new(swarm::auth::SwarmKey::from_file(path)?)), None => { - warn!("No swarm key: any node that learns the swarm ID can join"); + warn!("No swarm key: any node that can reach a member may join"); None } }; @@ -86,124 +86,139 @@ async fn main() -> Result<()> { "Using Mycelium REST API" ); - let seeds: Vec<String> = args.seeds.clone(); + let force = force::Force::parse(args.force.iter().map(String::as_str), config::FORCE) + .map_err(anyhow::Error::msg)?; + let files = Arc::new(files::Files::default()); + let mut seeds: Vec<String> = args.seeds.clone(); + let (state_file, remembered) = match &args.state_file { + Some(path) => { + let (file, remembered) = + swarm::state_file::StateFile::open(path.clone(), Arc::clone(&files)); + (Some(Arc::new(file)), remembered) + } + None => (None, None), + }; + if let Some(remembered) = &remembered { + info!( + swarm_id = %remembered.swarm_id, + seeds = remembered.seeds.len(), + "Remembering a swarm from the state file" + ); + for seed in &remembered.seeds { + if !seeds.contains(seed) { + seeds.push(seed.clone()); + } + } + } else if args.genesis && state_file.is_none() { + warn!( + "--genesis without --state-file: every restart with --genesis starts another new swarm" + ); + } debug!(seed_count = seeds.len(), "Loaded bootstrap configuration"); - let (sync_tx, sync_rx) = mpsc::unbounded_channel(); - let (change_tx, _) = watch::channel(0_u64); - let (change_events, _) = broadcast::channel(256); - - let initial_epoch = match args.debug_epoch { - Some(generation) => Epoch { + 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 mut initial_doc = initial_doc; + if let Some(spec) = &args.remake { + remake::remake_genesis(spec, &files, &mut initial_doc) + .with_context(|| format!("Making the genesis document (--remake {spec})"))?; + info!(remake = %spec, "Made the genesis document"); + } + let initial_epoch = match (starting_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"); - let firewall = Arc::new(Firewall::default()); - let traffic = Arc::new(Traffic::default()); - let mycelium_transport = Arc::new(mycelium::MyceliumTransport::connect(mycelium_setup).await?); debug!(local_id = mycelium_transport.local_id(), "Transport ready"); - let state = AppState { - doc: Arc::new(RwLock::new(Replica::new(initial_epoch, AutoCommit::new()))), - epoch_tx, - lifecycle: Arc::new(RwLock::new(NodeLifecycle::Joining { - discovered_swarm_id: None, - })), - peer_states: Arc::new(RwLock::new(HashMap::new())), - // Counted outside the firewall: what it drops outbound counts as sent (it plays a - // lossy network), what it drops inbound never arrived. - transport: Arc::new(CountingTransport::new( - Arc::new(FirewallTransport::new( - mycelium_transport, - Arc::clone(&firewall), - )), - Arc::clone(&traffic), - )), - firewall, - traffic, - stats: Default::default(), - sync_events: sync_tx, - change_tx, - initial_sync_done: Arc::new(AtomicBool::new(false)), - change_events, - timing: Timing::from_keepalive(Duration::from_secs(args.keepalive_secs)), - remake: Arc::new(remake::RemakeState::default()), - swarm_key, - run_id: uuid::Uuid::new_v4().into_bytes(), + let 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), + )) + }); + let node = node::Node::new(node::NodeConfig { + transport: mycelium_transport, + initial_epoch, + initial_doc, + earlier_actors, + timing: state::Timing::from_keepalive(Duration::from_secs(args.keepalive_secs)), neighbours_k: args.neighbours, alias: args.alias.clone(), electorate_window: Duration::from_secs(args.electorate_window_days * 24 * 3600), - blobs: Arc::new(swarm::blob::Blobs::new(args.blob_chunk_kib * 1024)), - protocol_writers: Arc::new(args.protocol_writers.iter().cloned().collect()), - config: Arc::new(config.report()), - neighbours: Default::default(), - pulling: Default::default(), - announce_pending: Default::default(), - rtt: Default::default(), - throughput: Default::default(), - stability: Default::default(), - stale: Default::default(), - catch_up_source: Default::default(), - catch_up_preference: Default::default(), - }; - - trace!("Application state initialized"); - - let consumer_state = state.clone(); - tokio::spawn(async move { - let inner_state = consumer_state.clone(); - let _ = consumer_state - .transport - .start_consumer(Box::new(move |sender, msg, _payload_len| { - let state_clone = inner_state.clone(); - Box::pin(async move { peer::handle_message(&state_clone, sender, msg).await }) - })) - .await; - }); - trace!("Mycelium consumer task spawned"); - - let sync_state = state.clone(); - tokio::spawn(async move { - sync::run_sync_worker(sync_state, sync_rx).await; + blob_chunk_bytes: args.blob_chunk_kib * 1024, + swarm_key, + protocol_writers: args.protocol_writers.iter().cloned().collect(), + files, + state_file, + document_file, + config_report: config.report(), }); - trace!("Sync worker task spawned"); + let state = node.state.clone(); - let peer_worker_state = state.clone(); - tokio::spawn(async move { - peer::run_peer_worker(peer_worker_state).await; - }); - trace!("Peer worker task spawned"); + // 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 starting_swarm { + Some((swarm_id, origin)) => { + state.set_expected_swarm_id(swarm_id).await?; + bootstrap::Start::FromDocument(origin) + } + 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 + } + } + }; - // Switches to committed epochs and catches up with newer ones. Starts before - // bootstrapping: a node joining a compacted swarm can only finish its initial sync - // after fetching the current epoch from a peer. - tokio::spawn(remake::run_worker(state.clone())); - trace!("Remake worker task spawned"); + node.spawn_workers(); + let joining = bootstrap::bootstrap(&state, &seeds, start); tokio::select! { - joined = bootstrap::bootstrap(&state, &seeds) => joined.context("Joining swarm")?, + joined = joining => joined.context("Joining swarm")?, _ = stopped(stop.clone()) => { info!("Stopped while joining the swarm"); return Ok(()); } } - // Start the change publisher after bootstrapping: streams only exist from then on, - // and it avoids diffing the initial sync. - let publisher_state = state.clone(); - tokio::spawn(async move { sync::run_change_publisher(publisher_state).await }); + node::spawn_ready_workers(&state); info!( swarm_id = ?state.swarm_id().await, ready = state.is_ready().await, "Bootstrap completed" ); + swarm::state_file::update(&state, true).await; let auth_state = AuthState::new(&args.htpasswd); let app = api::router(state.clone(), auth_state, args.debug_api); @@ -222,6 +237,7 @@ async fn main() -> Result<()> { }); let server_fut = server_builder.into_future(); + let stopping = state.clone(); tokio::pin!(server_fut); tokio::select! { @@ -247,6 +263,8 @@ 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/node.rs b/src/node.rs new file mode 100644 index 0000000..57d961c --- /dev/null +++ b/src/node.rs @@ -0,0 +1,174 @@ +//! Building a node and starting its workers: what `main` does once it has read the +//! configuration and connected to Mycelium, and what the simulator (`transport::sim`) +//! does for every simulated node. Everything here is independent of where the +//! transport and the settings come from. + +use std::collections::{HashMap, HashSet}; +use std::sync::Arc; +use std::sync::atomic::AtomicBool; +use std::time::Duration; + +use automerge::{ActorId, AutoCommit}; +use tokio::sync::{RwLock, broadcast, mpsc, watch}; +use tracing::trace; + +use crate::document::file::DocumentFile; +use crate::document::replica::{Epoch, Replica}; +use crate::files::Files; +use crate::remake; +use crate::state::{AppState, NodeLifecycle, Timing}; +use crate::swarm::auth::SwarmKey; +use crate::swarm::state_file::StateFile; +use crate::swarm::sync::SyncEvent; +use crate::swarm::{self, peer, sync}; +use crate::transport::Transport; +use crate::transport::firewall::{Firewall, FirewallTransport}; +use crate::transport::traffic::{CountingTransport, Traffic}; + +/// What a node is built from. +pub(crate) struct NodeConfig { + /// The node's transport, without the counting and firewall layers (added here). + pub(crate) transport: Arc<dyn Transport>, + pub(crate) initial_epoch: Epoch, + pub(crate) initial_doc: AutoCommit, + /// Actors this node wrote with in earlier runs (`--document-file`). + pub(crate) earlier_actors: Vec<ActorId>, + /// `Timing::from_keepalive(--keepalive)`. + pub(crate) timing: Timing, + pub(crate) neighbours_k: usize, + pub(crate) alias: Option<String>, + pub(crate) electorate_window: Duration, + pub(crate) blob_chunk_bytes: u32, + pub(crate) swarm_key: Option<Arc<SwarmKey>>, + pub(crate) protocol_writers: HashSet<String>, + pub(crate) files: Arc<Files>, + pub(crate) state_file: Option<Arc<StateFile>>, + pub(crate) document_file: Option<Arc<DocumentFile>>, + /// For `GET /admin/config`. + pub(crate) config_report: serde_json::Value, +} + +/// A built node whose workers haven't started yet. +pub(crate) struct Node { + pub(crate) state: AppState, + sync_rx: mpsc::UnboundedReceiver<SyncEvent>, +} + +impl Node { + pub(crate) fn new(config: NodeConfig) -> Self { + let (sync_tx, sync_rx) = mpsc::unbounded_channel(); + let (change_tx, _) = watch::channel(0_u64); + let (change_events, _) = broadcast::channel(256); + let (epoch_tx, _) = watch::channel(config.initial_epoch); + let firewall = Arc::new(Firewall::default()); + let traffic = Arc::new(Traffic::default()); + let initial_actor = config.initial_doc.get_actor().clone(); + + let state = AppState { + doc: Arc::new(RwLock::new(Replica::new( + config.initial_epoch, + config.initial_doc, + ))), + epoch_tx, + lifecycle: Arc::new(RwLock::new(NodeLifecycle::Joining { + discovered_swarm_id: None, + })), + peer_states: Arc::new(RwLock::new(HashMap::new())), + // Counted outside the firewall: what it drops outbound counts as sent (it + // plays a lossy network), what it drops inbound never arrived. + transport: Arc::new(CountingTransport::new( + Arc::new(FirewallTransport::new( + config.transport, + Arc::clone(&firewall), + )), + Arc::clone(&traffic), + )), + firewall, + traffic, + stats: Default::default(), + sync_events: sync_tx, + change_tx, + initial_sync_done: Arc::new(AtomicBool::new(false)), + change_events, + timing: config.timing, + remake: Arc::new(remake::RemakeState::default()), + swarm_key: config.swarm_key, + run_id: uuid::Uuid::new_v4().into_bytes(), + neighbours_k: config.neighbours_k, + alias: config.alias, + electorate_window: config.electorate_window, + blobs: Arc::new(swarm::blob::Blobs::new(config.blob_chunk_bytes)), + protocol_writers: Arc::new(config.protocol_writers), + state_file: config.state_file, + document_file: config.document_file, + files: config.files, + config: Arc::new(config.config_report), + neighbours: Default::default(), + topology_changed: Default::default(), + pulling: Default::default(), + announce_pending: Default::default(), + rtt: Default::default(), + throughput: Default::default(), + stability: Default::default(), + stale: Default::default(), + catch_up_source: Default::default(), + catch_up_preference: Default::default(), + catch_up_avoided: Default::default(), + }; + for actor in config.earlier_actors { + state.remake.add_own_actor(actor); + } + state.remake.add_own_actor(initial_actor); + trace!("Application state initialized"); + Self { state, sync_rx } + } + + /// Starts what runs from the start: the message consumer and the sync, peer and + /// remake workers. The remake worker starts before bootstrapping: a node joining a + /// compacted swarm can only finish its initial sync after fetching the current epoch + /// from a peer. + pub(crate) fn spawn_workers(self) -> AppState { + let Self { state, sync_rx } = self; + + let consumer_state = state.clone(); + tokio::spawn(async move { + let inner_state = consumer_state.clone(); + let _ = consumer_state + .transport + .start_consumer(Box::new(move |sender, msg, _payload_len| { + let state_clone = inner_state.clone(); + Box::pin(async move { peer::handle_message(&state_clone, sender, msg).await }) + })) + .await; + }); + trace!("Message consumer task spawned"); + + let sync_state = state.clone(); + tokio::spawn(async move { + sync::run_sync_worker(sync_state, sync_rx).await; + }); + trace!("Sync worker task spawned"); + + let peer_worker_state = state.clone(); + tokio::spawn(async move { + peer::run_peer_worker(peer_worker_state).await; + }); + trace!("Peer worker task spawned"); + + tokio::spawn(remake::run_worker(state.clone())); + trace!("Remake worker task spawned"); + + tokio::spawn(swarm::topology::run_worker(state.clone())); + trace!("Topology worker task spawned"); + + state + } +} + +/// Starts what runs once bootstrap is done: the change publisher (streams only exist +/// from then on, and it avoids diffing the initial sync) and the document file writer. +pub(crate) fn spawn_ready_workers(state: &AppState) { + let publisher_state = state.clone(); + tokio::spawn(async move { sync::run_change_publisher(publisher_state).await }); + tokio::spawn(crate::document::file::run(state.clone())); +} diff --git a/src/remake/api.rs b/src/remake/api.rs index 826c81e..0428a43 100644 --- a/src/remake/api.rs +++ b/src/remake/api.rs @@ -17,7 +17,8 @@ //! can't commit it any more anyway (`coordinator`). `DELETE /admin/remakes` does the //! same for the undecided remake of the local epoch, without having to know its ID. //! -//! Request body for POST (all optional): +//! Request body for POST (all optional; `baseHeads` not for `compact`, whose base is +//! fixed when it commits): //! `{ "kind": "compact", "spec": null, "baseHeads": ["<hex>", …], //! "replayWindow": {"epochs": 1, "seconds": 604800}, //! "archiveRetention": {"epochs": 2, "seconds": 2592000}, @@ -72,6 +73,7 @@ use super::records::{ }; use super::{Host, kind, vote}; use crate::document::store::StoreError; +use crate::force::Force; #[derive(Debug, Default, Deserialize)] pub(crate) struct WaitQuery { @@ -79,12 +81,20 @@ pub(crate) struct WaitQuery { wait: bool, } +#[derive(Debug, Deserialize)] +pub(crate) struct AbortQuery { + #[serde(default)] + wait: bool, + /// The checks to override (`force`): `coordinatorActive`. + force: Option<String>, +} + #[derive(Debug, Default, Deserialize)] #[serde(rename_all = "camelCase")] pub(crate) struct StartQuery { #[serde(default)] wait: bool, - /// The checks to override (`Force`). + /// The checks to override (`force`). force: Option<String>, // Only for an import uploaded as `application/octet-stream`, which has no JSON // body to carry them: @@ -159,41 +169,63 @@ struct StartRequest { switch_delay_secs: Option<u64>, } -/// The checks `?force=` overrides: `true` all of them, or a comma-separated list. -#[derive(Debug, Default, Clone, Copy)] -struct Force { - switch_delay: bool, - supersede: bool, +/// What `?force=` may override on a proposal (`force`). +const FORCE: &[&str] = &["switchDelaySecs", "supersede", "coordinatorActive"]; +/// What `?force=` may override on an abort. +const ABORT_FORCE: &[&str] = &["coordinatorActive"]; + +/// When a node other than the coordinator may abort an expired remake: a whole grace +/// period after it expired. Its coordinator stops committing at expiry +/// (`coordinator::decide`), so an abort from elsewhere and a commit can only cross if +/// their clocks are more than a grace period apart. +fn abortable_at(record: &Record, grace: u64) -> u64 { + record + .proposal + .settle_until + .saturating_add(grace.saturating_mul(2)) } -impl Force { - fn parse(value: Option<&str>) -> Result<Self, String> { - let mut force = Force::default(); - match value.map(str::trim) { - None | Some("" | "false") => {} - Some("true") => { - force = Force { - switch_delay: true, - supersede: true, - }; - } - Some(list) => { - for name in list.split(',').map(str::trim) { - match name { - "switchDelaySecs" => force.switch_delay = true, - "supersede" => force.supersede = true, - other => { - return Err(format!( - "unknown force {other:?}: use true, or a list of \ - switchDelaySecs and supersede" - )); - } - } - } - } - } - Ok(force) +/// Why this node may not abort expired `record` now, and the `force` name that would +/// override it (none for one that only time cures). +fn abort_obstacle( + record: &Record, + me: &str, + active_peers: &HashSet<String>, + force: &Force, + now: u64, + grace: u64, +) -> Option<(String, Option<&'static str>)> { + let at = abortable_at(record, grace); + if now < at { + return Some(( + format!( + "only its coordinator may abort it until a grace period after it expired ({}s from now)", + (at - now).div_ceil(1000) + ), + None, + )); + } + let coordinator = &record.proposal.coordinator; + if coordinator != me && active_peers.contains(coordinator) && !force.allows("coordinatorActive") + { + return Some(( + format!( + "its coordinator {coordinator} is an active peer, so it is stuck rather than gone (look at its log); add force=coordinatorActive to abort it anyway" + ), + Some("coordinatorActive"), + )); } + None +} + +/// The active peers this node sees. +async fn active_peers<H: Host>(host: &H) -> HashSet<String> { + host.peer_summaries() + .await + .into_iter() + .filter(|peer| peer.active) + .map(|peer| peer.id) + .collect() } /// An expired remake of the epoch a proposal would supersede. @@ -201,11 +233,20 @@ impl Force { struct ExpiredRemake { id: String, coordinator: String, + coordinator_active: bool, /// When it expired (Unix milliseconds). since: u64, + /// When another node may abort it (`abortable_at`). + abortable_at: u64, tally: vote::Tally, } +/// Superseding expired remakes, as the proposal's `force` asks. +struct Supersede { + force: Force, + active_peers: HashSet<String>, +} + #[derive(Debug, Serialize)] #[serde(rename_all = "camelCase")] struct Report { @@ -222,9 +263,15 @@ struct Report { enum StartError { #[error("remake {0} is already open for this epoch")] Open(String), - /// Expired remakes still hold votes; superseding them takes `force=supersede`. - #[error("expired remakes of this epoch still hold votes")] - Expired(Vec<ExpiredRemake>), + /// Expired remakes still hold votes, and aren't superseded: not asked to + /// (`force=supersede`), or not possible yet (`abort_obstacle`). + #[error("{message}")] + Expired { + message: String, + remakes: Vec<ExpiredRemake>, + /// The `force` name that would override it, if any. + force: Option<&'static str>, + }, /// The proposer votes for its own remake, so it must be free to. #[error("this node can't vote for a new remake: {0}")] VoteHeld(String), @@ -332,7 +379,7 @@ async fn start_kind<H: Host>( if !host.is_node_ready().await { return error_response(StatusCode::SERVICE_UNAVAILABLE, "node is not ready", None); } - let force = match Force::parse(query.force.as_deref()) { + let force = match Force::parse(query.force.as_deref(), FORCE) { Ok(force) => force, Err(reason) => return error_response(StatusCode::BAD_REQUEST, reason, None), }; @@ -395,6 +442,15 @@ async fn start_kind<H: Host>( } }; + if kind.base_at_commit() && request.base_heads.is_some() { + return error_response( + StatusCode::BAD_REQUEST, + format!( + "a {kind_name} remake fixes its base when it commits; baseHeads is not accepted" + ), + None, + ); + } let base = match &request.base_heads { None => None, Some(heads) => { @@ -426,7 +482,7 @@ async fn start_kind<H: Host>( let keepalive = dormant_after / 4; let switch_delay = request.switch_delay_secs.map(Duration::from_secs); match switch_delay { - Some(delay) if delay < keepalive && force.switch_delay => { + Some(delay) if delay < keepalive && force.allows("switchDelaySecs") => { warn!( switch_delay_secs = delay.as_secs(), keepalive_secs = keepalive.as_secs(), @@ -454,7 +510,8 @@ async fn start_kind<H: Host>( let mut proposal = Proposal { coordinator: host.node_id(), from_epoch: host.current_epoch().into(), - base_heads: Vec::new(), + // Filled in when the proposal is written, unless the kind fixes it at the commit. + base_heads: None, policy: Policy { replay_window: request .replay_window @@ -491,7 +548,10 @@ async fn start_kind<H: Host>( id.clone(), proposal, base, - force.supersede, + Supersede { + force: force.clone(), + active_peers: active_peers(&host).await, + }, now, grace, )); @@ -516,33 +576,28 @@ async fn start_kind<H: Host>( Some(json!({ "remake": open })), ); } - Err(StartError::Expired(expired)) => { - let active: HashSet<String> = host - .peer_summaries() - .await - .into_iter() - .filter(|peer| peer.active) - .map(|peer| peer.id) - .collect(); - let me = host.node_id(); - let expired: Vec<JsonValue> = expired + Err(StartError::Expired { + message, + remakes, + force, + }) => { + let expired: Vec<JsonValue> = remakes .iter() .map(|remake| { json!({ "id": remake.id, "coordinator": remake.coordinator, - "coordinatorActive": remake.coordinator == me - || active.contains(&remake.coordinator), + "coordinatorActive": remake.coordinator_active, "expiredAt": remake.since, + "abortableAt": remake.abortable_at, "tally": remake.tally.to_string(), }) }) .collect(); return error_response( StatusCode::CONFLICT, - "an expired remake of this epoch still holds votes: abort it, or add \ - ?force=supersede to abort it and propose in its place", - Some(json!({ "expired": expired, "force": "supersede" })), + message, + Some(json!({ "expired": expired, "force": force })), ); } Err(error @ StartError::VoteHeld(_)) => { @@ -590,7 +645,7 @@ async fn propose<H: Host>( id: String, mut proposal: Proposal, base: Option<Vec<ChangeHash>>, - supersede: bool, + supersede: Supersede, now: u64, grace: u64, ) -> Result<Record, StartError> { @@ -598,6 +653,7 @@ async fn propose<H: Host>( let electorate_host = host.clone(); let me = host.node_id(); let started = host.remake().started_at() + grace; + let base_at_commit = kind::by_name(&proposal.kind).is_some_and(|kind| kind.base_at_commit()); let record = host .commit_doc(move |doc| -> Result<Record, StartError> { let mut all = records::load_all(doc); @@ -620,18 +676,50 @@ async fn propose<H: Host>( .map(|(id, _)| id) .collect(); if !expired.is_empty() { - if !supersede { - let details = all - .iter() + let remakes = || { + all.iter() .filter(|(id, _)| expired.contains(&id)) .map(|(id, record)| ExpiredRemake { id: id.clone(), coordinator: record.proposal.coordinator.clone(), + coordinator_active: supersede + .active_peers + .contains(&record.proposal.coordinator), since: record.proposal.settle_until.saturating_add(grace), + abortable_at: abortable_at(record, grace), tally: vote::tally(id, record, &all, ®ister), }) - .collect(); - return Err(StartError::Expired(details)); + .collect() + }; + if !supersede.force.allows("supersede") { + return Err(StartError::Expired { + message: "an expired remake of this epoch still holds votes: abort it, \ + or add ?force=supersede to abort it and propose in its place" + .to_owned(), + remakes: remakes(), + force: Some("supersede"), + }); + } + let obstacle = + all.iter() + .filter(|(id, _)| expired.contains(&id)) + .find_map(|(id, record)| { + let (why, force) = abort_obstacle( + record, + &me, + &supersede.active_peers, + &supersede.force, + now, + grace, + )?; + Some((format!("can't supersede remake {id}: {why}"), force)) + }); + if let Some((message, force)) = obstacle { + return Err(StartError::Expired { + message, + remakes: remakes(), + force, + }); } let decision = Decision::Aborted { reason: format!("superseded by remake {record_id}"), @@ -656,22 +744,22 @@ async fn propose<H: Host>( } // The base is the requested heads, or everything before the proposal (the - // proposal itself is never in it). - let base = match base { - Some(base) => { - if let Some(unknown) = base - .iter() - .find(|head| doc.get_change_by_hash(head).is_none()) - { - return Err(StartError::UnknownBase(unknown.to_string())); + // proposal itself is never in it); or, for a compaction, left to the commit. + if !base_at_commit { + let base = match base { + Some(base) => { + if let Some(unknown) = base + .iter() + .find(|head| doc.get_change_by_hash(head).is_none()) + { + return Err(StartError::UnknownBase(unknown.to_string())); + } + base } - base - } - None => doc.get_heads(), - }; - let mut heads: Vec<String> = base.iter().map(|h| h.to_string()).collect(); - heads.sort(); - proposal.base_heads = heads; + None => doc.get_heads(), + }; + proposal.base_heads = Some(records::encode_heads(&base)); + } // We vote for our own remake, so we are always in the electorate. let mut electorate = electorate_host.electorate(doc, now); @@ -731,16 +819,16 @@ pub(crate) async fn get<H: Host>( pub(crate) async fn abort<H: Host>( State(host): State<H>, Path(id): Path<String>, - Query(query): Query<WaitQuery>, + Query(query): Query<AbortQuery>, ) -> Response { - abort_by_id(host, id, query.wait).await + abort_by_id(host, id, query).await } /// `DELETE /admin/remakes`: aborts whichever remake of the local epoch is still /// undecided (proposed or expired), the oldest if there are several. pub(crate) async fn abort_current<H: Host>( State(host): State<H>, - Query(query): Query<WaitQuery>, + Query(query): Query<AbortQuery>, ) -> Response { let local = EpochRecord::from(host.current_epoch()); let open = host @@ -756,7 +844,7 @@ pub(crate) async fn abort_current<H: Host>( .await; match open { - Some(id) => abort_by_id(host, id, query.wait).await, + Some(id) => abort_by_id(host, id, query).await, None => error_response( StatusCode::NOT_FOUND, "no remake of the local epoch is in progress", @@ -765,14 +853,19 @@ pub(crate) async fn abort_current<H: Host>( } } -async fn abort_by_id<H: Host>(host: H, id: String, wait: bool) -> Response { +async fn abort_by_id<H: Host>(host: H, id: String, query: AbortQuery) -> Response { + let force = match Force::parse(query.force.as_deref(), ABORT_FORCE) { + Ok(force) => force, + Err(reason) => return error_response(StatusCode::BAD_REQUEST, reason, None), + }; + let wait = query.wait; let Some(current) = load_report(&host, &id).await else { return error_response(StatusCode::NOT_FOUND, "no such remake", None); }; match current.phase { Phase::Proposed => {} - Phase::Expired => return abort_expired(&host, &id).await, + Phase::Expired => return abort_expired(&host, current, &force).await, Phase::Committed => { return error_response( StatusCode::CONFLICT, @@ -823,9 +916,30 @@ async fn abort_by_id<H: Host>(host: H, id: String, wait: bool) -> Response { } } -/// Aborts expired remake `id` from this node: its coordinator is gone, and can't commit -/// it any more (`coordinator`). Answers with the final report. -async fn abort_expired<H: Host>(host: &H, id: &str) -> Response { +/// Aborts expired remake `current` from this node: its coordinator is gone, and can't +/// commit it any more (`coordinator`), unless `abort_obstacle` says otherwise. Answers +/// with the final report. +async fn abort_expired<H: Host>(host: &H, current: Report, force: &Force) -> Response { + let grace = grace(host); + if let Some((why, force)) = abort_obstacle( + ¤t.record, + &host.node_id(), + &active_peers(host).await, + force, + unix_millis(), + grace, + ) { + return error_response( + StatusCode::CONFLICT, + format!("can't abort it yet: {why}"), + Some(json!({ + "remake": current, + "abortableAt": abortable_at(¤t.record, grace), + "force": force, + })), + ); + } + let id = current.id.as_str(); let record_id = id.to_owned(); let decision = Decision::Aborted { reason: format!( @@ -834,11 +948,12 @@ async fn abort_expired<H: Host>(host: &H, id: &str) -> Response { ), decided_at: unix_millis(), }; - let grace = grace(host); let written = host .commit_doc(move |doc| -> Result<bool, StoreError> { - let expired = records::load(doc, &record_id) - .is_some_and(|record| record.phase(unix_millis(), grace) == Phase::Expired); + let now = unix_millis(); + let expired = records::load(doc, &record_id).is_some_and(|record| { + record.phase(now, grace) == Phase::Expired && now >= abortable_at(&record, grace) + }); if expired { records::write_decision(doc, &record_id, &decision)?; } diff --git a/src/remake/catchup.rs b/src/remake/catchup.rs index cad5a22..631fb25 100644 --- a/src/remake/catchup.rs +++ b/src/remake/catchup.rs @@ -31,7 +31,7 @@ use std::collections::{BTreeMap, HashMap}; use std::convert::Infallible; use std::time::Duration; -use automerge::{AutoCommit, ChangeHash}; +use automerge::{ActorId, AutoCommit, ChangeHash}; use tokio::time::Instant; use tracing::{debug, error, info, warn}; @@ -177,8 +177,12 @@ async fn start<H: Host>(host: &H, local: Epoch, target: Epoch, attempts: u32) { host.set_stale_view(Some((local, copy))); } let was_ready = host.suspend().await; + let actor = ActorId::random(); + host.remake().add_own_actor(actor.clone()); let old = match host - .switch_epoch(target, |_live| Ok::<_, Infallible>(AutoCommit::new())) + .switch_epoch(target, move |_live| { + Ok::<_, Infallible>(AutoCommit::new().with_actor(actor)) + }) .await { Ok(old) => old, @@ -276,7 +280,9 @@ async fn finish<H: Host>(host: &H) { // Everything heavy runs on a copy, without a lock. let copy = host.read_doc(|doc| doc.clone()).await; - let prepared = tokio::task::spawn_blocking(move || prepare(copy, old, previous, target)).await; + let own = host.remake().own_actors(); + let prepared = + tokio::task::spawn_blocking(move || prepare(copy, old, previous, target, own)).await; let (old, metadata, plan, own_missing, had_history) = match prepared { Ok(Prepared::Ready { @@ -376,6 +382,8 @@ async fn finish<H: Host>(host: &H) { }, ); host.remake().set_catch_up(None); + let actor = host.read_doc(|doc| doc.get_actor().clone()).await; + host.remake().reset_own_actors(actor); if was_ready { host.resume().await; } @@ -395,7 +403,15 @@ async fn finish<H: Host>(host: &H) { } /// Verifies the synced document and decides about the old document's own changes. -fn prepare(mut copy: AutoCommit, mut old: AutoCommit, previous: Epoch, target: Epoch) -> Prepared { +/// `own`: this node's actors (`RemakeState::own_actors`); the old document's own actor +/// counts in any case. +fn prepare( + mut copy: AutoCommit, + mut old: AutoCommit, + previous: Epoch, + target: Epoch, + mut own: Vec<ActorId>, +) -> Prepared { if copy.get_change_by_hash(&ChangeHash(target.id)).is_none() { return Prepared::NotYet(old); } @@ -421,7 +437,7 @@ fn prepare(mut copy: AutoCommit, mut old: AutoCommit, previous: Epoch, target: E .ok() .filter(|next| next.previous == previous); - let own = old.get_actor().clone(); + own.push(old.get_actor().clone()); let clock = next .as_ref() .map_or_else(BTreeMap::new, |next| next.clock.clone()); @@ -573,7 +589,7 @@ mod tests { #[test] fn replays_within_the_window() { let (old, swarm, target) = scenario(&json!({})); - match prepare(swarm, old, Epoch::GENESIS, target) { + match prepare(swarm, old, Epoch::GENESIS, target, Vec::new()) { Prepared::Ready { plan: Plan::Replay(paths, _), own_missing, @@ -589,7 +605,7 @@ mod tests { #[test] fn quarantines_outside_the_window() { let (old, swarm, target) = scenario(&json!({ "replayWindow": { "epochs": 0 } })); - match prepare(swarm, old, Epoch::GENESIS, target) { + match prepare(swarm, old, Epoch::GENESIS, target, Vec::new()) { Prepared::Ready { plan: Plan::Quarantine { paths, .. }, .. @@ -602,7 +618,7 @@ mod tests { fn new_node_has_nothing_to_replay() { let (_, swarm, target) = scenario(&json!({})); assert!(matches!( - prepare(swarm, AutoCommit::new(), Epoch::GENESIS, target), + prepare(swarm, AutoCommit::new(), Epoch::GENESIS, target, Vec::new()), Prepared::Ready { plan: Plan::Nothing, had_history: false, @@ -615,7 +631,7 @@ mod tests { fn waits_for_the_target_epoch() { let (old, _, target) = scenario(&json!({})); assert!(matches!( - prepare(AutoCommit::new(), old, Epoch::GENESIS, target), + prepare(AutoCommit::new(), old, Epoch::GENESIS, target, Vec::new()), Prepared::NotYet(_) )); } diff --git a/src/remake/coordinator.rs b/src/remake/coordinator.rs index 9bcf8ba..427de08 100644 --- a/src/remake/coordinator.rs +++ b/src/remake/coordinator.rs @@ -12,7 +12,7 @@ //! The decision is written in the same step that checks the records once more, so a //! commit never overrides an abort request or a competitor's commit that arrived while //! the snapshot was being built. Nor is a commit written once the remake has expired: -//! from then on, any node may abort it (`api`), and the two mustn't race. +//! a grace period later, any node may abort it (`api`), and the two mustn't race. //! //! Committing builds the snapshot at the base heads and records the new epoch, so //! every node can verify its own build against it. @@ -141,7 +141,23 @@ fn obstacle(id: &str, record: &Record, all: &[(String, Record)]) -> Option<Strin } async fn commit<H: Host>(host: &H, id: &str, record: &Record) -> Result<(), StoreError> { - let Some(mut request) = record.snapshot_request(id) else { + let content = match &record.proposal.content { + Some(reference) => match content::ensure(host, reference).await { + Ok(bytes) => Some(bytes), + Err(reason) => return decide_abort(host, id, reason).await, + }, + None => None, + }; + + // Clone under the read lock, build without any lock. + let mut copy = host.read_doc(|doc| doc.clone()).await; + // A base left to the commit is what this node has now. Every node that sees the + // decision has it too: the decision change depends on it. + let (base, fixed_now) = match &record.proposal.base_heads { + None => (Some(copy.get_heads()), true), + Some(heads) => (records::decode_heads(heads), false), + }; + let Some(mut request) = base.and_then(|base| record.snapshot_request_at(id, base)) else { return decide_abort( host, id, @@ -149,15 +165,8 @@ async fn commit<H: Host>(host: &H, id: &str, record: &Record) -> Result<(), Stor ) .await; }; - if let Some(reference) = &record.proposal.content { - match content::ensure(host, reference).await { - Ok(bytes) => request.content = Some(bytes), - Err(reason) => return decide_abort(host, id, reason).await, - } - } - - // Clone under the read lock, build without any lock. - let copy = host.read_doc(|doc| doc.clone()).await; + let base_heads = fixed_now.then(|| records::encode_heads(&request.base_heads)); + request.content = content; let mut snapshot = match snapshot::build_detached(copy, request).await { Ok((snapshot, _copy)) => snapshot, Err(error) => { @@ -170,6 +179,7 @@ async fn commit<H: Host>(host: &H, id: &str, record: &Record) -> Result<(), Stor let decision = Decision::Committed { to_epoch: to_epoch.into(), + base_heads, snapshot_bytes, decided_at: records::unix_millis(), }; diff --git a/src/remake/delta.rs b/src/remake/delta.rs index c8f6b7c..51cc409 100644 --- a/src/remake/delta.rs +++ b/src/remake/delta.rs @@ -159,22 +159,23 @@ pub(crate) struct Missing { /// Splits `doc`'s history against a snapshot `clock`: a change (actor, seq) is in the /// snapshot iff seq <= clock[actor]. /// -/// Only changes this node wrote itself (actor `own`) are worth replaying: other nodes' -/// changes after the base were replayed by those nodes when they switched, and replaying -/// them again would overwrite newer writes in the new epoch with stale ones. +/// Only changes this node wrote itself (any of the actors in `own`: one per run and +/// epoch) are worth replaying when it rejoins: other nodes' changes after the base were +/// replayed by the nodes that switched, and replaying them again would overwrite newer +/// writes in the new epoch with stale ones. pub(crate) fn missing_from_snapshot( doc: &mut AutoCommit, clock: &BTreeMap<String, u64>, - own: &ActorId, + own: &[ActorId], ) -> Missing { - let own_hex = own.to_string(); + let own_hex: HashSet<String> = own.iter().map(ActorId::to_string).collect(); let changes: Vec<(ChangeHash, Vec<ChangeHash>, bool)> = doc .get_changes(&[]) .iter() .map(|change| { let actor = change.actor_id().to_string(); let in_snapshot = clock.get(&actor).is_some_and(|&seq| change.seq() <= seq); - let own_missing = actor == own_hex && !in_snapshot; + let own_missing = own_hex.contains(&actor) && !in_snapshot; (change.hash(), change.deps().to_vec(), own_missing) }) .collect(); @@ -370,7 +371,7 @@ mod tests { doc.commit(); let after = doc.get_heads(); - let missing = missing_from_snapshot(&mut doc, &clock, &own); + let missing = missing_from_snapshot(&mut doc, &clock, std::slice::from_ref(&own)); assert_eq!(missing.own, 1); assert_eq!(missing.before, with_foreign); assert_ne!(missing.before, base); @@ -381,7 +382,7 @@ mod tests { // Nothing of ours missing: nothing to replay. let full: BTreeMap<String, u64> = [(own.to_string(), 2)].into_iter().collect(); - let none = missing_from_snapshot(&mut doc, &full, &own); + let none = missing_from_snapshot(&mut doc, &full, std::slice::from_ref(&own)); assert_eq!(none.own, 0); let mut heads = after.clone(); heads.sort(); diff --git a/src/remake/kind.rs b/src/remake/kind.rs index 8131f7f..42bdd88 100644 --- a/src/remake/kind.rs +++ b/src/remake/kind.rs @@ -17,6 +17,9 @@ //! Protocol data under `_continuum` is not a kind's business: the generic part carries //! it forward and replays it unchanged. +use std::path::Path; + +use anyhow::{Context, anyhow, bail}; use automerge::AutoCommit; use base64::engine::{Engine, general_purpose::STANDARD as BASE64}; use serde::{Deserialize, Serialize}; @@ -24,6 +27,7 @@ use serde_json::Value as JsonValue; use super::snapshot::{META_ROOT, canonical}; use crate::document::store::{self, StoreError}; +use crate::files::{FileKind, Files}; /// Name of the kind used when a request doesn't name one. pub(crate) const DEFAULT_KIND: &str = "compact"; @@ -71,6 +75,14 @@ pub(crate) trait RemakeKind: Send + Sync + 'static { /// Every node: where a user path written after the base goes. fn translate(&self, path: &[String], translation: &JsonValue) -> Translation; + + /// Whether the coordinator fixes the base when it commits instead of when it + /// proposes (no `baseHeads` then). Right for a kind whose content is the old + /// document at the base: the later the base, the fewer writes every switching node + /// replays as fresh values (whole lists and texts, concurrently on every node). + fn base_at_commit(&self) -> bool { + false + } } /// The kind with this name. @@ -83,6 +95,50 @@ pub(crate) fn by_name(name: &str) -> Option<&'static dyn RemakeKind> { } } +/// Builds a new swarm's genesis content into `target` (`--genesis --remake SPEC`), the +/// way a remake of that kind builds a new epoch from an empty one: +/// * `import:PATH`: an Automerge document, e.g. a backup (its content, without history +/// and protocol data); +/// * `replace:PATH`: a JSON document, the store root; +/// * `compact`: nothing (an empty document compacts to itself). +pub(crate) fn remake_genesis( + spec: &str, + files: &Files, + target: &mut AutoCommit, +) -> anyhow::Result<()> { + let (name, path) = match spec.split_once(':') { + Some((name, path)) => (name, Some(Path::new(path))), + None => (spec, None), + }; + let kind = by_name(name).ok_or_else(|| anyhow!("unknown remake kind {name:?}"))?; + let request = match (kind.name(), path) { + ("compact", None) => return Ok(()), + ("compact", Some(_)) => bail!("compact takes no file"), + (_, None) => bail!("{name} needs a file: {name}:PATH"), + (name, Some(path)) => { + let bytes = files + .read(FileKind::Remake, path) + .with_context(|| format!("Reading {}", path.display()))? + .with_context(|| format!("{} doesn't exist", path.display()))?; + match name { + "import" => serde_json::json!({ "document": BASE64.encode(&bytes) }), + "replace" => { + let document: JsonValue = serde_json::from_slice(&bytes) + .with_context(|| format!("{} is not JSON", path.display()))?; + serde_json::json!({ "document": document }) + } + other => bail!("{other} can't make a genesis document"), + } + } + }; + let prepared = kind + .propose(&request) + .map_err(|reason| anyhow!("{name}: {reason}"))?; + kind.build(&AutoCommit::new(), prepared.content.as_deref(), target)?; + target.commit(); + Ok(()) +} + /// Paths owned by the protocol rather than by any kind. pub(crate) fn is_reserved(path: &[String]) -> bool { path.first().is_some_and(|root| root == META_ROOT) @@ -97,6 +153,10 @@ impl RemakeKind for Compact { "compact" } + fn base_at_commit(&self) -> bool { + true + } + fn propose(&self, spec: &JsonValue) -> Result<Prepared, String> { match spec { JsonValue::Null => Ok(Prepared::default()), diff --git a/src/remake/local.rs b/src/remake/local.rs index 04ac0b0..65a2735 100644 --- a/src/remake/local.rs +++ b/src/remake/local.rs @@ -9,7 +9,7 @@ use std::collections::{HashMap, HashSet}; use std::sync::{Arc, Mutex, MutexGuard}; use std::time::Duration; -use automerge::AutoCommit; +use automerge::{ActorId, AutoCommit}; use serde::Serialize; use tokio::time::Instant; @@ -55,6 +55,11 @@ struct Inner { content_fetches: HashSet<[u8; 32]>, /// Remakes this node coordinates now (`coordinator`). coordinating: HashSet<String>, + /// The actors this node writes or wrote with in the current document: this run's, + /// and those of earlier runs it knows of (`--document-file`). Its own writes, for a + /// rejoin. A switch or rejoin starts the list over with the new document's actor, so + /// it only grows with restarts within one epoch. + own_actors: Vec<ActorId>, } /// Content no remake refers to any more is kept this long after it was stored, for @@ -147,6 +152,24 @@ impl RemakeState { self.started_at } + /// This node writes with `actor` from now on (or did, in an earlier run). + pub(crate) fn add_own_actor(&self, actor: ActorId) { + let mut inner = self.lock(); + if !inner.own_actors.contains(&actor) { + inner.own_actors.push(actor); + } + } + + /// The current document is a new one, written with `actor` only (a switch or rejoin + /// is done). + pub(crate) fn reset_own_actors(&self, actor: ActorId) { + self.lock().own_actors = vec![actor]; + } + + pub(crate) fn own_actors(&self) -> Vec<ActorId> { + self.lock().own_actors.clone() + } + /// Whether this node starts coordinating `id` now (false if it already does). pub(crate) fn start_coordinating(&self, id: &str) -> bool { self.lock().coordinating.insert(id.to_owned()) diff --git a/src/remake/mod.rs b/src/remake/mod.rs index ca17956..9085475 100644 --- a/src/remake/mod.rs +++ b/src/remake/mod.rs @@ -32,6 +32,7 @@ pub(crate) mod snapshot; mod switch; mod vote; +pub(crate) use kind::remake_genesis; pub(crate) use local::RemakeState; pub(crate) use switch::run_worker; diff --git a/src/remake/records.rs b/src/remake/records.rs index cd3e5be..283f612 100644 --- a/src/remake/records.rs +++ b/src/remake/records.rs @@ -8,9 +8,10 @@ //! * `abortRequested`: written by any node on an operator's request. It is an input to //! the coordinator's decision, not a decision itself. //! -//! Once a remake has expired, its coordinator no longer commits it, and any node may -//! abort it on an operator's request (`api`). Should clocks disagree by more than the -//! grace margin, a commit and such an abort could still be written concurrently. Such +//! Once a remake has expired, its coordinator no longer commits it, and a grace period +//! later any node may abort it on an operator's request (`api`). Should clocks disagree +//! by more than that grace period, a commit and such an abort could still be written +//! concurrently. Such //! a record is *contested*: it reads as committed, and nodes don't switch to it on //! their own. //! @@ -130,7 +131,9 @@ pub(crate) struct Proposal { pub(crate) coordinator: String, pub(crate) from_epoch: EpochRecord, /// Hex change hashes, sorted. The proposal record itself is not part of the base. - pub(crate) base_heads: Vec<String>, + /// None for a kind whose base is fixed when it commits (`Decision::Committed`). + #[serde(default, skip_serializing_if = "Option::is_none")] + pub(crate) base_heads: Option<Vec<String>>, pub(crate) policy: Policy, pub(crate) settle_ms: u64, pub(crate) proposed_at: u64, @@ -181,13 +184,19 @@ fn default_kind() -> String { DEFAULT_KIND.to_owned() } -impl Proposal { - pub(crate) fn base_change_hashes(&self) -> Option<Vec<ChangeHash>> { - self.base_heads - .iter() - .map(|hex| decode_hex32(hex).map(ChangeHash)) - .collect() - } +/// Hex change hashes as hashes; None if one is unreadable. +pub(crate) fn decode_heads(heads: &[String]) -> Option<Vec<ChangeHash>> { + heads + .iter() + .map(|hex| decode_hex32(hex).map(ChangeHash)) + .collect() +} + +/// Hashes as sorted hex, as records keep them. +pub(crate) fn encode_heads(heads: &[ChangeHash]) -> Vec<String> { + let mut hex: Vec<String> = heads.iter().map(ChangeHash::to_string).collect(); + hex.sort(); + hex } #[derive(Clone, Debug, Serialize, Deserialize)] @@ -196,6 +205,9 @@ pub(crate) enum Decision { #[serde(rename_all = "camelCase")] Committed { to_epoch: EpochRecord, + /// The base, when the proposal leaves it to the commit (sorted hex). + #[serde(default, skip_serializing_if = "Option::is_none")] + base_heads: Option<Vec<String>>, snapshot_bytes: u64, decided_at: u64, }, @@ -275,13 +287,39 @@ impl Record { } } + /// The base: the proposal's, or for a kind that fixes it at the commit, the + /// decision's. None until committed then, or if unreadable. + pub(crate) fn base(&self) -> Option<Vec<ChangeHash>> { + let heads = match (&self.proposal.base_heads, &self.decision) { + (Some(heads), _) => heads, + ( + None, + Some(Decision::Committed { + base_heads: Some(heads), + .. + }), + ) => heads, + _ => return None, + }; + decode_heads(heads) + } + /// Everything needed to (re)build this remake's snapshot but the content (which the - /// caller adds, `content::ensure`); None if the proposal is unreadable. + /// caller adds, `content::ensure`); None if the proposal or base is unreadable. pub(crate) fn snapshot_request(&self, id: &str) -> Option<SnapshotRequest> { + self.snapshot_request_at(id, self.base()?) + } + + /// The same, at `base` (the coordinator fixing a base at the commit). + pub(crate) fn snapshot_request_at( + &self, + id: &str, + base: Vec<ChangeHash>, + ) -> Option<SnapshotRequest> { Some(SnapshotRequest { remake_id: id.to_owned(), previous: self.proposal.from_epoch.to_epoch()?, - base_heads: self.proposal.base_change_hashes()?, + base_heads: base, policy: serde_json::to_value(&self.proposal.policy).ok()?, proposed_at: self.proposal.proposed_at, kind: self.proposal.kind.clone(), @@ -499,7 +537,7 @@ mod tests { Proposal { coordinator: "node-a".into(), from_epoch: Epoch::GENESIS.into(), - base_heads: vec!["ab".repeat(32)], + base_heads: Some(vec!["ab".repeat(32)]), policy: Policy { replay_window: Policy::default_replay_window(), archive_retention: Policy::default_archive_retention(), @@ -535,6 +573,7 @@ mod tests { generation: 1, id: "cd".repeat(32), }, + base_heads: None, snapshot_bytes: 123, decided_at: 11_000, }; @@ -552,7 +591,7 @@ mod tests { let (id, record) = &all[0]; assert_eq!(id, "c-1"); assert_eq!(record.proposal.policy, p.policy); - assert_eq!(record.proposal.base_change_hashes().unwrap().len(), 1); + assert_eq!(record.base().unwrap().len(), 1); assert!(record.abort_requested.is_some()); assert_eq!(record.votes.keys().collect::<Vec<_>>(), ["node-b"]); assert_eq!(record.proposal.quorum, 2); @@ -608,6 +647,31 @@ mod tests { } #[test] + fn a_base_fixed_at_the_commit() { + let mut record = Record { + proposal: Proposal { + base_heads: None, + ..proposal(10_000) + }, + decision: None, + abort_requested: None, + contested: false, + votes: BTreeMap::new(), + }; + assert_eq!(record.base(), None); + record.decision = Some(Decision::Committed { + to_epoch: EpochRecord { + generation: 1, + id: "cd".repeat(32), + }, + base_heads: Some(vec!["ef".repeat(32)]), + snapshot_bytes: 1, + decided_at: 2, + }); + assert_eq!(record.base(), Some(vec![ChangeHash([0xef; 32])])); + } + + #[test] fn disagreeing_decisions_are_contested() { let mut base = AutoCommit::new(); write_proposal(&mut base, "c-1", &proposal(10_000)).unwrap(); @@ -618,6 +682,7 @@ mod tests { generation: 1, id: "cd".repeat(32), }, + base_heads: None, snapshot_bytes: 1, decided_at: 2, }; diff --git a/src/remake/snapshot.rs b/src/remake/snapshot.rs index e50a2a2..20569e0 100644 --- a/src/remake/snapshot.rs +++ b/src/remake/snapshot.rs @@ -38,6 +38,8 @@ pub(crate) const REMAKES: &str = "remakes"; /// The votes register (`records`). pub(crate) const VOTES: &str = "votes"; pub(crate) const EPOCHS: &str = "epochs"; +/// The swarm ID, written once by the genesis node (`swarm::bootstrap`). +pub(crate) const SWARM: &str = "swarm"; /// Everything that determines a snapshot besides the base content. pub(crate) struct SnapshotParams<'a> { @@ -224,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. @@ -521,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( @@ -695,7 +750,8 @@ mod tests { assert!(verify_document(&mut fetched, wrong).is_err()); let own = late.get_actor().clone(); - let missing = delta::missing_from_snapshot(&mut late, &metadata.clock, &own); + let missing = + delta::missing_from_snapshot(&mut late, &metadata.clock, std::slice::from_ref(&own)); assert_eq!(missing.own, 1); let heads = late.get_heads(); let mut adopted = fetched.with_actor(ActorId::from(&b"late-node-next"[..])); diff --git a/src/remake/switch.rs b/src/remake/switch.rs index cf5ea75..1a4dbf9 100644 --- a/src/remake/switch.rs +++ b/src/remake/switch.rs @@ -205,6 +205,10 @@ async fn switch<H: Host>( .take_prebuilt(id) .filter(|snapshot| snapshot.epoch == expected); + let actor = ActorId::random(); + host.remake().add_own_actor(actor.clone()); + let new_actor = actor.clone(); + // 2. Build (or reuse), verify and replay without holding any lock. let (new_doc, copy_heads, mut stats) = tokio::task::spawn_blocking( move || -> Result<(AutoCommit, Vec<ChangeHash>, ReplayStats), String> { @@ -222,7 +226,7 @@ async fn switch<H: Host>( } // A fresh actor: this node's writes in the new epoch. - let mut doc = snapshot.doc.with_actor(ActorId::random()); + let mut doc = snapshot.doc.with_actor(actor); // Keep this remake's record visible in the new epoch, for its history. if carry_record { @@ -242,6 +246,7 @@ async fn switch<H: Host>( // 3. Swap, replaying what arrived since the copy. let (old, late) = swap_in(host, expected, new_doc, copy_heads, late_translator).await?; stats.absorb(late); + host.remake().reset_own_actors(new_actor); // 4. Writes the remake had no place for wait for an operator (the old document // holds them); the old document is kept as configured either way. diff --git a/src/remake/vote.rs b/src/remake/vote.rs index 16de87a..b4936dc 100644 --- a/src/remake/vote.rs +++ b/src/remake/vote.rs @@ -304,7 +304,7 @@ mod tests { proposal: Proposal { coordinator: electorate[0].to_owned(), from_epoch: Epoch::GENESIS.into(), - base_heads: Vec::new(), + base_heads: Some(Vec::new()), policy: Policy { replay_window: Policy::default_replay_window(), archive_retention: Policy::default_archive_retention(), diff --git a/src/simtest/mod.rs b/src/simtest/mod.rs new file mode 100644 index 0000000..18975aa --- /dev/null +++ b/src/simtest/mod.rs @@ -0,0 +1,292 @@ +//! Tests of whole nodes on the simulated network (`transport::sim`): several nodes in +//! one process, built and started as `main` does (`node`), talking through the +//! Mycelium wire encoding. +//! +//! The clock is real (see `transport::sim` for why), so the nodes run with a short +//! keepalive and tests wait for conditions with a timeout instead of advancing time. +//! Timeouts are generous: a test fails only if something doesn't happen at all. +//! +//! `RUST_LOG=continuum=debug cargo test simtest -- --nocapture` shows the nodes' logs +//! (all nodes interleaved; messages name peers by ID). + +mod timing; + +use std::collections::HashSet; +use std::future::Future; +use std::sync::Arc; +use std::time::Duration; + +use automerge::{AutoCommit, ChangeHash}; +use serde_json::Value as JsonValue; +use tokio::time::Instant; + +use crate::document::replica::Epoch; +use crate::document::store; +use crate::files::Files; +use crate::node::{self, Node, NodeConfig}; +use crate::state::{AppState, PeerState, Timing}; +use crate::swarm::bootstrap::{self, Start}; +use crate::transport::sim::SimNet; + +/// Keepalive of simulated nodes: dormancy after 1 s, probes of dormant peers every +/// 2.5 s (`Timing::new`). +pub(crate) const KEEPALIVE: Duration = Duration::from_millis(250); +/// First wait for a sync answer before sending again (10 s in production, where large +/// messages take a while): a lost sync message costs a test 1 s, a heal 2 s. +pub(crate) const SYNC_REPLY_TIMEOUT: Duration = Duration::from_secs(1); +/// How long the simulated net may take to deliver a message: well above its delays. +/// A catch-up source that stops sending sync messages is given up after this. +pub(crate) const DELIVERY_LIMIT: Duration = Duration::from_secs(3); + +/// A simulated swarm: the net and the nodes started on it. +pub(crate) struct Sim { + pub(crate) net: SimNet, + /// Every node started, for the description of a failed wait. + nodes: std::sync::Mutex<Vec<(String, AppState)>>, +} + +impl Sim { + pub(crate) fn new(seed: u64) -> Self { + init_logging(); + Self { + net: SimNet::new(seed), + nodes: Default::default(), + } + } + + /// Starts node `id` with the test defaults: builds it, starts its workers, and + /// bootstraps it in the background (through `seeds`, other nodes' IDs). Returns at + /// once; [`wait_ready`] waits for it. + pub(crate) fn start(&self, id: &str, seeds: &[&str], start: Start) -> AppState { + self.start_with(id, seeds, start, |_| {}) + } + + /// [`start`](Self::start), with the configuration adjusted first. + pub(crate) fn start_with( + &self, + id: &str, + seeds: &[&str], + start: Start, + adjust: impl FnOnce(&mut NodeConfig), + ) -> AppState { + let mut config = NodeConfig { + transport: self.net.transport(id), + initial_epoch: Epoch::GENESIS, + initial_doc: AutoCommit::new(), + earlier_actors: Vec::new(), + timing: Timing { + delivery_limit: DELIVERY_LIMIT, + ..Timing::new(KEEPALIVE, SYNC_REPLY_TIMEOUT) + }, + neighbours_k: 8, + alias: Some(id.to_owned()), + electorate_window: Duration::from_secs(30 * 24 * 3600), + blob_chunk_bytes: 64 * 1024, + swarm_key: None, + protocol_writers: HashSet::new(), + files: Arc::new(Files::default()), + state_file: None, + document_file: None, + config_report: JsonValue::Null, + }; + adjust(&mut config); + let state = Node::new(config).spawn_workers(); + self.nodes + .lock() + .unwrap() + .push((id.to_owned(), state.clone())); + + let seeds: Vec<String> = seeds.iter().map(|seed| (*seed).to_owned()).collect(); + let bootstrapping = state.clone(); + let id = id.to_owned(); + tokio::spawn(async move { + if let Err(error) = bootstrap::bootstrap(&bootstrapping, &seeds, start).await { + panic!("bootstrapping {id} failed: {error:#}"); + } + node::spawn_ready_workers(&bootstrapping); + }); + state + } +} + +/// Logs to the test output if `RUST_LOG` is set. +fn init_logging() { + if std::env::var_os("RUST_LOG").is_some() { + let _ = tracing_subscriber::fmt() + .with_env_filter(tracing_subscriber::EnvFilter::from_default_env()) + .with_test_writer() + .compact() + .try_init(); + } +} + +impl Sim { + /// Waits until `condition` holds, checking every 20 ms. After `timeout`, panics with + /// a description of every node (`describe`). + pub(crate) async fn wait_until<F, Fut>(&self, what: &str, timeout: Duration, mut condition: F) + where + F: FnMut() -> Fut, + Fut: Future<Output = bool>, + { + let deadline = Instant::now() + timeout; + loop { + if condition().await { + return; + } + if Instant::now() >= deadline { + panic!( + "timed out after {timeout:?} waiting until {what}\n{}", + self.describe().await + ); + } + tokio::time::sleep(Duration::from_millis(20)).await; + } + } + + /// Waits until every node is ready. + pub(crate) async fn wait_ready(&self, nodes: &[&AppState], timeout: Duration) { + self.wait_until("every node is ready", timeout, move || async move { + for node in nodes { + if !node.is_ready().await { + return false; + } + } + true + }) + .await; + } + + /// Waits until every node has the same heads. + pub(crate) async fn wait_converged(&self, nodes: &[&AppState], timeout: Duration) { + self.wait_until( + "every node has the same heads", + timeout, + move || async move { + let first = heads(nodes[0]).await; + for node in &nodes[1..] { + if heads(node).await != first { + return false; + } + } + true + }, + ) + .await; + } + + /// Every node's state as far as sync and liveness go, and what the net counted. + pub(crate) async fn describe(&self) -> String { + use std::fmt::Write; + + let nodes = self.nodes.lock().unwrap().clone(); + let now = Instant::now(); + let ago = |at: Option<Instant>| at.map(|at| now.saturating_duration_since(at)); + let mut out = format!("net: {:?}\n", self.net.counts()); + for (id, node) in &nodes { + let heads: Vec<String> = heads(node) + .await + .iter() + .map(|head| head.to_string()[..8].to_owned()) + .collect(); + let _ = writeln!( + out, + "node {id}: ready={} epoch={} heads={heads:?} initial_sync_done={}", + node.is_ready().await, + node.local_epoch(), + node.initial_sync_done + .load(std::sync::atomic::Ordering::Acquire), + ); + let _ = writeln!( + out, + " view: {:?}", + *node + .neighbours + .read() + .unwrap_or_else(|poisoned| poisoned.into_inner()) + ); + let peers: Vec<_> = node + .peer_states + .read() + .await + .iter() + .map(|(peer, state)| (peer.clone(), Arc::clone(state))) + .collect(); + for (peer, peer_state) in peers { + let p = peer_state.lock().await; + let _ = writeln!( + out, + " peer {peer}: neighbour={} leads={} standing={:?} liveness={:?} hello={:?} \ + reports_ready={:?} epoch={:?} last_seen_ago={:?} last_sync_rx_ago={:?} \ + unreachable={} their_heads={:?} in_flight={} sending={} reply_due_in={:?} \ + retry_at={:?}", + crate::swarm::topology::is_neighbour(node, &peer), + p.leads, + p.standing, + p.liveness, + p.hello_state, + p.reports_ready, + p.epoch, + ago(p.last_seen), + ago(p.last_sync_rx), + p.is_unreachable(&node.timing, now), + p.sync.state.their_heads.as_ref().map(|heads| heads + .iter() + .map(|head| head.to_string()[..8].to_owned()) + .collect::<Vec<_>>()), + p.sync.state.in_flight, + p.sync.sending.is_some(), + p.sync + .reply + .as_ref() + .map(|wait| wait.deadline.saturating_duration_since(now)), + p.sync.retry_at.map(|at| at.saturating_duration_since(now)), + ); + } + } + out + } +} + +fn path(path: &str) -> Vec<String> { + path.split('/').map(str::to_owned).collect() +} + +/// Writes `value` at `path` (`a/b/c`) as one change, as the API does. +pub(crate) async fn write(node: &AppState, at: &str, value: JsonValue) { + let at = path(at); + node.mutate_doc(move |doc| store::write_path(doc, &at, &value)) + .await + .expect("write"); +} + +/// The value at `path` (`a/b/c`), if any. +pub(crate) async fn read(node: &AppState, at: &str) -> Option<JsonValue> { + store::read_path(&node.doc.read().await.doc, &path(at)).expect("read") +} + +pub(crate) async fn heads(node: &AppState) -> Vec<ChangeHash> { + let mut heads = node.doc.read().await.heads().to_vec(); + heads.sort_unstable(); + heads +} + +/// Whether `node` has `peer` among its neighbours as computed from the member list. +/// Before that list is known everyone counts as a neighbour, so this waits for it. +pub(crate) fn has_neighbour(node: &AppState, peer: &str) -> bool { + node.neighbours + .read() + .unwrap_or_else(|poisoned| poisoned.into_inner()) + .is_some() + && crate::swarm::topology::is_neighbour(node, peer) +} + +/// Something about `node`'s state for `peer`; None if it doesn't know the peer. +pub(crate) async fn peer<T>( + node: &AppState, + peer: &str, + f: impl FnOnce(&PeerState) -> T, +) -> Option<T> { + let peer_state = node.get_peer_state(peer).await?; + let peer_state = peer_state.lock().await; + Some(f(&peer_state)) +} diff --git a/src/simtest/timing.rs b/src/simtest/timing.rs new file mode 100644 index 0000000..7810f28 --- /dev/null +++ b/src/simtest/timing.rs @@ -0,0 +1,285 @@ +//! The timing rules of `peers.md` on whole nodes: joining and converging, dormancy and +//! return, and late messages that must not undo what newer ones said. + +use std::time::Duration; + +use serde_json::json; + +use super::{Sim, has_neighbour, heads, peer, read, write}; +use crate::document::replica::Epoch; +use crate::state::{AppState, Liveness}; +use crate::swarm::bootstrap::Start; +use crate::transport::sim::LinkParams; +use crate::transport::{HeadsSummary, HelloRequest, Keepalive, MessageKind, ParsedMessage, RunId}; + +/// Long enough for anything that should happen within a few keepalives. +const SOON: Duration = Duration::from_secs(10); +/// With lost messages: several reply timeouts (1 s, doubling) and heals in a row. +const LOSSY: Duration = Duration::from_secs(30); + +#[tokio::test(flavor = "multi_thread", worker_threads = 4)] +async fn three_nodes_join_and_converge() { + let sim = Sim::new(1); + let a = sim.start("a", &[], Start::Genesis); + let b = sim.start("b", &["a"], Start::Join); + let c = sim.start("c", &["a"], Start::Join); + let (a, b, c) = (&a, &b, &c); + sim.wait_ready(&[a, b, c], SOON).await; + + let swarm = a.swarm_id().await; + assert!(swarm.is_some()); + assert_eq!(b.swarm_id().await, swarm); + assert_eq!(c.swarm_id().await, swarm); + + // A write anywhere reaches everyone. + write(b, "users/bob", json!({"name": "Bob"})).await; + sim.wait_until("c has b's write", SOON, move || async move { + read(c, "users/bob/name").await == Some(json!("Bob")) + }) + .await; + + // Concurrent writes merge to one document. + write(a, "users/alice", json!(1)).await; + write(c, "users/carol", json!(3)).await; + sim.wait_converged(&[a, b, c], SOON).await; + for node in [a, b, c] { + assert_eq!(read(node, "users/alice").await, Some(json!(1))); + assert_eq!(read(node, "users/carol").await, Some(json!(3))); + } +} + +#[tokio::test(flavor = "multi_thread", worker_threads = 4)] +async fn converges_with_duplicated_and_reordered_messages() { + let sim = Sim::new(2); + sim.net.set_default_link(LinkParams { + duplicate: 0.3, + delay: (Duration::from_millis(1), Duration::from_millis(40)), + ..LinkParams::default() + }); + let a = sim.start("a", &[], Start::Genesis); + let b = sim.start("b", &["a"], Start::Join); + let c = sim.start("c", &["a"], Start::Join); + let (a, b, c) = (&a, &b, &c); + sim.wait_ready(&[a, b, c], SOON).await; + + for round in 0..10 { + write(a, &format!("a/{round}"), json!(round)).await; + write(b, &format!("b/{round}"), json!(round)).await; + write(c, &format!("c/{round}"), json!(round)).await; + } + sim.wait_converged(&[a, b, c], SOON).await; + for node in [a, b, c] { + for prefix in ["a", "b", "c"] { + assert_eq!(read(node, &format!("{prefix}/9")).await, Some(json!(9))); + } + } + assert!(sim.net.counts().duplicated > 0); +} + +/// Lost messages are made up for: sync messages by the reply timeout, sessions out of +/// step by the keepalive heal. Joining under loss too: HELLOs are retried. +#[tokio::test(flavor = "multi_thread", worker_threads = 4)] +async fn converges_over_lossy_links() { + let sim = Sim::new(5); + sim.net.set_default_link(LinkParams { + loss: 0.15, + duplicate: 0.05, + delay: (Duration::from_millis(1), Duration::from_millis(30)), + ..LinkParams::default() + }); + let a = sim.start("a", &[], Start::Genesis); + let b = sim.start("b", &["a"], Start::Join); + let c = sim.start("c", &["a"], Start::Join); + let (a, b, c) = (&a, &b, &c); + sim.wait_ready(&[a, b, c], LOSSY).await; + + for round in 0..10 { + write(a, &format!("a/{round}"), json!(round)).await; + write(b, &format!("b/{round}"), json!(round)).await; + write(c, &format!("c/{round}"), json!(round)).await; + tokio::time::sleep(Duration::from_millis(50)).await; + } + sim.wait_converged(&[a, b, c], LOSSY).await; + for node in [a, b, c] { + for prefix in ["a", "b", "c"] { + assert_eq!(read(node, &format!("{prefix}/9")).await, Some(json!(9))); + } + } + assert!(sim.net.counts().lost > 0); +} + +/// R24: a catch-up source that is alive (it answers handshakes and keepalives) but whose +/// sync messages never arrive is given up after three unanswered sync messages and +/// `delivery_limit` without one, and another peer takes over. +#[tokio::test(flavor = "multi_thread", worker_threads = 4)] +async fn stalled_catch_up_source_is_replaced() { + let sim = Sim::new(6); + let a = sim.start("a", &[], Start::Genesis); + let b = sim.start("b", &["a"], Start::Join); + let (a, b) = (&a, &b); + sim.wait_ready(&[a, b], SOON).await; + write(a, "data", json!("x")).await; + sim.wait_converged(&[a, b], SOON).await; + + // c joins through a, so a becomes its source (the first ready peer it hears + // from); b it learns of from a. + sim.net.set_link( + "a", + "c", + LinkParams { + drop: vec![MessageKind::Sync], + ..LinkParams::default() + }, + ); + let c = sim.start("c", &["a"], Start::Join); + let c = &c; + sim.wait_ready(&[c], LOSSY).await; + + assert_eq!(read(c, "data").await, Some(json!("x"))); + let sources = c.stats.snapshot().catch_up_sources; + assert!(sources >= 2, "{sources} catch-up sources"); +} + +/// A peer that hears nothing for `dormant_after` is dormant; once messages flow again +/// it is active, re-handshakes, and syncs what it missed. +#[tokio::test(flavor = "multi_thread", worker_threads = 4)] +async fn peer_goes_dormant_and_returns() { + let sim = Sim::new(3); + let a = sim.start("a", &[], Start::Genesis); + let b = sim.start("b", &["a"], Start::Join); + let (a, b) = (&a, &b); + sim.wait_ready(&[a, b], SOON).await; + // Only neighbours are judged, and a node joins the member list (which neighbours + // are computed from) within 5 s of becoming ready: until then the other doesn't + // count it as a neighbour. + sim.wait_until( + "a and b are each other's neighbours", + SOON, + move || async move { has_neighbour(a, "b") && has_neighbour(b, "a") }, + ) + .await; + + sim.net.take_down("b"); + sim.wait_until("a finds b dormant", SOON, move || is_dormant(a, "b")) + .await; + sim.wait_until("b finds a dormant", SOON, move || is_dormant(b, "a")) + .await; + let dormancies = peer(a, "b", |p| p.incidents.dormancies).await.unwrap(); + assert!(dormancies >= 1); + + // Written while apart. + write(a, "while/apart", json!(true)).await; + assert_ne!(heads(a).await, heads(b).await); + + sim.net.bring_up("b"); + sim.wait_until("a finds b active", SOON, move || async move { + !is_dormant(a, "b").await + }) + .await; + sim.wait_until("b finds a active", SOON, move || async move { + !is_dormant(b, "a").await + }) + .await; + sim.wait_until("b has what a wrote meanwhile", SOON, move || async move { + read(b, "while/apart").await == Some(json!(true)) + }) + .await; + sim.wait_converged(&[a, b], SOON).await; +} + +async fn is_dormant(node: &AppState, other: &str) -> bool { + peer(node, other, |p| { + matches!(p.liveness, Liveness::Dormant { .. }) + }) + .await + == Some(true) +} + +/// R17: a message sent before the peer's last epoch change and delivered late doesn't +/// flip the epoch we recorded for it back; a new run of the peer starts its count over. +/// The peer is a "ghost" whose messages are injected, so their order is the test's. +#[tokio::test(flavor = "multi_thread", worker_threads = 4)] +async fn late_epoch_does_not_flip_a_peers_epoch() { + let sim = Sim::new(4); + let a = sim.start("a", &[], Start::Genesis); + sim.wait_ready(&[&a], SOON).await; + let swarm_id = a.swarm_id().await.unwrap(); + + let epoch = |generation: u64| Epoch { + generation, + id: [generation as u8; 32], + }; + let hello = |run: RunId, epoch: Epoch, epoch_seq: u32| { + ParsedMessage::Hello(HelloRequest { + swarm_id: Some(swarm_id.clone()), + epoch, + epoch_seq, + auth: Vec::new(), + ready: true, + run_id: run, + }) + }; + let ping = |epoch: Epoch, epoch_seq: u32| { + ParsedMessage::Ping(Keepalive { + heads: HeadsSummary::of(Vec::new()), + epoch, + epoch_seq, + ready: true, + syncs_with_you: false, + }) + }; + let a = &a; + let recorded = move || async move { + peer(a, "ghost", |p| (p.epoch, p.epoch_seq)) + .await + .expect("a knows the ghost") + }; + + let first_run: RunId = [1; 16]; + sim.net + .inject("ghost", "a", &hello(first_run, Epoch::GENESIS, 0)) + .await + .unwrap(); + assert_eq!(recorded().await, (Some(Epoch::GENESIS), Some(0))); + + // The ghost switched twice; we hear of the second switch first. + sim.net + .inject("ghost", "a", &ping(epoch(2), 2)) + .await + .unwrap(); + assert_eq!(recorded().await, (Some(epoch(2)), Some(2))); + + // The PING from between the switches arrives late: ignored. + sim.net + .inject("ghost", "a", &ping(epoch(1), 1)) + .await + .unwrap(); + assert_eq!(recorded().await, (Some(epoch(2)), Some(2))); + // So is one from before both. + sim.net + .inject("ghost", "a", &ping(Epoch::GENESIS, 0)) + .await + .unwrap(); + assert_eq!(recorded().await, (Some(epoch(2)), Some(2))); + + // The ghost restarted: its new run counts from 0 again. + let second_run: RunId = [2; 16]; + sim.net + .inject("ghost", "a", &hello(second_run, epoch(2), 0)) + .await + .unwrap(); + assert_eq!(recorded().await, (Some(epoch(2)), Some(0))); + assert_eq!( + peer(a, "ghost", |p| p.run_id).await.unwrap(), + Some(second_run) + ); + sim.net + .inject("ghost", "a", &ping(epoch(3), 1)) + .await + .unwrap(); + assert_eq!(recorded().await, (Some(epoch(3)), Some(1))); + + // One ahead peer isn't enough for a ready node to rejoin. + assert_eq!(a.local_epoch(), Epoch::GENESIS); + assert!(a.is_ready().await); +} diff --git a/src/state.rs b/src/state.rs index 9c5552f..78b058c 100644 --- a/src/state.rs +++ b/src/state.rs @@ -127,8 +127,14 @@ pub(crate) struct PeerState { /// from it in between (None while they match). pub(crate) diverged_since: Option<Instant>, - /// Epoch the peer advertised in its most recent message. + /// Epoch the peer advertised in its most recent message (late ones aside: see + /// `epoch_seq`). pub(crate) epoch: Option<Epoch>, + /// The peer's `epoch_seq` with `epoch`: how many times it changed epoch in its run. + /// An epoch with a lower one is older than what we have, delivered late. + pub(crate) epoch_seq: Option<u32>, + /// The peer's run, from its HELLO or HELLO_REPLY: a new one starts `epoch_seq` over. + pub(crate) run_id: Option<crate::transport::RunId>, /// Whether the peer was ready according to its most recent sync message or keepalive. pub(crate) reports_ready: Option<bool>, @@ -215,18 +221,22 @@ pub(crate) struct SyncReplyWait { /// First wait for a sync answer; generous, since Mycelium delivers large messages in /// many small packets. const SYNC_REPLY_TIMEOUT: Duration = Duration::from_secs(10); -/// Upper bound (Mycelium gives up on a message after five minutes). -const SYNC_REPLY_TIMEOUT_MAX: Duration = Duration::from_secs(300); +/// Upper bound, as a multiple of the first wait: 5 minutes (Mycelium gives up on a +/// message after five minutes). +const SYNC_REPLY_TIMEOUT_MAX_FACTOR: u32 = 30; +/// How long Mycelium tries to deliver a message before dropping it. +const MYCELIUM_DELIVERY_LIMIT: Duration = Duration::from_secs(300); impl SyncSession { - pub(crate) fn new() -> Self { + /// `reply_backoff`: waits for sync answers (`Timing::sync_reply_backoff`). + pub(crate) fn new(reply_backoff: Backoff) -> Self { Self { state: sync::State::new(), sending: None, retry_at: None, send_backoff: Backoff::new(SEND_RETRY, SEND_RETRY_MAX), reply: None, - reply_backoff: Backoff::new(SYNC_REPLY_TIMEOUT, SYNC_REPLY_TIMEOUT_MAX), + reply_backoff, } } @@ -315,13 +325,13 @@ impl SyncSession { } impl PeerState { - pub(crate) fn new(leads: bool) -> Self { + pub(crate) fn new(leads: bool, timing: &Timing) -> Self { Self { leads, standing: Standing::default(), hello_state: HelloState::default(), hello_backoff: Backoff::new(HELLO_RETRY, HELLO_RETRY_MAX), - sync: SyncSession::new(), + sync: SyncSession::new(timing.sync_reply_backoff()), liveness: Liveness::Active, added_at: Instant::now(), last_seen: None, @@ -330,6 +340,8 @@ impl PeerState { last_resync_at: None, diverged_since: None, epoch: None, + epoch_seq: None, + run_id: None, reports_ready: None, syncs_with_us: None, reported_heads: None, @@ -342,7 +354,10 @@ impl PeerState { /// Start Automerge sync with this peer from scratch. A message still being handed to /// Mycelium was generated against the old state; its outcome is then ignored. pub(crate) fn reset_sync(&mut self) { - self.sync = SyncSession::new(); + // The same waits for answers, from the start. + let mut reply_backoff = self.sync.reply_backoff; + reply_backoff.reset(); + self.sync = SyncSession::new(reply_backoff); } /// An overdue sync answer from an established, active peer (see @@ -517,6 +532,12 @@ pub(crate) struct Timing { pub(crate) dormant_probe: Duration, /// While catching up: PING the source after this much silence (`swarm::source`). pub(crate) catch_up_probe: Duration, + /// First wait for an answer to a sync message before sending again; doubles while + /// answers stay out (`SyncSession::reply_backoff`). + pub(crate) sync_reply_timeout: Duration, + /// How long the transport tries to deliver a message before dropping it (Mycelium: + /// five minutes): a peer silent for longer has nothing on the way to us any more. + pub(crate) delivery_limit: Duration, } impl Timing { @@ -534,18 +555,34 @@ impl Timing { // WAN, mobile or metered links, or 100+ nodes | 60-120 s | At 100 nodes and 30 s it’s about 6–7 messages/s per node. Fine, but rarely worth it on battery or metered links. // Several hundred nodes or more | - | The full mesh becomes the problem in itself, not the interval. Message counts grow O(N²) across the swarm, which calls for partial views (SWIM-style random probing and syncing with a subset of peers). pub(crate) fn from_keepalive(keepalive: Duration) -> Self { + Self::new(keepalive, SYNC_REPLY_TIMEOUT) + } + + /// [`from_keepalive`](Self::from_keepalive) with another first sync reply timeout: + /// for tests, whose messages don't travel for seconds. + pub(crate) fn new(keepalive: Duration, sync_reply_timeout: Duration) -> Self { Self { keepalive, dormant_after: keepalive * 4, // dormant_after >= 4* keepalive to tolerate three lost PINGs in a row // >= 2 keepalives, so a second exchange confirms the difference; and >= twice // the first sync reply timeout, so a lost message's sender resends it first. - resync_after: (keepalive * 2).max(SYNC_REPLY_TIMEOUT * 2), + resync_after: (keepalive * 2).max(sync_reply_timeout * 2), dormant_probe: keepalive * 10, // >= 10*: dormant peers cost almost nothing, but still rejoin within minutes // Seconds, not a keepalive interval: a dead source should cost little time. // Only the source gets these, and only while catching up. catch_up_probe: (keepalive / 30).clamp(Duration::from_secs(1), Duration::from_secs(5)), + sync_reply_timeout, + delivery_limit: MYCELIUM_DELIVERY_LIMIT, } } + + /// Waits for sync answers: `sync_reply_timeout`, doubling up to 30 times as much. + pub(crate) fn sync_reply_backoff(&self) -> Backoff { + Backoff::new( + self.sync_reply_timeout, + self.sync_reply_timeout * SYNC_REPLY_TIMEOUT_MAX_FACTOR, + ) + } } #[derive(Clone)] @@ -583,8 +620,8 @@ pub(crate) struct AppState { /// Per-node remake state (prebuilt snapshots, switch status, archive). pub(crate) remake: Arc<RemakeState>, - /// Pre-shared key every member must prove in the handshake; None admits anyone who - /// knows the swarm ID. + /// Pre-shared key every member must prove in the handshake; None admits any node + /// that reaches us. pub(crate) swarm_key: Option<Arc<crate::swarm::auth::SwarmKey>>, /// Random per run of this node; our HELLOs carry it and replies must echo it. @@ -602,6 +639,12 @@ pub(crate) struct AppState { /// HTTP API users who may write protocol data through the store API /// (`--protocol-writers`). pub(crate) protocol_writers: Arc<std::collections::HashSet<String>>, + /// This node's own files, and how much it read and wrote them. + pub(crate) files: Arc<crate::files::Files>, + /// This node's document on disk (`--document-file`), if configured. + pub(crate) document_file: Option<Arc<crate::document::file::DocumentFile>>, + /// What this node remembers across restarts (`--state-file`), if configured. + pub(crate) state_file: Option<Arc<crate::swarm::state_file::StateFile>>, /// The effective configuration with sources, imported values left out /// (`config::Loaded::report`), for `GET /admin/config`. pub(crate) config: Arc<serde_json::Value>, @@ -618,6 +661,9 @@ pub(crate) struct AppState { pub(crate) announce_pending: Arc<std::sync::Mutex<Option<std::collections::HashSet<String>>>>, /// Our neighbours, once known (`swarm::topology`). pub(crate) neighbours: Arc<std::sync::RwLock<Option<crate::swarm::topology::NeighbourView>>>, + /// Something the neighbours are computed from may have changed: the member list or a + /// peer's reachability (`swarm::topology::run_worker`). + pub(crate) topology_changed: Arc<tokio::sync::Notify>, /// While rejoining a newer epoch, the previous document, served read-only (and /// marked stale) instead of answering 503. @@ -627,6 +673,9 @@ pub(crate) struct AppState { pub(crate) catch_up_source: Arc<std::sync::Mutex<Option<CatchUpSource>>>, /// Since when a closer peer has had a head start to become the source. pub(crate) catch_up_preference: Arc<std::sync::Mutex<Option<Instant>>>, + /// A catch-up source given up for not answering our sync messages, and until when it + /// isn't chosen again (`swarm::source`). + pub(crate) catch_up_avoided: Arc<std::sync::Mutex<Option<(String, Instant)>>>, } /// The peer a catching-up node syncs with first (`swarm::source`). @@ -635,6 +684,12 @@ pub(crate) struct CatchUpSource { pub(crate) peer: String, /// When it was chosen or last sent us anything. pub(crate) last_heard: Instant, + /// When it was chosen or last sent us a sync message: progress, not just liveness. + pub(crate) last_sync: Instant, + /// Our sync messages to it in a row whose answer didn't come in time. + pub(crate) reply_timeouts: u32, + /// When we last warned that catching up from it stalls. + pub(crate) last_warned: Option<Instant>, /// Our last probe PING to it. pub(crate) last_probe: Option<Instant>, /// Its keepalives in a row saying it wouldn't sync with us. @@ -670,6 +725,12 @@ impl AppState { *self.epoch_tx.borrow() } + /// The local epoch and `Replica::epoch_seq`, read together. + pub(crate) async fn epoch_stamp(&self) -> (Epoch, u32) { + let replica = self.doc.read().await; + (replica.epoch, replica.epoch_seq) + } + pub(crate) async fn is_ready(&self) -> bool { matches!(*self.lifecycle.read().await, NodeLifecycle::Ready { .. }) } @@ -779,7 +840,7 @@ impl AppState { /// State for a peer we haven't known so far (not inserted). pub(crate) fn new_peer_state(&self, pk: &str) -> Arc<Mutex<PeerState>> { let leads = crate::swarm::pair::leads(self.local_id(), pk); - Arc::new(Mutex::new(PeerState::new(leads))) + Arc::new(Mutex::new(PeerState::new(leads, &self.timing))) } pub(crate) fn notify_store_changed(&self) { @@ -1207,6 +1268,7 @@ impl Host for AppState { let new_doc = f(&mut replica.doc)?; let old_doc = std::mem::replace(&mut replica.doc, new_doc); replica.epoch = epoch; + replica.epoch_seq += 1; replica.refresh_heads(); // Updated under the same lock, so the cached epoch never disagrees // with the document for anyone holding the lock. diff --git a/src/swarm/blob.rs b/src/swarm/blob.rs index 8c891ae..42cfbf3 100644 --- a/src/swarm/blob.rs +++ b/src/swarm/blob.rs @@ -171,8 +171,10 @@ async fn prepare(state: &AppState, id: BlobId) -> Option<Arc<Prepared>> { } async fn reply(state: &AppState, peer: &str, id: BlobId, body: BlobBody) { + let (epoch, epoch_seq) = state.epoch_stamp().await; let message = ParsedMessage::Blob(BlobMessage { - epoch: state.local_epoch(), + epoch, + epoch_seq, id, body, }); @@ -223,8 +225,10 @@ impl Drop for Registration<'_> { } async fn request(state: &AppState, peer: &str, id: BlobId, chunks: Vec<u32>) { + let (epoch, epoch_seq) = state.epoch_stamp().await; let message = ParsedMessage::Blob(BlobMessage { - epoch: state.local_epoch(), + epoch, + epoch_seq, id, body: BlobBody::Request { chunks }, }); 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" + ); } diff --git a/src/swarm/members.rs b/src/swarm/members.rs index 221437e..895e0cc 100644 --- a/src/swarm/members.rs +++ b/src/swarm/members.rs @@ -25,7 +25,7 @@ use crate::state::{AppState, unix_millis}; const META_ROOT: &str = "_continuum"; -fn members_dir() -> Vec<String> { +pub(crate) fn members_dir() -> Vec<String> { vec![META_ROOT.to_owned(), "members".to_owned()] } diff --git a/src/swarm/mod.rs b/src/swarm/mod.rs index adb5d1c..d424529 100644 --- a/src/swarm/mod.rs +++ b/src/swarm/mod.rs @@ -11,6 +11,7 @@ pub(crate) mod relay; pub(crate) mod rtt; pub(crate) mod source; pub(crate) mod stability; +pub(crate) mod state_file; pub(crate) mod stats; pub(crate) mod sync; pub(crate) mod throughput; diff --git a/src/swarm/peer.rs b/src/swarm/peer.rs index 3d7da32..e8248bf 100644 --- a/src/swarm/peer.rs +++ b/src/swarm/peer.rs @@ -110,11 +110,12 @@ pub(crate) async fn send_ping(state: &AppState, peer: &str) -> Result<()> { pub(crate) async fn local_keepalive(state: &AppState, peer: &str) -> transport::Keepalive { // Heads and epoch come from the same locked replica, so they always match. The heads // are cached (`Replica`), so the read lock is enough. - let (heads, epoch) = { + let (heads, epoch, epoch_seq) = { let replica = state.doc.read().await; ( transport::HeadsSummary::of(replica.heads().to_vec()), replica.epoch, + replica.epoch_seq, ) }; let syncs_with_you = match state.get_peer_state(peer).await { @@ -129,6 +130,7 @@ pub(crate) async fn local_keepalive(state: &AppState, peer: &str) -> transport:: transport::Keepalive { heads, epoch, + epoch_seq, ready: state.is_ready().await, syncs_with_you, } @@ -276,6 +278,7 @@ async fn transmit_hello(state: &AppState, peer: &str) -> Result<()> { .as_ref() .map(|key| key.hello_tag(swarm_id.as_deref().unwrap_or(""), state.local_id(), peer)) .unwrap_or_default(); + let (epoch, epoch_seq) = state.epoch_stamp().await; state.lock_rtt().hello_sent(peer, Instant::now()); let result = state .transport @@ -283,7 +286,8 @@ async fn transmit_hello(state: &AppState, peer: &str) -> Result<()> { peer, &ParsedMessage::Hello(transport::HelloRequest { swarm_id, - epoch: state.local_epoch(), + epoch, + epoch_seq, auth, ready: state.is_ready().await, run_id: state.run_id, @@ -295,7 +299,11 @@ async fn transmit_hello(state: &AppState, peer: &str) -> Result<()> { let mut peer_state = peer_state_mutex.lock().await; if peer_state.hello_state == HelloState::Sending { + let was_unreachable = peer_state.unreachable_since.is_some(); peer_state.hello_sent(Instant::now()); + if !was_unreachable && peer_state.unreachable_since.is_some() { + state.topology_changed.notify_one(); + } let wait = match peer_state.liveness { Liveness::Dormant { .. } => state.timing.dormant_probe, Liveness::Active => peer_state.hello_backoff.next(), @@ -354,14 +362,17 @@ async fn send_hello_reply( .map(|key| key.hello_reply_tag(&swarm_id, state.local_id(), peer)) .unwrap_or_default(); let ready = state.is_ready().await; + let (epoch, epoch_seq) = state.epoch_stamp().await; let reply = transport::HelloReply { swarm_id, peers, - epoch: state.local_epoch(), + epoch, + epoch_seq, auth, ready, run_id, away_ms: away.map(|away| away.as_millis() as u64), + sender_run_id: state.run_id, }; debug!(to = %peer, ready, "Sending HELLO_REPLY"); @@ -399,15 +410,24 @@ async fn discover_peers(state: &AppState, from_peer: &str, advertised_peers: &[S } /// New neighbours get our HELLO (we may not have known them before: the member list is -/// how a partial mesh discovers them). Former neighbours get no more HELLOs, keepalives -/// or sync openings from us; their session is reset, so we don't keep pushing changes -/// to them either. They are still answered if they contact us. -async fn apply_neighbour_change(state: &AppState, change: topology::NeighbourChange) { +/// how a partial mesh discovers them) and a sync opening. The opening matters for a +/// neighbour whose handshake is already complete, e.g. one that returned from dormancy +/// while it wasn't our neighbour: it was only answered then, and with no HELLO to send, +/// nothing else would open sync until the keepalive heal (`resync_after`). Former +/// neighbours get no more HELLOs, keepalives or sync openings from us; their session is +/// reset, so we don't keep pushing changes to them either. They are still answered if +/// they contact us. +pub(crate) async fn apply_neighbour_change(state: &AppState, change: topology::NeighbourChange) { for peer in &change.added { remember_peer(state, peer).await; if let Err(error) = send_hello_once(state, peer).await { warn!(peer = %peer, error = %format!("{error:#}"), "HELLO to new neighbour failed; will retry"); } + // Skipped there unless the handshake is complete; otherwise its reply restarts + // sync anyway. + let _ = state + .sync_events + .send(SyncEvent::PeerNeedsSync(peer.clone())); } for peer in &change.removed { if let Some(peer_state) = state.get_peer_state(peer).await { @@ -427,12 +447,20 @@ pub(crate) async fn run_peer_worker(state: AppState) { info!("Starting peer maintenance worker"); let mut last_forget_check: Option<Instant> = None; + let mut was_ready = false; loop { tick.tick().await; // While joining, keep handshakes fast; dormancy would slow probes to minutes. let ready = state.is_ready().await; + if ready && !was_ready { + // Just became ready: register in the member list (and do the other periodic + // checks) now rather than at the next check. Until it is registered, other + // members don't count this node as a neighbour. + last_forget_check = None; + } + was_ready = ready; let peers: Vec<(String, Arc<Mutex<PeerState>>)> = state .peer_states @@ -461,12 +489,14 @@ pub(crate) async fn run_peer_worker(state: AppState) { if let Some(changes) = unanswered { debug!(peer = %peer, changes, "No answer to our sync message; sending again"); + crate::swarm::source::reply_timed_out(&state, peer); let _ = state .sync_events .send(SyncEvent::PeerNeedsSync(peer.clone())); } if became_dormant { + state.topology_changed.notify_one(); warn!( peer = %peer, silent_secs = state.timing.dormant_after.as_secs(), @@ -493,6 +523,8 @@ pub(crate) async fn run_peer_worker(state: AppState) { crate::swarm::members::ensure_registered(&state).await; crate::swarm::stability::tick(&state).await; crate::swarm::stats::measure_saved_size(&state); + crate::swarm::source::warn_if_stalled(&state); + crate::swarm::state_file::update(&state, false).await; { let known = state.peer_states.read().await; state.traffic.retain(|peer| known.contains_key(peer)); @@ -502,9 +534,8 @@ pub(crate) async fn run_peer_worker(state: AppState) { .unwrap_or_else(|poisoned| poisoned.into_inner()) .retain(|peer| known.contains_key(peer)); } - if let Some(change) = topology::refresh(&state).await { - apply_neighbour_change(&state, change).await; - } + // Fallback for what doesn't notify (`topology::run_worker`). + state.topology_changed.notify_one(); last_forget_check = Some(Instant::now()); } } @@ -581,9 +612,20 @@ pub(crate) async fn peer_swarm(state: &AppState, peer: &str) -> Option<String> { // --- epochs ------------------------------------------------------------------ -/// Records the epoch a peer advertised and compares it with ours. Returns true when both -/// are on the same epoch, the only case in which the documents may be synced. -async fn observe_peer_epoch(state: &AppState, peer: &str, remote: Epoch) -> bool { +/// Records the epoch a peer advertised and compares it with ours. Returns true when the +/// message's epoch is ours, the only case in which its documents may be synced. +/// +/// `seq` is the peer's `epoch_seq` with it. A message with a lower one than we have was +/// sent before the peer's last epoch change and delivered late: its epoch isn't +/// recorded, so it can't flip our view of the peer back. The count starts over when the +/// peer restarts, which its HELLO or HELLO_REPLY tells by a new run (`run`). +async fn observe_peer_epoch( + state: &AppState, + peer: &str, + remote: Epoch, + seq: u32, + run: Option<transport::RunId>, +) -> bool { let local = state.local_epoch(); let relation = local.relation_to(&remote); @@ -593,6 +635,21 @@ async fn observe_peer_epoch(state: &AppState, peer: &str, remote: Epoch) -> bool let previous = { let mut peer_state = peer_state_mutex.lock().await; + if let Some(run) = run.filter(|run| peer_state.run_id != Some(*run)) { + peer_state.run_id = Some(run); + peer_state.epoch_seq = None; + } + if peer_state.epoch_seq.is_some_and(|newest| seq < newest) { + debug!( + peer = %peer, + epoch = %remote, + seq, + newest = ?peer_state.epoch_seq, + "Ignoring a late message's epoch" + ); + return relation == EpochRelation::Same; + } + peer_state.epoch_seq = Some(seq); let previous = peer_state.epoch.replace(remote); if previous.is_some_and(|previous| previous != remote) { // The peer switched documents; our sync state describes its old one. Count @@ -749,6 +806,10 @@ pub(crate) async fn handle_message( let peer_state_mutex = state.get_or_create_peer_state(&peer).await; let mut peer_state = peer_state_mutex.lock().await; let previously_seen = peer_state.last_seen; + if peer_state.unreachable_since.is_some() { + // Reachable again (`mark_seen` clears it). + state.topology_changed.notify_one(); + } (peer_state.mark_seen(Instant::now()), previously_seen) }; crate::swarm::source::heard_from(state, &peer); @@ -757,7 +818,13 @@ pub(crate) async fn handle_message( info!(peer = %peer, "Dormant peer is active again; re-handshaking"); } - let same_epoch = observe_peer_epoch(state, &peer, message.epoch()).await; + let run = match &message { + transport::ParsedMessage::Hello(hello) => Some(hello.run_id), + transport::ParsedMessage::HelloReply(reply) => Some(reply.sender_run_id), + _ => None, + }; + let same_epoch = + observe_peer_epoch(state, &peer, message.epoch(), message.epoch_seq(), run).await; //debug!(from = %peer, message = ?message, "Received message"); @@ -807,6 +874,7 @@ pub(crate) async fn handle_message( debug!(peer = %peer, epoch = %envelope.epoch, "Dropping sync message from another epoch"); return Ok(()); } + crate::swarm::source::synced_from(state, &peer); handle_sync(state, &peer, envelope).await?; } transport::ParsedMessage::Ping(remote) => { diff --git a/src/swarm/source.rs b/src/swarm/source.rs index 378c40a..9eac445 100644 --- a/src/swarm/source.rs +++ b/src/swarm/source.rs @@ -15,12 +15,22 @@ //! is quiet for `Timing::catch_up_probe`, we PING it; after three unanswered probes, or //! when it says it isn't ready, or twice in a row that it wouldn't sync with us (the //! first time may be a handshake still under way), the next peer takes over. +//! +//! Alive isn't enough: a source that answers keepalives but not our sync messages (a +//! big answer lost on the way, a stuck sync state) would keep us catching up forever. +//! So it is also given up after three of our sync messages in a row went unanswered, +//! once nothing from it can still be on the way: no sync message from it for +//! `Timing::delivery_limit`, after which Mycelium has delivered or dropped anything it +//! sent before. It isn't chosen again for another `delivery_limit`, so another peer +//! takes over rather than the same one stalling again; if it is the only candidate, +//! catching up waits that long before trying it again. Meanwhile, every minute without a sync message from it is logged as a +//! warning. use std::sync::atomic::Ordering; use std::time::Duration; use tokio::time::Instant; -use tracing::{debug, info}; +use tracing::{debug, info, warn}; use crate::state::{AppState, CatchUpSource, PeerState}; use crate::swarm::sync::SyncEvent; @@ -35,6 +45,12 @@ const UNANSWERED_PROBES: u32 = 3; /// Keepalives in a row saying "I wouldn't sync with you" before the source is given up. const REFUSALS: u32 = 2; +/// Unanswered sync messages in a row before the source may be given up. +const REPLY_TIMEOUTS: u32 = 3; + +/// How often a stalled catch-up is logged. +const STALL_WARNING: Duration = Duration::from_secs(60); + fn lock_preference(state: &AppState) -> std::sync::MutexGuard<'_, Option<Instant>> { state .catch_up_preference @@ -42,6 +58,13 @@ fn lock_preference(state: &AppState) -> std::sync::MutexGuard<'_, Option<Instant .unwrap_or_else(|poisoned| poisoned.into_inner()) } +fn lock_avoided(state: &AppState) -> std::sync::MutexGuard<'_, Option<(String, Instant)>> { + state + .catch_up_avoided + .lock() + .unwrap_or_else(|poisoned| poisoned.into_inner()) +} + fn catching_up(state: &AppState) -> bool { !state.initial_sync_done.load(Ordering::Acquire) } @@ -60,12 +83,20 @@ pub(crate) fn defer(state: &AppState, peer: &str, peer_state: &PeerState) -> Opt return None; } let stall = state.timing.catch_up_probe * UNANSWERED_PROBES; + let now = Instant::now(); + let avoided_until = lock_avoided(state) + .as_ref() + .filter(|(avoided, until)| avoided == peer && now < *until) + .map(|(_, until)| *until); + if let Some(until) = avoided_until { + // Given up for not answering; the timer brings it back once it may be chosen. + return Some(until - now); + } if peer_state.reports_ready == Some(false) || peer_state.syncs_with_us == Some(false) { // Not a candidate; its next message re-evaluates this, the timer is a fallback. return Some(stall); } - let now = Instant::now(); let mut source = lock(state); let quiet = match &*source { Some(current) if current.peer == peer => return None, @@ -115,6 +146,9 @@ pub(crate) fn defer(state: &AppState, peer: &str, peer_state: &PeerState) -> Opt *source = Some(CatchUpSource { peer: peer.to_owned(), last_heard: now, + last_sync: now, + reply_timeouts: 0, + last_warned: None, last_probe: None, refusals: 0, }); @@ -128,6 +162,78 @@ pub(crate) fn heard_from(state: &AppState, peer: &str) { } } +/// A sync message from `peer` arrived: progress, if it is the source. +pub(crate) fn synced_from(state: &AppState, peer: &str) { + if let Some(current) = lock(state).as_mut().filter(|current| current.peer == peer) { + current.last_sync = Instant::now(); + current.reply_timeouts = 0; + } +} + +/// Our sync message to `peer` went unanswered (`SyncSession` reply deadline). Gives up +/// the source after `REPLY_TIMEOUTS` in a row, once nothing from it can be on the way. +pub(crate) fn reply_timed_out(state: &AppState, peer: &str) { + if !catching_up(state) { + return; + } + let given_up = { + let mut source = lock(state); + let Some(current) = source.as_mut().filter(|current| current.peer == peer) else { + return; + }; + current.reply_timeouts += 1; + let silent = current.last_sync.elapsed(); + let give_up = + current.reply_timeouts >= REPLY_TIMEOUTS && silent >= state.timing.delivery_limit; + let timeouts = current.reply_timeouts; + if give_up { + *source = None; + *lock_avoided(state) = Some(( + peer.to_owned(), + Instant::now() + state.timing.delivery_limit, + )); + } + give_up.then_some((timeouts, silent)) + }; + if let Some((timeouts, silent)) = given_up { + warn!( + peer = %peer, + reply_timeouts = timeouts, + silent_secs = silent.as_secs(), + "Catch-up source doesn't answer our sync messages; catching up from another peer" + ); + resync_everyone(state); + } +} + +/// Warns, at most every `STALL_WARNING`, while the source sends no sync messages. Called +/// by the peer worker on its maintenance tick. +pub(crate) fn warn_if_stalled(state: &AppState) { + if !catching_up(state) { + return; + } + let mut source = lock(state); + let Some(current) = source.as_mut() else { + return; + }; + let silent = current.last_sync.elapsed(); + if silent < STALL_WARNING + || current + .last_warned + .is_some_and(|at| at.elapsed() < STALL_WARNING) + { + return; + } + current.last_warned = Some(Instant::now()); + warn!( + peer = %current.peer, + silent_secs = silent.as_secs(), + reply_timeouts = current.reply_timeouts, + heard_secs_ago = current.last_heard.elapsed().as_secs(), + "Catching up stalls: no sync message from the catch-up source" + ); +} + /// What `peer` said about itself, in a sync message (`syncs_with_us: None`) or a /// keepalive. Gives up the source if that rules it out. pub(crate) fn reported(state: &AppState, peer: &str, ready: bool, syncs_with_us: Option<bool>) { @@ -188,6 +294,7 @@ pub(crate) fn probe_due(state: &AppState, peer: &str, now: Instant) -> bool { pub(crate) fn finished(state: &AppState, from: &str) { *lock(state) = None; *lock_preference(state) = None; + *lock_avoided(state) = None; let state = state.clone(); let from = from.to_owned(); tokio::spawn(async move { diff --git a/src/swarm/state_file.rs b/src/swarm/state_file.rs new file mode 100644 index 0000000..62a6dbe --- /dev/null +++ b/src/swarm/state_file.rs @@ -0,0 +1,185 @@ +//! The state file (`--state-file`): what a node remembers across restarts, so it +//! rejoins its swarm on its own: the swarm ID and some members to reach it through. +//! Not the document: a restarted node still starts empty and syncs. +//! +//! Written rarely: once the node is ready, then when the remembered members change +//! (checked every few minutes, written at most once an hour), and at shutdown if they +//! changed since, atomically (`files`). + +use std::path::PathBuf; +use std::sync::{Arc, Mutex, MutexGuard}; +use std::time::Duration; + +use serde::{Deserialize, Serialize}; +use tokio::time::Instant; +use tracing::{error, info, warn}; + +use crate::files::{FileKind, Files}; +use crate::state::AppState; +use crate::swarm::members; + +/// Members remembered as seeds. +const MAX_SEEDS: usize = 16; +/// How often the remembered members are compared with the registry. +const CHECK_INTERVAL: Duration = Duration::from_secs(300); +/// Writes at least this far apart, but for the first one and at shutdown. +const MIN_WRITE_INTERVAL: Duration = Duration::from_secs(3600); + +/// The file's content. +#[derive(Clone, Debug, Default, PartialEq, Eq, Serialize, Deserialize)] +#[serde(rename_all = "camelCase")] +pub(crate) struct Remembered { + pub(crate) swarm_id: String, + /// Public keys of members to join through (sorted). + #[serde(default)] + pub(crate) seeds: Vec<String>, +} + +pub(crate) struct StateFile { + path: PathBuf, + files: Arc<Files>, + // A std mutex: never held across an await. + inner: Mutex<Inner>, +} + +#[derive(Default)] +struct Inner { + written: Option<Remembered>, + written_at: Option<Instant>, + checked_at: Option<Instant>, +} + +impl StateFile { + /// Opens `path`, returning what it remembers. A missing file remembers nothing; an + /// unreadable one is moved aside (`.bad`) and logged, and remembers nothing either. + pub(crate) fn open(path: PathBuf, files: Arc<Files>) -> (Self, Option<Remembered>) { + let remembered = match files.read(FileKind::State, &path) { + Ok(Some(bytes)) => match serde_json::from_slice::<Remembered>(&bytes) { + Ok(remembered) if !remembered.swarm_id.is_empty() => Some(remembered), + Ok(_) => None, + Err(error) => { + error!( + path = %path.display(), + error = %error, + "Unreadable state file; moving it aside and remembering nothing" + ); + match files.move_aside(&path, ".bad") { + Ok(aside) => { + warn!(moved_to = %aside.display(), "Moved the state file aside") + } + Err(error) => warn!(error = %error, "Couldn't move the state file aside"), + } + None + } + }, + Ok(None) => None, + Err(error) => { + error!(path = %path.display(), error = %error, "Couldn't read the state file; remembering nothing"); + None + } + }; + let file = Self { + path, + files, + inner: Mutex::new(Inner { + written: remembered.clone(), + ..Inner::default() + }), + }; + (file, remembered) + } + + fn lock(&self) -> MutexGuard<'_, Inner> { + self.inner + .lock() + .unwrap_or_else(|poisoned| poisoned.into_inner()) + } + + fn write(&self, remembered: &Remembered) -> std::io::Result<()> { + let mut bytes = serde_json::to_vec_pretty(remembered)?; + bytes.push(b'\n'); + self.files.write_atomic(FileKind::State, &self.path, &bytes) + } +} + +/// Brings the state file up to date if it is due: `now` for the first write and at +/// shutdown, otherwise only every few minutes and at most once an hour. Called by the +/// peer worker on its maintenance tick. +pub(crate) async fn update(state: &AppState, now: bool) { + let Some(file) = state.state_file.clone() else { + return; + }; + if !state.is_ready().await { + return; + } + let Some(swarm_id) = state.swarm_id().await else { + return; + }; + { + let mut inner = file.lock(); + if !now + && inner + .checked_at + .is_some_and(|at| at.elapsed() < CHECK_INTERVAL) + { + return; + } + inner.checked_at = Some(Instant::now()); + } + + let seeds = { + let replica = state.doc.read().await; + seeds(&replica.doc, state.local_id(), state.electorate_window) + }; + let remembered = Remembered { swarm_id, seeds }; + + { + let inner = file.lock(); + if inner.written.as_ref() == Some(&remembered) { + return; + } + let first = inner.written_at.is_none(); + if !now + && !first + && inner + .written_at + .is_some_and(|at| at.elapsed() < MIN_WRITE_INTERVAL) + { + return; + } + } + + let writer = Arc::clone(&file); + let content = remembered.clone(); + match tokio::task::spawn_blocking(move || writer.write(&content)).await { + Ok(Ok(())) => { + info!( + path = %file.path.display(), + seeds = remembered.seeds.len(), + "Wrote the state file" + ); + let mut inner = file.lock(); + inner.written = Some(remembered); + inner.written_at = Some(Instant::now()); + } + Ok(Err(error)) => { + error!(path = %file.path.display(), error = %error, "Couldn't write the state file"); + } + Err(error) => error!(error = %error, "State file write task failed"), + } +} + +/// Members to rejoin through: those alive within the electorate window, stable ones +/// first, at most `MAX_SEEDS`, sorted. Changes only when members join, leave or change +/// tier, so the file rarely does. +fn seeds(doc: &automerge::AutoCommit, me: &str, window: Duration) -> Vec<String> { + let now = crate::state::unix_millis(); + let (mut stable, transient): (Vec<String>, Vec<String>) = members::electorate(doc, now, window) + .into_iter() + .filter(|pk| pk != me) + .partition(|pk| members::tier(doc, pk) == members::Tier::Stable); + stable.extend(transient); + stable.truncate(MAX_SEEDS); + stable.sort(); + stable +} diff --git a/src/swarm/sync.rs b/src/swarm/sync.rs index 80d6d56..d4503bb 100644 --- a/src/swarm/sync.rs +++ b/src/swarm/sync.rs @@ -141,10 +141,10 @@ async fn deliver_sync_to_peer(state: &AppState, peer: &str) -> Result<()> { tokio::task::spawn_blocking(move || -> Result<Option<transport::EncodedMessage>> { // Label the message with the epoch of the document it was generated // from, read under the same lock. - let (epoch, message) = { + let (epoch, epoch_seq, message) = { let mut replica = doc.blocking_write(); let mut peer_state = psm.blocking_lock(); - let epoch = replica.epoch; + let (epoch, epoch_seq) = (replica.epoch, replica.epoch_seq); let sent_before = peer_state.sync.state.sent_hashes.clone(); let message = { let sync_doc = replica.doc.sync(); @@ -178,7 +178,7 @@ async fn deliver_sync_to_peer(state: &AppState, peer: &str) -> Result<()> { regenerate: false, }); } - (epoch, message) + (epoch, epoch_seq, message) }; let Some(message) = message else { return Ok(None); @@ -186,6 +186,7 @@ async fn deliver_sync_to_peer(state: &AppState, peer: &str) -> Result<()> { let envelope = transport::SyncEnvelope { swarm_id, epoch, + epoch_seq, message, ready, }; @@ -607,6 +608,13 @@ pub(crate) async fn run_change_publisher(state: AppState) { warn!(path = ?path, "Concurrent writes produced a conflict"); } + if paths + .iter() + .any(|path| crate::swarm::topology::affects_neighbours(path)) + { + state.topology_changed.notify_one(); + } + // Fails only when nobody is subscribed. let _ = state.change_events.send(Arc::new(ChangeSet { paths })); } diff --git a/src/swarm/topology.rs b/src/swarm/topology.rs index 57856b6..b40b120 100644 --- a/src/swarm/topology.rs +++ b/src/swarm/topology.rs @@ -107,6 +107,26 @@ pub(crate) struct NeighbourChange { pub(crate) removed: Vec<String>, } +/// Recomputes the neighbours whenever something they depend on may have changed +/// (`AppState::topology_changed`): the member list (`swarm::sync::run_change_publisher`), +/// a peer becoming unreachable or reachable again (`swarm::peer`), and every 5 s as a +/// fallback, which also covers unreachability expiring after `dormant_probe`. A burst of +/// notifications coalesces into one recomputation. +pub(crate) async fn run_worker(state: AppState) { + loop { + state.topology_changed.notified().await; + if let Some(change) = refresh(&state).await { + crate::swarm::peer::apply_neighbour_change(&state, change).await; + } + } +} + +/// Whether a change at `path` may change the neighbours: one in the member list, or the +/// root (the whole document replaced, as after an epoch switch). +pub(crate) fn affects_neighbours(path: &[String]) -> bool { + path.is_empty() || path.starts_with(&members::members_dir()) +} + /// Recomputes the neighbours once ready, if the member list, the reachable members or /// `k` changed. pub(crate) async fn refresh(state: &AppState) -> Option<NeighbourChange> { diff --git a/src/transport/mod.rs b/src/transport/mod.rs index c5769ca..c3db962 100644 --- a/src/transport/mod.rs +++ b/src/transport/mod.rs @@ -11,6 +11,8 @@ pub use crate::document::replica::Epoch; pub mod firewall; pub mod mycelium; +#[cfg(test)] +pub(crate) mod sim; pub mod traffic; /// Random identifier of one run of a node (from start to exit). @@ -86,6 +88,8 @@ pub struct Keepalive { /// Only comparable between nodes on the same epoch. pub heads: HeadsSummary, pub epoch: Epoch, + /// See `SyncEnvelope::epoch_seq`. + pub epoch_seq: u32, /// The sender is ready (same meaning as `SyncEnvelope::ready`). pub ready: bool, /// The sender would sync with the recipient: counts it as an active member on its @@ -131,6 +135,8 @@ impl std::fmt::Display for BlobId { pub struct BlobMessage { /// The sender's epoch (every message carries it). pub epoch: Epoch, + /// See `SyncEnvelope::epoch_seq`. + pub epoch_seq: u32, pub id: BlobId, pub body: BlobBody, } @@ -215,6 +221,17 @@ impl ParsedMessage { ParsedMessage::Blob(blob) => blob.epoch, } } + + /// `SyncEnvelope::epoch_seq`, as every message carries it. + pub fn epoch_seq(&self) -> u32 { + match self { + ParsedMessage::Hello(hello) => hello.epoch_seq, + ParsedMessage::HelloReply(reply) => reply.epoch_seq, + ParsedMessage::Sync(envelope) => envelope.epoch_seq, + ParsedMessage::Ping(keepalive) | ParsedMessage::Pong(keepalive) => keepalive.epoch_seq, + ParsedMessage::Blob(blob) => blob.epoch_seq, + } + } } #[derive(Debug, Clone)] @@ -223,6 +240,8 @@ pub struct HelloRequest { /// another swarm doesn't answer. pub swarm_id: Option<String>, pub epoch: Epoch, + /// See `SyncEnvelope::epoch_seq`. + pub epoch_seq: u32, /// Proof of the swarm key (`swarm::auth`); empty without one. pub auth: Vec<u8>, /// The sender is ready (same meaning as `SyncEnvelope::ready`). @@ -236,6 +255,8 @@ pub struct HelloReply { pub swarm_id: String, pub peers: Vec<String>, pub epoch: Epoch, + /// See `SyncEnvelope::epoch_seq`. + pub epoch_seq: u32, /// Proof of the swarm key (`swarm::auth`); empty without one. pub auth: Vec<u8>, /// The sender is ready. @@ -246,6 +267,8 @@ pub struct HelloReply { /// How long before this HELLO the sender had last heard from the recipient, if ever /// (in milliseconds): tells a restarted node how long it was away. pub away_ms: Option<u64>, + /// The sender's own run (as its HELLOs carry it): a new one means it restarted. + pub sender_run_id: RunId, } #[derive(Debug, Clone)] @@ -253,6 +276,10 @@ pub struct SyncEnvelope { pub swarm_id: String, /// Epoch of the document the message was generated from. pub epoch: Epoch, + /// How many times the sender changed epoch in its current run, when it sent this + /// (`Replica::epoch_seq`). Messages may arrive late and out of order: the receiver + /// ignores an epoch older than the newest it has seen in the same session. + pub epoch_seq: u32, pub message: Message, /// The sender was ready when it generated the message: its document is the swarm's, /// not one it is still filling (joining, rejoining). Only such a peer can tell a diff --git a/src/transport/mycelium.rs b/src/transport/mycelium.rs index 82e6016..efab475 100644 --- a/src/transport/mycelium.rs +++ b/src/transport/mycelium.rs @@ -561,13 +561,7 @@ impl Transport for MyceliumTransport { } fn encode(&self, peer: &str, message: &ParsedMessage) -> Result<EncodedMessage> { - let (outbound, payload_len) = format_message(peer, message)?; - Ok(EncodedMessage { - peer: peer.to_owned(), - body: serde_json::to_vec(&outbound)?.into(), - kind: message.kind(), - payload_len, - }) + encode_message(peer, message) } async fn send_encoded(&self, message: &EncodedMessage) -> Result<()> { @@ -712,6 +706,26 @@ impl Transport for MyceliumTransport { } } +/// A message as the Mycelium REST API takes it (the JSON body of an outbound message). +/// Also the simulator's wire format (`transport::sim`), so tests go through the same +/// encoding. +pub(crate) fn encode_message(peer: &str, message: &ParsedMessage) -> Result<EncodedMessage> { + let (outbound, payload_len) = format_message(peer, message)?; + Ok(EncodedMessage { + peer: peer.to_owned(), + body: serde_json::to_vec(&outbound)?.into(), + kind: message.kind(), + payload_len, + }) +} + +/// The message in a body made by [`encode_message`]. +#[cfg(test)] +pub(crate) fn decode_message(body: &[u8]) -> Result<ParsedMessage> { + let outbound: OutboundMessage = serde_json::from_slice(body)?; + Ok(decode_payload(&outbound.topic, &outbound.payload)?) +} + fn encode_topic(topic: &str) -> String { BASE64.encode(topic.as_bytes()) } @@ -802,16 +816,20 @@ fn format_message( struct EpochShadow { generation: u64, id: ByteBuf, + /// The sender's epoch changes in its current run (`SyncEnvelope::epoch_seq`). + #[serde(default)] + seq: u32, } -fn epoch_to_shadow(epoch: &Epoch) -> EpochShadow { +fn epoch_to_shadow(epoch: &Epoch, seq: u32) -> EpochShadow { EpochShadow { generation: epoch.generation, id: ByteBuf::from(epoch.id.to_vec()), + seq, } } -fn epoch_from_shadow(shadow: EpochShadow) -> Result<Epoch, TransportError> { +fn epoch_from_shadow(shadow: &EpochShadow) -> Result<Epoch, TransportError> { if shadow.generation == 0 && shadow.id.is_empty() { return Ok(Epoch::GENESIS); } @@ -850,7 +868,7 @@ fn keepalive_to_bytes(msg: &Keepalive) -> Result<Vec<u8>, rmp_serde::encode::Err rmp_serde::to_vec(&KeepaliveShadow { heads, digest, - epoch: epoch_to_shadow(&msg.epoch), + epoch: epoch_to_shadow(&msg.epoch, msg.epoch_seq), ready: msg.ready, syncs_with_you: msg.syncs_with_you, }) @@ -880,7 +898,8 @@ fn bytes_to_keepalive(bytes: &[u8]) -> Result<Keepalive, TransportError> { }; Ok(Keepalive { heads, - epoch: epoch_from_shadow(shadow.epoch)?, + epoch: epoch_from_shadow(&shadow.epoch)?, + epoch_seq: shadow.epoch.seq, ready: shadow.ready, syncs_with_you: shadow.syncs_with_you, }) @@ -908,6 +927,7 @@ pub struct HelloReplyShadow<'a> { ready: bool, run_id: ByteBuf, away_ms: Option<u64>, + sender_run_id: ByteBuf, } #[derive(Serialize, Deserialize)] @@ -955,7 +975,7 @@ mod automerge_msg_codec { pub fn hello_request_to_bytes(msg: &HelloRequest) -> Result<Vec<u8>, rmp_serde::encode::Error> { rmp_serde::to_vec(&HelloRequestShadow { swarm_id: msg.swarm_id.clone(), - epoch: epoch_to_shadow(&msg.epoch), + epoch: epoch_to_shadow(&msg.epoch, msg.epoch_seq), auth: ByteBuf::from(msg.auth.clone()), ready: msg.ready, run_id: ByteBuf::from(msg.run_id.to_vec()), @@ -965,7 +985,8 @@ pub fn bytes_to_hello_request(bytes: &[u8]) -> Result<HelloRequest, TransportErr let shadow: HelloRequestShadow = rmp_serde::from_slice(bytes)?; Ok(HelloRequest { swarm_id: shadow.swarm_id, - epoch: epoch_from_shadow(shadow.epoch)?, + epoch: epoch_from_shadow(&shadow.epoch)?, + epoch_seq: shadow.epoch.seq, auth: shadow.auth.into_vec(), ready: shadow.ready, run_id: run_id_from_shadow(&shadow.run_id)?, @@ -984,11 +1005,12 @@ pub fn hello_reply_to_bytes(msg: &HelloReply) -> Result<Vec<u8>, rmp_serde::enco let shadow = HelloReplyShadow { swarm_id: Cow::Borrowed(&msg.swarm_id), peers: msg.peers.clone(), // Vec must be cloned for serialization wrapper matching - epoch: epoch_to_shadow(&msg.epoch), + epoch: epoch_to_shadow(&msg.epoch, msg.epoch_seq), auth: ByteBuf::from(msg.auth.clone()), ready: msg.ready, run_id: ByteBuf::from(msg.run_id.to_vec()), away_ms: msg.away_ms, + sender_run_id: ByteBuf::from(msg.sender_run_id.to_vec()), }; rmp_serde::to_vec(&shadow) } @@ -997,11 +1019,13 @@ pub fn bytes_to_hello_reply(bytes: &[u8]) -> Result<HelloReply, TransportError> Ok(HelloReply { swarm_id: shadow.swarm_id.into_owned(), peers: shadow.peers, - epoch: epoch_from_shadow(shadow.epoch)?, + epoch: epoch_from_shadow(&shadow.epoch)?, + epoch_seq: shadow.epoch.seq, auth: shadow.auth.into_vec(), ready: shadow.ready, run_id: run_id_from_shadow(&shadow.run_id)?, away_ms: shadow.away_ms, + sender_run_id: run_id_from_shadow(&shadow.sender_run_id)?, }) } @@ -1010,7 +1034,7 @@ pub fn sync_envelope_to_bytes(msg: &SyncEnvelope) -> Result<Vec<u8>, rmp_serde:: let shadow = SyncEnvelopeShadow { swarm_id: Cow::Borrowed(&msg.swarm_id), message: msg.message.clone(), // Clone required by automerge_msg_codec ownership rules - epoch: epoch_to_shadow(&msg.epoch), + epoch: epoch_to_shadow(&msg.epoch, msg.epoch_seq), ready: msg.ready, }; rmp_serde::to_vec(&shadow) @@ -1019,7 +1043,8 @@ pub fn bytes_to_sync_envelope(bytes: &[u8]) -> Result<SyncEnvelope, TransportErr let shadow: SyncEnvelopeShadow = rmp_serde::from_slice(bytes)?; Ok(SyncEnvelope { swarm_id: shadow.swarm_id.into_owned(), - epoch: epoch_from_shadow(shadow.epoch)?, + epoch: epoch_from_shadow(&shadow.epoch)?, + epoch_seq: shadow.epoch.seq, message: shadow.message, ready: shadow.ready, }) @@ -1084,7 +1109,7 @@ fn blob_to_bytes(msg: &BlobMessage) -> Result<Vec<u8>, rmp_serde::encode::Error> BlobBody::Missing => BlobBodyShadow::Missing, }; rmp_serde::to_vec(&BlobShadow { - epoch: epoch_to_shadow(&msg.epoch), + epoch: epoch_to_shadow(&msg.epoch, msg.epoch_seq), id, body, }) @@ -1123,7 +1148,8 @@ fn bytes_to_blob(bytes: &[u8]) -> Result<BlobMessage, TransportError> { BlobBodyShadow::Missing => BlobBody::Missing, }; Ok(BlobMessage { - epoch: epoch_from_shadow(shadow.epoch)?, + epoch: epoch_from_shadow(&shadow.epoch)?, + epoch_seq: shadow.epoch.seq, id, body, }) @@ -1181,6 +1207,7 @@ mod tests { generation: u64::MAX, id: [0xff; 32], }, + epoch_seq: u32::MAX, ready: true, syncs_with_you: true, }; @@ -1203,6 +1230,7 @@ mod tests { let id = BlobId::Change(ChangeHash([9; 32])); let chunk = BlobMessage { epoch, + epoch_seq: 4, id, body: BlobBody::Chunk { index: 5, @@ -1215,12 +1243,14 @@ mod tests { let decoded = bytes_to_blob(&bytes).unwrap(); assert_eq!(decoded.id, id); assert_eq!(decoded.epoch, epoch); + assert_eq!(decoded.epoch_seq, 4); assert!( matches!(decoded.body, BlobBody::Chunk { index: 5, ref data } if data.len() == 64 * 1024) ); let manifest = BlobMessage { epoch, + epoch_seq: 4, id, body: BlobBody::Manifest { size: 100, diff --git a/src/transport/sim.rs b/src/transport/sim.rs new file mode 100644 index 0000000..1556d69 --- /dev/null +++ b/src/transport/sim.rs @@ -0,0 +1,448 @@ +//! A simulated network for tests: many nodes in one process, each with a +//! [`SimTransport`] on a shared [`SimNet`]. +//! +//! Messages go through the Mycelium wire encoding (`mycelium::encode_message`), so the +//! codec is exercised as in production. The net delays, loses and duplicates them per +//! link, from a seeded generator; nodes can be taken down (everything to and from them +//! is lost) and brought up again; and a test can inject a message as if a node had sent +//! it, e.g. one that arrives late. +//! +//! Time is tokio's real clock, not a paused one: a node's document work runs on the +//! blocking pool (`spawn_blocking`), which a paused clock doesn't wait for, so timers +//! would fire while the work is still running. Tests use a short keepalive instead. +//! Delays and losses come from the seed; when a message's turn comes still depends on +//! the scheduler, so a run is reproducible in its network decisions, not in its +//! interleavings. + +use std::collections::{HashMap, HashSet}; +use std::sync::{Arc, Mutex, MutexGuard}; +use std::time::Duration; + +use anyhow::{Result, bail}; +use async_trait::async_trait; +use bytes::Bytes; +use tokio::sync::{mpsc, oneshot}; +use tracing::{debug, warn}; + +use super::mycelium::{decode_message, encode_message}; +use super::{EncodedMessage, MessageHandler, MessageKind, ParsedMessage, Transport}; + +/// How one direction of a link behaves. +#[derive(Clone, Debug)] +pub(crate) struct LinkParams { + /// Probability that a message is lost. + pub(crate) loss: f64, + /// Probability that a message arrives twice (each copy with its own delay). + pub(crate) duplicate: f64, + /// Delay, uniform in this range. Overlapping delays reorder messages. + pub(crate) delay: (Duration, Duration), + /// Messages of these kinds are always lost: a peer that answers some messages but + /// not others (e.g. keepalives but not sync). + pub(crate) drop: Vec<MessageKind>, +} + +impl Default for LinkParams { + /// A clean, fast link: nothing lost, a few milliseconds. + fn default() -> Self { + Self { + loss: 0.0, + duplicate: 0.0, + delay: (Duration::from_millis(1), Duration::from_millis(5)), + drop: Vec::new(), + } + } +} + +/// A message on its way into a node. +struct Delivery { + from: String, + body: Bytes, + payload_len: usize, + /// Told once the node has handled the message (injections wait for it). + handled: Option<oneshot::Sender<()>>, +} + +/// What the net counted, for assertions. +#[derive(Clone, Copy, Debug, Default, PartialEq, Eq)] +pub(crate) struct NetCounts { + pub(crate) sent: u64, + pub(crate) delivered: u64, + pub(crate) lost: u64, + pub(crate) duplicated: u64, +} + +struct NetInner { + rng: SplitMix64, + default_link: LinkParams, + links: HashMap<(String, String), LinkParams>, + down: HashSet<String>, + inboxes: HashMap<String, mpsc::UnboundedSender<Delivery>>, + counts: NetCounts, +} + +/// The network all simulated nodes share. Cloning is cheap. +#[derive(Clone)] +pub(crate) struct SimNet { + inner: Arc<Mutex<NetInner>>, +} + +impl SimNet { + pub(crate) fn new(seed: u64) -> Self { + Self { + inner: Arc::new(Mutex::new(NetInner { + rng: SplitMix64(seed), + default_link: LinkParams::default(), + links: HashMap::new(), + down: HashSet::new(), + inboxes: HashMap::new(), + counts: NetCounts::default(), + })), + } + } + + fn lock(&self) -> MutexGuard<'_, NetInner> { + self.inner + .lock() + .unwrap_or_else(|poisoned| poisoned.into_inner()) + } + + /// A transport for a new node `id` on this net. + pub(crate) fn transport(&self, id: &str) -> Arc<SimTransport> { + let (tx, rx) = mpsc::unbounded_channel(); + let previous = self.lock().inboxes.insert(id.to_owned(), tx); + assert!(previous.is_none(), "node {id} is already on the net"); + Arc::new(SimTransport { + id: id.to_owned(), + net: self.clone(), + inbox: tokio::sync::Mutex::new(Some(rx)), + }) + } + + /// How links without their own parameters behave. + pub(crate) fn set_default_link(&self, params: LinkParams) { + self.lock().default_link = params; + } + + /// How messages from `from` to `to` are carried (one direction). + pub(crate) fn set_link(&self, from: &str, to: &str, params: LinkParams) { + self.lock() + .links + .insert((from.to_owned(), to.to_owned()), params); + } + + /// Everything to and from `id` is lost until [`bring_up`](Self::bring_up), including + /// messages already on their way. The node itself keeps running. + pub(crate) fn take_down(&self, id: &str) { + self.lock().down.insert(id.to_owned()); + } + + pub(crate) fn bring_up(&self, id: &str) { + self.lock().down.remove(id); + } + + pub(crate) fn counts(&self) -> NetCounts { + self.lock().counts + } + + /// Hands a message to the net: decides now whether and when it arrives. + fn send(&self, from: &str, message: &EncodedMessage) { + let delays = { + let mut guard = self.lock(); + let inner = &mut *guard; + inner.counts.sent += 1; + if inner.down.contains(from) || inner.down.contains(&message.peer) { + inner.counts.lost += 1; + return; + } + let params = inner + .links + .get(&(from.to_owned(), message.peer.clone())) + .unwrap_or(&inner.default_link) + .clone(); + if params.drop.contains(&message.kind) || inner.rng.chance(params.loss) { + inner.counts.lost += 1; + return; + } + let copies = if inner.rng.chance(params.duplicate) { + inner.counts.duplicated += 1; + 2 + } else { + 1 + }; + (0..copies) + .map(|_| inner.rng.duration_in(params.delay)) + .collect::<Vec<_>>() + }; + for delay in delays { + let net = self.clone(); + let delivery = Delivery { + from: from.to_owned(), + body: message.body.clone(), + payload_len: message.payload_len, + handled: None, + }; + let to = message.peer.clone(); + tokio::spawn(async move { + tokio::time::sleep(delay).await; + net.arrive(&to, delivery); + }); + } + } + + /// The message reaches `to`, unless either end went down meanwhile. + fn arrive(&self, to: &str, delivery: Delivery) { + let mut guard = self.lock(); + let inner = &mut *guard; + if inner.down.contains(to) || inner.down.contains(&delivery.from) { + inner.counts.lost += 1; + return; + } + // An unknown recipient (an address no node has) loses the message, as + // Mycelium would. + let Some(inbox) = inner.inboxes.get(to) else { + inner.counts.lost += 1; + return; + }; + if inbox.send(delivery).is_ok() { + inner.counts.delivered += 1; + } else { + inner.counts.lost += 1; + } + } + + /// Delivers `message` to `to` as if `from` had sent it, now, bypassing links and + /// `take_down`; returns once `to` has handled it. `from` needn't be a node on the + /// net. For messages a real node wouldn't send at that moment, e.g. one sent long + /// ago and delivered late. + pub(crate) async fn inject(&self, from: &str, to: &str, message: &ParsedMessage) -> Result<()> { + let encoded = encode_message(to, message)?; + let (handled_tx, handled_rx) = oneshot::channel(); + let delivery = Delivery { + from: from.to_owned(), + body: encoded.body, + payload_len: encoded.payload_len, + handled: Some(handled_tx), + }; + { + let inner = self.lock(); + let Some(inbox) = inner.inboxes.get(to) else { + bail!("no node {to} on the net"); + }; + if inbox.send(delivery).is_err() { + bail!("node {to} has no consumer running"); + } + } + handled_rx.await?; + Ok(()) + } + + fn is_known(&self, id: &str) -> bool { + self.lock().inboxes.contains_key(id) + } +} + +/// One node's end of the [`SimNet`]. Its identity is the ID it was created with; seeds +/// are node IDs. +pub(crate) struct SimTransport { + id: String, + net: SimNet, + inbox: tokio::sync::Mutex<Option<mpsc::UnboundedReceiver<Delivery>>>, +} + +#[async_trait] +impl Transport for SimTransport { + fn local_id(&self) -> &str { + &self.id + } + + /// A seed is a node ID; one no node has doesn't resolve (yet). + async fn resolve_seed(&self, seed: &str) -> Result<String> { + if self.net.is_known(seed) { + Ok(seed.to_owned()) + } else { + bail!("no node {seed} on the net") + } + } + + fn encode(&self, peer: &str, message: &ParsedMessage) -> Result<EncodedMessage> { + encode_message(peer, message) + } + + /// Always accepted: the net loses messages silently, like Mycelium once it has + /// taken them. + async fn send_encoded(&self, message: &EncodedMessage) -> Result<()> { + self.net.send(&self.id, message); + Ok(()) + } + + /// Handles messages one at a time, in arrival order, as the Mycelium consumer does. + async fn start_consumer(&self, handler: MessageHandler) -> Result<()> { + let Some(mut inbox) = self.inbox.lock().await.take() else { + bail!("consumer of {} already started", self.id); + }; + while let Some(delivery) = inbox.recv().await { + match decode_message(&delivery.body) { + Ok(message) => { + debug!(node = %self.id, from = %delivery.from, message = ?message, "Simulated delivery"); + if let Err(error) = handler(delivery.from, message, delivery.payload_len).await + { + warn!(node = %self.id, error = %format!("{error:#}"), "Failed to process inbound message"); + } + } + Err(error) => { + warn!(node = %self.id, error = %format!("{error:#}"), "Undecodable simulated message"); + } + } + if let Some(handled) = delivery.handled { + let _ = handled.send(()); + } + } + Ok(()) + } + + /// A simulated node has no daemon that could exit. + async fn wait_for_exit(&self) -> std::io::Result<std::process::ExitStatus> { + std::future::pending().await + } +} + +/// A small, seedable generator (SplitMix64): enough for network decisions, and no +/// dependency. +struct SplitMix64(u64); + +impl SplitMix64 { + fn next_u64(&mut self) -> u64 { + self.0 = self.0.wrapping_add(0x9e37_79b9_7f4a_7c15); + let mut z = self.0; + z = (z ^ (z >> 30)).wrapping_mul(0xbf58_476d_1ce4_e5b9); + z = (z ^ (z >> 27)).wrapping_mul(0x94d0_49bb_1331_11eb); + z ^ (z >> 31) + } + + /// Uniform in [0, 1). + fn next_f64(&mut self) -> f64 { + (self.next_u64() >> 11) as f64 / (1_u64 << 53) as f64 + } + + fn chance(&mut self, probability: f64) -> bool { + probability > 0.0 && self.next_f64() < probability + } + + fn duration_in(&mut self, (min, max): (Duration, Duration)) -> Duration { + if max <= min { + return min; + } + min + (max - min).mul_f64(self.next_f64()) + } +} + +#[cfg(test)] +mod tests { + use super::*; + use crate::document::replica::Epoch; + use crate::transport::{HeadsSummary, Keepalive}; + + fn ping() -> ParsedMessage { + ParsedMessage::Ping(Keepalive { + heads: HeadsSummary::of(Vec::new()), + epoch: Epoch::GENESIS, + epoch_seq: 0, + ready: true, + syncs_with_you: false, + }) + } + + /// Collects what a node receives. + fn record(transport: Arc<SimTransport>) -> Arc<Mutex<Vec<String>>> { + let received = Arc::new(Mutex::new(Vec::new())); + let sink = Arc::clone(&received); + tokio::spawn(async move { + transport + .start_consumer(Box::new(move |from, _message, _len| { + sink.lock().unwrap().push(from); + Box::pin(async { Ok(()) }) + })) + .await + }); + received + } + + #[tokio::test] + async fn delivers_decodes_and_loses_when_down() { + let net = SimNet::new(1); + let a = net.transport("a"); + let received = record(net.transport("b")); + + let settle = || tokio::time::sleep(Duration::from_millis(30)); + a.send_message("b", &ping()).await.unwrap(); + settle().await; + net.take_down("b"); + a.send_message("b", &ping()).await.unwrap(); + settle().await; + net.bring_up("b"); + a.send_message("b", &ping()).await.unwrap(); + settle().await; + + assert_eq!(*received.lock().unwrap(), vec!["a", "a"]); + let counts = net.counts(); + assert_eq!((counts.sent, counts.delivered, counts.lost), (3, 2, 1)); + assert_eq!(a.resolve_seed("b").await.unwrap(), "b"); + assert!(a.resolve_seed("c").await.is_err()); + } + + #[tokio::test] + async fn loss_and_duplication_follow_the_seed() { + let run = |seed| async move { + let net = SimNet::new(seed); + net.set_default_link(LinkParams { + loss: 0.3, + duplicate: 0.2, + ..LinkParams::default() + }); + let a = net.transport("a"); + let _b = net.transport("b"); + for _ in 0..200 { + a.send_message("b", &ping()).await.unwrap(); + } + tokio::time::sleep(Duration::from_millis(50)).await; + net.counts() + }; + let first = run(7).await; + assert_eq!(first, run(7).await); + assert!(first.lost > 30 && first.lost < 90, "{first:?}"); + assert!(first.duplicated > 10, "{first:?}"); + } + + #[tokio::test] + async fn a_link_overrides_the_default_in_one_direction() { + let net = SimNet::new(1); + net.set_link( + "a", + "b", + LinkParams { + loss: 1.0, + ..LinkParams::default() + }, + ); + let a = net.transport("a"); + let b = net.transport("b"); + let at_a = record(Arc::clone(&a)); + let at_b = record(Arc::clone(&b)); + + a.send_message("b", &ping()).await.unwrap(); + b.send_message("a", &ping()).await.unwrap(); + tokio::time::sleep(Duration::from_millis(30)).await; + + assert!(at_b.lock().unwrap().is_empty()); + assert_eq!(*at_a.lock().unwrap(), vec!["b"]); + } + + #[tokio::test] + async fn inject_returns_once_handled() { + let net = SimNet::new(1); + let received = record(net.transport("b")); + net.inject("ghost", "b", &ping()).await.unwrap(); + assert_eq!(*received.lock().unwrap(), vec!["ghost"]); + assert!(net.inject("ghost", "nobody", &ping()).await.is_err()); + } +} |