diff options
Diffstat (limited to 'src')
| -rw-r--r-- | src/api/mod.rs | 7 | ||||
| -rw-r--r-- | src/config/mod.rs | 45 | ||||
| -rw-r--r-- | src/main.rs | 37 | ||||
| -rw-r--r-- | src/remake/api.rs | 46 | ||||
| -rw-r--r-- | src/remake/coordinator.rs | 30 | ||||
| -rw-r--r-- | src/remake/kind.rs | 12 | ||||
| -rw-r--r-- | src/remake/records.rs | 88 | ||||
| -rw-r--r-- | src/remake/snapshot.rs | 2 | ||||
| -rw-r--r-- | src/remake/vote.rs | 2 | ||||
| -rw-r--r-- | src/state.rs | 6 | ||||
| -rw-r--r-- | src/swarm/bootstrap.rs | 184 | ||||
| -rw-r--r-- | src/swarm/mod.rs | 1 | ||||
| -rw-r--r-- | src/swarm/peer.rs | 1 | ||||
| -rw-r--r-- | src/swarm/state_file.rs | 202 |
14 files changed, 540 insertions, 123 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/config/mod.rs b/src/config/mod.rs index ef28542..b4ea0aa 100644 --- a/src/config/mod.rs +++ b/src/config/mod.rs @@ -86,7 +86,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 +100,28 @@ 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. 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( + "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 +227,9 @@ 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) swarm_key_file: Option<PathBuf>, pub(crate) alias: Option<String>, #[serde(default = "default_keepalive_secs")] @@ -400,15 +417,19 @@ fn unset_entry(provenance: Provenance) -> Entry { /// values from earlier sources (lower precedence), so e.g. `--mycelium-rest-url` on the /// command line overrides `mycelium-key-file` from a file. Both set in one source is an /// error (`Settings::check`). -const CONFLICTS: &[(&[&str], &[&str])] = &[( - &["mycelium-rest-url"], - &[ - "mycelium-key-file", - "mycelium-network-key-file", - "mycelium-network-name", - "mycelium-tcp-listen-port", - ], -)]; +const CONFLICTS: &[(&[&str], &[&str])] = &[ + ( + &["mycelium-rest-url"], + &[ + "mycelium-key-file", + "mycelium-network-key-file", + "mycelium-network-name", + "mycelium-tcp-listen-port", + ], + ), + // A new swarm has nobody to join. + (&["genesis"], &["seeds"]), +]; /// The options `name` can't be combined with. fn conflicting(name: &str) -> impl Iterator<Item = &'static str> + '_ { diff --git a/src/main.rs b/src/main.rs index 217a9a4..4cfae66 100644 --- a/src/main.rs +++ b/src/main.rs @@ -86,7 +86,30 @@ async fn main() -> Result<()> { "Using Mycelium REST API" ); - let seeds: Vec<String> = args.seeds.clone(); + 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()); + (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(); @@ -141,6 +164,7 @@ async fn main() -> Result<()> { 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()), + state_file, config: Arc::new(config.report()), neighbours: Default::default(), pulling: Default::default(), @@ -155,6 +179,11 @@ async fn main() -> Result<()> { trace!("Application state initialized"); + // Join the remembered swarm and no other; HELLOs claim it. + if let Some(remembered) = remembered { + state.set_expected_swarm_id(remembered.swarm_id).await?; + } + let consumer_state = state.clone(); tokio::spawn(async move { let inner_state = consumer_state.clone(); @@ -186,8 +215,9 @@ async fn main() -> Result<()> { tokio::spawn(remake::run_worker(state.clone())); trace!("Remake worker task spawned"); + let joining = bootstrap::bootstrap(&state, &seeds, args.genesis); 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(()); @@ -204,6 +234,7 @@ async fn main() -> Result<()> { 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 +253,7 @@ async fn main() -> Result<()> { }); let server_fut = server_builder.into_future(); + let stopping = state.clone(); tokio::pin!(server_fut); tokio::select! { @@ -247,6 +279,7 @@ async fn main() -> Result<()> { } info!("HTTP API server stopped"); + swarm::state_file::update(&stopping, true).await; Ok(()) } diff --git a/src/remake/api.rs b/src/remake/api.rs index 826c81e..3530423 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}, @@ -395,6 +396,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) => { @@ -454,7 +464,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 @@ -598,6 +609,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); @@ -656,22 +668,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); diff --git a/src/remake/coordinator.rs b/src/remake/coordinator.rs index 9bcf8ba..3a7b071 100644 --- a/src/remake/coordinator.rs +++ b/src/remake/coordinator.rs @@ -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/kind.rs b/src/remake/kind.rs index 8131f7f..b36db4c 100644 --- a/src/remake/kind.rs +++ b/src/remake/kind.rs @@ -71,6 +71,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 need carrying over, + /// and a write made after it is lost if its author restarts before switching. + fn base_at_commit(&self) -> bool { + false + } } /// The kind with this name. @@ -97,6 +105,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/records.rs b/src/remake/records.rs index cd3e5be..0e40960 100644 --- a/src/remake/records.rs +++ b/src/remake/records.rs @@ -130,7 +130,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 +183,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 +204,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 +286,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 +536,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 +572,7 @@ mod tests { generation: 1, id: "cd".repeat(32), }, + base_heads: None, snapshot_bytes: 123, decided_at: 11_000, }; @@ -552,7 +590,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 +646,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 +681,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..ca50c86 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> { 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/state.rs b/src/state.rs index 9c5552f..37bde4a 100644 --- a/src/state.rs +++ b/src/state.rs @@ -583,8 +583,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 +602,8 @@ 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>>, + /// 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>, diff --git a/src/swarm/bootstrap.rs b/src/swarm/bootstrap.rs index 27ffdd8..84ff156 100644 --- a/src/swarm/bootstrap.rs +++ b/src/swarm/bootstrap.rs @@ -1,55 +1,76 @@ //! 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. use std::sync::atomic::Ordering; use std::time::Duration; -use anyhow::{Result, bail}; -use tracing::{debug, error, info, warn}; +use anyhow::{Context, Result}; +use serde_json::Value as JsonValue; +use tokio::time::Instant; +use tracing::{debug, info, warn}; use uuid::Uuid; +use crate::document::store; +use crate::remake::snapshot::{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)"); +pub(crate) async fn bootstrap(state: &AppState, seeds: &[String], genesis: bool) -> Result<()> { + if genesis { + // A remembered swarm starts again under its ID: nodes of it still running then + // merge with this one instead of ending up in another swarm. + let swarm_id = match state.swarm_id().await { + Some(remembered) => { + warn!( + swarm_id = %remembered, + "--genesis with a remembered swarm: starting it again, empty; nodes of it still running merge with this one" + ); + remembered + } + None => { + let swarm_id = Uuid::new_v4().to_string(); + info!(swarm_id = %swarm_id, "Starting a new swarm as its genesis node (--genesis)"); + swarm_id + } + }; + let recorded = JsonValue::String(swarm_id.clone()); + state + .mutate_doc(move |doc| store::write_path(doc, &swarm_id_path(), &recorded)) + .await + .context("Recording the swarm ID")?; state.initial_sync_done.store(true, Ordering::Release); - state.mark_ready(Uuid::new_v4().to_string()).await?; + state.mark_ready(swarm_id).await?; + // Remembered members, if any, to find those still running. + resolve_seeds(state, &mut seeds.to_vec()).await; return Ok(()); } - 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) - } - Err(e) => warn!(seed = %seed, error = %e, "Ignoring seed argument"), - } - } - - if peers.is_empty() { - bail!("None of the configured seeds could be resolved"); - } - - 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 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 +82,72 @@ 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).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, - )) +/// Resolves the seeds in `unresolved`, says HELLO to each that resolves, and keeps the +/// rest for later. +async fn resolve_seeds(state: &AppState, unresolved: &mut Vec<String>) { + let mut still = Vec::new(); + for seed in unresolved.drain(..) { + match state.transport.resolve_seed(&seed).await { + Ok(peer) => { + debug!(seed = %seed, peer = %peer, "Resolved seed"); + remember_peer(state, &peer).await; + if let Err(error) = send_hello_once(state, &peer).await { + warn!(peer = %peer, error = %format!("{error:#}"), "Initial HELLO failed; will retry"); + } + } + Err(error) => { + warn!(seed = %seed, error = %error, "Seed doesn't resolve; will retry"); + still.push(seed); } - 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" - ); - 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("; ") - ); } - tokio::time::sleep(BOOTSTRAP_RETRY).await; } + *unresolved = still; +} + +/// Why the node is still joining, for the log. +async fn report_joining(state: &AppState, seeds: &[String], unresolved: &[String]) { + if seeds.is_empty() { + warn!( + "Still no seeds: nothing to join. Add --seeds to join a swarm, or --genesis to start a new one" + ); + return; + } + let mut peer_diagnostics: Vec<String> = Vec::new(); + let peer_states = state.peer_states.read().await.clone(); + for (peer, peer_state_mutex) in &peer_states { + let peer_state = peer_state_mutex.lock().await; + peer_diagnostics.push(format!( + "{peer}: validated_swarm={:?}, epoch={:?}", + peer_state.standing.swarm_id(), + peer_state.epoch, + )); + } + warn!( + local_swarm = ?state.swarm_id().await, + local_epoch = %state.local_epoch(), + initial_sync_done = state.initial_sync_done.load(Ordering::Acquire), + unresolved_seeds = ?unresolved, + peers = %peer_diagnostics.join("; "), + "Still joining the swarm; trying on" + ); } 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..daf1af4 100644 --- a/src/swarm/peer.rs +++ b/src/swarm/peer.rs @@ -493,6 +493,7 @@ 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::state_file::update(&state, false).await; { let known = state.peer_states.read().await; state.traffic.retain(|peer| known.contains_key(peer)); diff --git a/src/swarm/state_file.rs b/src/swarm/state_file.rs new file mode 100644 index 0000000..0a9913d --- /dev/null +++ b/src/swarm/state_file.rs @@ -0,0 +1,202 @@ +//! 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: a temporary file next to it, flushed, then renamed over +//! it, so a crash leaves the old file or the new one. + +use std::io::Write; +use std::path::{Path, PathBuf}; +use std::sync::{Mutex, MutexGuard}; +use std::time::Duration; + +use serde::{Deserialize, Serialize}; +use tokio::time::Instant; +use tracing::{error, info, warn}; + +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, + // 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) -> (Self, Option<Remembered>) { + let remembered = match std::fs::read(&path) { + Ok(bytes) => match serde_json::from_slice::<Remembered>(&bytes) { + Ok(remembered) if !remembered.swarm_id.is_empty() => Some(remembered), + Ok(_) => None, + Err(error) => { + let aside = with_suffix(&path, ".bad"); + error!( + path = %path.display(), + moved_to = %aside.display(), + error = %error, + "Unreadable state file; moving it aside and remembering nothing" + ); + if let Err(error) = std::fs::rename(&path, &aside) { + warn!(error = %error, "Couldn't move the state file aside"); + } + None + } + }, + Err(error) if error.kind() == std::io::ErrorKind::NotFound => None, + Err(error) => { + error!(path = %path.display(), error = %error, "Couldn't read the state file; remembering nothing"); + None + } + }; + let file = Self { + path, + 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 bytes = serde_json::to_vec_pretty(remembered)?; + let temporary = with_suffix(&self.path, ".tmp"); + { + let mut file = std::fs::File::create(&temporary)?; + file.write_all(&bytes)?; + file.write_all(b"\n")?; + file.sync_all()?; + } + std::fs::rename(&temporary, &self.path)?; + // The rename itself, durably (best effort: not every platform syncs directories). + if let Some(dir) = self.path.parent().filter(|dir| !dir.as_os_str().is_empty()) { + if let Ok(dir) = std::fs::File::open(dir) { + let _ = dir.sync_all(); + } + } + Ok(()) + } +} + +fn with_suffix(path: &Path, suffix: &str) -> PathBuf { + let mut name = path.as_os_str().to_owned(); + name.push(suffix); + PathBuf::from(name) +} + +/// 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 = std::sync::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 +} |