feat: persist job counter across restarts

daemon/Cargo.toml

@@ -14,6 +14,7 @@ path = "main.rs"
 doc = false
 
 [dependencies]
+atomicwrites = "0.4.1"
 clap = { version = "4.2.5", features = ["derive", "env"] }
 env_logger.workspace = true
 log.workspace = true
@@ -24,3 +25,4 @@ zinnia_runtime = { workspace = true }
 
 [dev-dependencies]
 pretty_assertions = { workspace = true }
+tempfile = "3.5.0"

daemon/main.rs

@@ -1,6 +1,8 @@
 mod args;
+mod state;
 mod station_reporter;
 
+use std::path::PathBuf;
 use std::rc::Rc;
 use std::time::Duration;
 
@@ -36,6 +38,9 @@ async fn run(config: CliArgs) -> Result<()> {
         ));
     }
 
+    let state_file = PathBuf::from(config.state_root).join("state.json");
+    log::debug!("Using state file: {}", state_file.display());
+
     log_info_activity("Module Runtime started.");
 
     let file = &config.files[0];
@@ -56,6 +61,7 @@ async fn run(config: CliArgs) -> Result<()> {
         ),
         wallet_address: config.wallet_address,
         reporter: Rc::new(StationReporter::new(
+            state_file,
             Duration::from_millis(200),
             module_name.into(),
         )),

daemon/state.rs

@@ -0,0 +1,99 @@
+use atomicwrites::{AtomicFile, OverwriteBehavior};
+use serde::{Deserialize, Serialize};
+use std::io::Write;
+use std::path::Path;
+
+#[derive(Serialize, Deserialize, Debug)]
+pub struct State {
+    pub total_jobs_completed: u64,
+}
+
+impl Default for State {
+    fn default() -> Self {
+        Self {
+            total_jobs_completed: 0,
+        }
+    }
+}
+
+pub fn load(state_file: &Path) -> State {
+    log::debug!("Loading initial state from {}", state_file.display());
+    match std::fs::read_to_string(state_file) {
+        Err(err) => {
+            if err.kind() != std::io::ErrorKind::NotFound {
+                log::warn!(
+                    "Cannot read initial state from {}: {}",
+                    state_file.display(),
+                    err
+                );
+            }
+            return State::default();
+        }
+        Ok(data) => match serde_json::from_str::<State>(&data) {
+            Err(err) => {
+                log::warn!(
+                    "Cannot parse initial state from {}: {}",
+                    state_file.display(),
+                    err
+                );
+                return State::default();
+            }
+            Ok(state) => {
+                log::debug!("Initial state: {:?}", state);
+                return state;
+            }
+        },
+    }
+}
+
+pub fn store(state_file: &Path, state: &State) {
+    if let Some(parent) = state_file.parent() {
+        if let Err(err) = std::fs::create_dir_all(&parent) {
+            log::warn!(
+                "Cannot create state directory {}: {}",
+                parent.display(),
+                err
+            );
+            return;
+        }
+    }
+
+    let payload = match serde_json::to_string_pretty(&state) {
+        Err(err) => {
+            log::warn!("Cannot serialize state: {}", err);
+            return;
+        }
+
+        Ok(payload) => payload,
+    };
+
+    let write_result = AtomicFile::new(state_file, OverwriteBehavior::AllowOverwrite)
+        .write(|f| f.write_all(payload.as_bytes()));
+    match write_result {
+        Err(err) => log::warn!("Cannot write state to {}: {}", state_file.display(), err),
+        Ok(()) => log::debug!("State stored in {}", state_file.display()),
+    }
+}
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+    use pretty_assertions::assert_eq;
+    use tempfile;
+    use zinnia_runtime::anyhow::Result;
+
+    #[test]
+    fn creates_missing_directories() -> Result<()> {
+        let state_dir = tempfile::tempdir()?;
+        let state_file = state_dir.path().join("subdir").join("state.json");
+        store(
+            &state_file,
+            &State {
+                total_jobs_completed: 1,
+            },
+        );
+        let loaded = load(&state_file);
+        assert_eq!(loaded.total_jobs_completed, 1);
+        Ok(())
+    }
+}