//! 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, } pub(crate) struct StateFile { path: PathBuf, // A std mutex: never held across an await. inner: Mutex, } #[derive(Default)] struct Inner { written: Option, written_at: Option, checked_at: Option, } 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) { let remembered = match std::fs::read(&path) { Ok(bytes) => match serde_json::from_slice::(&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 { let now = crate::state::unix_millis(); let (mut stable, transient): (Vec, Vec) = 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 }