diff options
Diffstat (limited to 'src/swarm/state_file.rs')
| -rw-r--r-- | src/swarm/state_file.rs | 189 |
1 files changed, 189 insertions, 0 deletions
diff --git a/src/swarm/state_file.rs b/src/swarm/state_file.rs new file mode 100644 index 0000000..8b883fc --- /dev/null +++ b/src/swarm/state_file.rs @@ -0,0 +1,189 @@ +//! 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, + state.clock.now_millis(), + ) + }; + 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, now: u64) -> Vec<String> { + 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 +} |