summaryrefslogtreecommitdiffstats
path: root/src/swarm/state_file.rs
diff options
context:
space:
mode:
Diffstat (limited to 'src/swarm/state_file.rs')
-rw-r--r--src/swarm/state_file.rs202
1 files changed, 202 insertions, 0 deletions
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
+}