add state fileHEADmaster

This allows storing the swarm ID and additionally discovered seeds,
which prevents a formerly genesis node to start a new document.

7 files changed, 266 insertions, 4 deletions

diff --git a/src/config/mod.rs b/src/config/mod.rs
index 73f4432..b4ea0aa 100644
--- a/src/config/mod.rs
+++ b/src/config/mod.rs
@@ -107,7 +107,14 @@ pub(crate) const OPTIONS: &[OptionSpec] = &[
         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.",
+        "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",
@@ -222,6 +229,7 @@ pub(crate) struct Settings {
     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")]
diff --git a/src/main.rs b/src/main.rs
index 09b365b..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();
@@ -205,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);
@@ -223,6 +253,7 @@ async fn main() -> Result<()> {
     });
 
     let server_fut = server_builder.into_future();
+    let stopping = state.clone();
     tokio::pin!(server_fut);
 
     tokio::select! {
@@ -248,6 +279,7 @@ async fn main() -> Result<()> {
     }
 
     info!("HTTP API server stopped");
+    swarm::state_file::update(&stopping, true).await;
     Ok(())
 }
 
diff --git a/src/state.rs b/src/state.rs
index 785eee8..37bde4a 100644
--- a/src/state.rs
+++ b/src/state.rs
@@ -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 1ae5173..84ff156 100644
--- a/src/swarm/bootstrap.rs
+++ b/src/swarm/bootstrap.rs
@@ -31,8 +31,22 @@ const RESOLVE_RETRY_MAX: Duration = Duration::from_secs(300);
 
 pub(crate) async fn bootstrap(state: &AppState, seeds: &[String], genesis: bool) -> Result<()> {
     if genesis {
-        let swarm_id = Uuid::new_v4().to_string();
-        info!(swarm_id = %swarm_id, "Starting a new swarm as its genesis node (--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))
@@ -40,6 +54,8 @@ pub(crate) async fn bootstrap(state: &AppState, seeds: &[String], genesis: bool)
             .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()).await;
         return Ok(());
     }
 
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
+}