add files module to consistently track stats

3 files changed, 34 insertions, 91 deletions

diff --git a/src/swarm/bootstrap.rs b/src/swarm/bootstrap.rs
index 3be462e..f41a00b 100644
--- a/src/swarm/bootstrap.rs
+++ b/src/swarm/bootstrap.rs
@@ -26,6 +26,7 @@ use uuid::Uuid;
 
 use crate::document::replica::Epoch;
 use crate::document::store;
+use crate::files::{FileKind, Files};
 use crate::remake::snapshot::{self, META_ROOT, SWARM};
 use crate::state::AppState;
 use crate::swarm::peer::{remember_peer, send_hello_once};
@@ -148,14 +149,14 @@ pub(crate) struct Backup {
     pub(crate) doc: AutoCommit,
     pub(crate) epoch: Epoch,
     pub(crate) swarm_id: String,
-    /// The file's size (`swarm::stats`, disk).
-    pub(crate) bytes: u64,
 }
 
 /// Loads a backup to recover from, as `GET /admin/document` saves it.
-pub(crate) fn load_backup(path: &Path) -> Result<Backup> {
-    let bytes =
-        std::fs::read(path).with_context(|| format!("Reading the backup {}", path.display()))?;
+pub(crate) fn load_backup(path: &Path, files: &Files) -> Result<Backup> {
+    let bytes = files
+        .read(FileKind::Backup, path)
+        .with_context(|| format!("Reading the backup {}", path.display()))?
+        .with_context(|| format!("The backup {} doesn't exist", path.display()))?;
     let mut doc = AutoCommit::load(&bytes)
         .with_context(|| format!("Loading the backup {}", path.display()))?
         .with_actor(ActorId::random());
@@ -171,7 +172,6 @@ pub(crate) fn load_backup(path: &Path) -> Result<Backup> {
         doc,
         epoch,
         swarm_id,
-        bytes: bytes.len() as u64,
     })
 }
 
diff --git a/src/swarm/state_file.rs b/src/swarm/state_file.rs
index 79caec2..62a6dbe 100644
--- a/src/swarm/state_file.rs
+++ b/src/swarm/state_file.rs
@@ -4,18 +4,17 @@
 //!
 //! 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.
+//! changed since, atomically (`files`).
 
-use std::io::Write;
-use std::path::{Path, PathBuf};
-use std::sync::{Mutex, MutexGuard};
+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;
 
@@ -38,8 +37,7 @@ pub(crate) struct Remembered {
 
 pub(crate) struct StateFile {
     path: PathBuf,
-    /// Bytes read when opened, if there was a file (`swarm::stats`, disk).
-    read_bytes: Option<u64>,
+    files: Arc<Files>,
     // A std mutex: never held across an await.
     inner: Mutex<Inner>,
 }
@@ -54,30 +52,27 @@ struct Inner {
 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 mut read_bytes = None;
-        let remembered = match std::fs::read(&path) {
-            Ok(bytes) => {
-                read_bytes = Some(bytes.len() as u64);
-                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");
+    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")
                         }
-                        None
+                        Err(error) => warn!(error = %error, "Couldn't move the state file aside"),
                     }
+                    None
                 }
-            }
-            Err(error) if error.kind() == std::io::ErrorKind::NotFound => None,
+            },
+            Ok(None) => None,
             Err(error) => {
                 error!(path = %path.display(), error = %error, "Couldn't read the state file; remembering nothing");
                 None
@@ -85,7 +80,7 @@ impl StateFile {
         };
         let file = Self {
             path,
-            read_bytes,
+            files,
             inner: Mutex::new(Inner {
                 written: remembered.clone(),
                 ..Inner::default()
@@ -94,43 +89,19 @@ impl StateFile {
         (file, remembered)
     }
 
-    pub(crate) fn read_bytes(&self) -> Option<u64> {
-        self.read_bytes
-    }
-
     fn lock(&self) -> MutexGuard<'_, Inner> {
         self.inner
             .lock()
             .unwrap_or_else(|poisoned| poisoned.into_inner())
     }
 
-    /// Returns the bytes written.
-    fn write(&self, remembered: &Remembered) -> std::io::Result<u64> {
+    fn write(&self, remembered: &Remembered) -> std::io::Result<()> {
         let mut bytes = serde_json::to_vec_pretty(remembered)?;
         bytes.push(b'\n');
-        let temporary = with_suffix(&self.path, ".tmp");
-        {
-            let mut file = std::fs::File::create(&temporary)?;
-            file.write_all(&bytes)?;
-            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(bytes.len() as u64)
+        self.files.write_atomic(FileKind::State, &self.path, &bytes)
     }
 }
 
-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.
@@ -178,11 +149,10 @@ pub(crate) async fn update(state: &AppState, now: bool) {
         }
     }
 
-    let writer = std::sync::Arc::clone(&file);
+    let writer = Arc::clone(&file);
     let content = remembered.clone();
     match tokio::task::spawn_blocking(move || writer.write(&content)).await {
-        Ok(Ok(bytes)) => {
-            state.stats.disk_written(bytes);
+        Ok(Ok(())) => {
             info!(
                 path = %file.path.display(),
                 seeds = remembered.seeds.len(),
diff --git a/src/swarm/stats.rs b/src/swarm/stats.rs
index ab5c4da..eb34276 100644
--- a/src/swarm/stats.rs
+++ b/src/swarm/stats.rs
@@ -33,17 +33,6 @@ struct Inner {
     /// The document version (`AppState::change_tx`) and epoch the saved size is of.
     measured_version: Option<(u64, Epoch)>,
     propagation: Propagation,
-    disk: Disk,
-}
-
-/// Files this node reads and writes for itself: the state file, a backup to recover
-/// from. Not the Mycelium daemon's, and not logs.
-#[derive(Debug, Clone, Copy, Default)]
-pub(crate) struct Disk {
-    pub(crate) reads: u64,
-    pub(crate) read_bytes: u64,
-    pub(crate) writes: u64,
-    pub(crate) written_bytes: u64,
 }
 
 /// Upper bounds (inclusive, milliseconds) of the propagation delay buckets; a last
@@ -118,7 +107,6 @@ pub(crate) struct Snapshot {
     pub(crate) max_heads: usize,
     pub(crate) stale_hello_replies: u64,
     pub(crate) propagation: Propagation,
-    pub(crate) disk: Disk,
 }
 
 impl Default for NodeStats {
@@ -163,20 +151,6 @@ impl NodeStats {
         }
     }
 
-    /// A file of `bytes` was read (`Disk`).
-    pub(crate) fn disk_read(&self, bytes: u64) {
-        let mut inner = self.lock();
-        inner.disk.reads += 1;
-        inner.disk.read_bytes += bytes;
-    }
-
-    /// A file of `bytes` was written (`Disk`).
-    pub(crate) fn disk_written(&self, bytes: u64) {
-        let mut inner = self.lock();
-        inner.disk.writes += 1;
-        inner.disk.written_bytes += bytes;
-    }
-
     pub(crate) fn stale_hello_reply(&self) {
         self.lock().stale_hello_replies += 1;
     }
@@ -231,7 +205,6 @@ impl NodeStats {
             max_heads: inner.max_heads,
             stale_hello_replies: inner.stale_hello_replies,
             propagation: inner.propagation,
-            disk: inner.disk,
         }
     }
 }