add files module to consistently track stats

8 files changed, 268 insertions, 143 deletions

diff --git a/src/api/stats.rs b/src/api/stats.rs
index 6c5e172..8899421 100644
--- a/src/api/stats.rs
+++ b/src/api/stats.rs
@@ -9,6 +9,8 @@ use axum::{
     http::{HeaderMap, header},
     response::{IntoResponse, Response},
 };
+use std::collections::BTreeMap;
+
 use serde::Serialize;
 
 use super::peers::EpochInfo;
@@ -35,11 +37,11 @@ pub(crate) struct StatsResponse {
     /// The peers listed in /peers together (a forgotten peer's are gone).
     incidents: Incidents,
     propagation: PropagationStats,
-    disk: DiskStats,
+    /// This node's own files (`files`), by kind.
+    disk: BTreeMap<&'static str, DiskStats>,
 }
 
-/// Files this node reads and writes for itself (the state file, a backup to recover
-/// from), since it started.
+/// Reads and writes of one kind of file, since the node started.
 #[derive(Debug, Serialize)]
 #[serde(rename_all = "camelCase")]
 struct DiskStats {
@@ -47,6 +49,7 @@ struct DiskStats {
     read_bytes: u64,
     writes: u64,
     written_bytes: u64,
+    errors: u64,
 }
 
 /// How long changes written elsewhere took from their commit there until applied here,
@@ -207,11 +210,20 @@ async fn collect(state: &AppState) -> StatsResponse {
                 .map(|(le_ms, count)| PropagationBucket { le_ms, count })
                 .collect(),
         },
-        disk: DiskStats {
-            reads: node.disk.reads,
-            read_bytes: node.disk.read_bytes,
-            writes: node.disk.writes,
-            written_bytes: node.disk.written_bytes,
-        },
+        disk: state
+            .files
+            .counts()
+            .into_iter()
+            .map(|(kind, counts)| {
+                let stats = DiskStats {
+                    reads: counts.reads,
+                    read_bytes: counts.read_bytes,
+                    writes: counts.writes,
+                    written_bytes: counts.written_bytes,
+                    errors: counts.errors,
+                };
+                (kind.as_str(), stats)
+            })
+            .collect(),
     }
 }
diff --git a/src/api/stats/prometheus.rs b/src/api/stats/prometheus.rs
index ea97bd2..e267042 100644
--- a/src/api/stats/prometheus.rs
+++ b/src/api/stats/prometheus.rs
@@ -297,31 +297,34 @@ pub(super) async fn render(state: &AppState, stats: &StatsResponse) -> String {
         stats.stale_hello_replies as f64,
     );
 
-    let disk = &stats.disk;
-    out.single(
-        "disk_reads_total",
-        "counter",
-        "Files this node read for itself (state file, backup).",
-        disk.reads as f64,
-    );
-    out.single(
-        "disk_read_bytes_total",
-        "counter",
-        "Bytes of those files.",
-        disk.read_bytes as f64,
-    );
-    out.single(
-        "disk_writes_total",
-        "counter",
-        "Files this node wrote for itself (state file).",
-        disk.writes as f64,
-    );
-    out.single(
-        "disk_written_bytes_total",
-        "counter",
-        "Bytes of those files.",
-        disk.written_bytes as f64,
-    );
+    type Field = fn(&super::DiskStats) -> u64;
+    let disk: [(&str, &str, Field); 5] = [
+        ("disk_reads_total", "Reads of this node's own files.", |c| {
+            c.reads
+        }),
+        ("disk_read_bytes_total", "Bytes read from them.", |c| {
+            c.read_bytes
+        }),
+        (
+            "disk_writes_total",
+            "Writes of this node's own files.",
+            |c| c.writes,
+        ),
+        ("disk_written_bytes_total", "Bytes written to them.", |c| {
+            c.written_bytes
+        }),
+        (
+            "disk_errors_total",
+            "Failed reads and writes of them.",
+            |c| c.errors,
+        ),
+    ];
+    for (name, help, field) in disk {
+        out.family(name, "counter", help);
+        for (kind, counts) in &stats.disk {
+            out.sample(name, &[("file", *kind)], field(counts) as f64);
+        }
+    }
 
     let propagation = &stats.propagation;
     out.family(
diff --git a/src/files.rs b/src/files.rs
new file mode 100644
index 0000000..81c8b1e
--- /dev/null
+++ b/src/files.rs
@@ -0,0 +1,174 @@
+//! The files this node reads and writes for itself, and how much it did (`/admin/stats`,
+//! `disk`): the state file (`swarm::state_file`) and a backup to recover from
+//! (`swarm::bootstrap`). Not the Mycelium daemon's files, configuration or logs.
+//!
+//! Writes replace a file atomically: a temporary file next to it, flushed to disk, then
+//! renamed over it, so a crash leaves the old content or the new one.
+
+use std::io::Write;
+use std::path::{Path, PathBuf};
+use std::sync::{Mutex, MutexGuard};
+
+/// What a file is, for the counts.
+#[derive(Clone, Copy, Debug, PartialEq, Eq)]
+pub(crate) enum FileKind {
+    State,
+    Backup,
+}
+
+impl FileKind {
+    pub(crate) const ALL: [FileKind; 2] = [FileKind::State, FileKind::Backup];
+
+    pub(crate) fn as_str(self) -> &'static str {
+        match self {
+            FileKind::State => "state",
+            FileKind::Backup => "backup",
+        }
+    }
+
+    fn index(self) -> usize {
+        self as usize
+    }
+}
+
+/// Counts for one kind of file, since the node started.
+#[derive(Clone, Copy, Debug, Default)]
+pub(crate) struct Counts {
+    pub(crate) reads: u64,
+    pub(crate) read_bytes: u64,
+    pub(crate) writes: u64,
+    pub(crate) written_bytes: u64,
+    /// Reads or writes that failed (a missing file read is no failure).
+    pub(crate) errors: u64,
+}
+
+/// Reads and writes this node's files, counting per kind.
+#[derive(Default)]
+pub(crate) struct Files {
+    // A std mutex: every access is short and never spans an await.
+    counts: Mutex<[Counts; FileKind::ALL.len()]>,
+}
+
+impl Files {
+    fn lock(&self) -> MutexGuard<'_, [Counts; FileKind::ALL.len()]> {
+        self.counts
+            .lock()
+            .unwrap_or_else(|poisoned| poisoned.into_inner())
+    }
+
+    fn count(&self, kind: FileKind, f: impl FnOnce(&mut Counts)) {
+        f(&mut self.lock()[kind.index()]);
+    }
+
+    /// The file's content; None if it doesn't exist.
+    pub(crate) fn read(&self, kind: FileKind, path: &Path) -> std::io::Result<Option<Vec<u8>>> {
+        match std::fs::read(path) {
+            Ok(bytes) => {
+                self.count(kind, |counts| {
+                    counts.reads += 1;
+                    counts.read_bytes += bytes.len() as u64;
+                });
+                Ok(Some(bytes))
+            }
+            Err(error) if error.kind() == std::io::ErrorKind::NotFound => Ok(None),
+            Err(error) => {
+                self.count(kind, |counts| counts.errors += 1);
+                Err(error)
+            }
+        }
+    }
+
+    /// Replaces the file's content atomically (see the module docs).
+    pub(crate) fn write_atomic(
+        &self,
+        kind: FileKind,
+        path: &Path,
+        bytes: &[u8],
+    ) -> std::io::Result<()> {
+        let result = write_atomic(path, bytes);
+        self.count(kind, |counts| match &result {
+            Ok(()) => {
+                counts.writes += 1;
+                counts.written_bytes += bytes.len() as u64;
+            }
+            Err(_) => counts.errors += 1,
+        });
+        result
+    }
+
+    /// Moves an unusable file aside (`<path><suffix>`), returning where to.
+    pub(crate) fn move_aside(&self, path: &Path, suffix: &str) -> std::io::Result<PathBuf> {
+        let aside = with_suffix(path, suffix);
+        std::fs::rename(path, &aside)?;
+        Ok(aside)
+    }
+
+    /// The counts per kind.
+    pub(crate) fn counts(&self) -> Vec<(FileKind, Counts)> {
+        let counts = *self.lock();
+        FileKind::ALL
+            .into_iter()
+            .map(|kind| (kind, counts[kind.index()]))
+            .collect()
+    }
+}
+
+fn write_atomic(path: &Path, bytes: &[u8]) -> std::io::Result<()> {
+    let temporary = with_suffix(path, ".tmp");
+    {
+        let mut file = std::fs::File::create(&temporary)?;
+        file.write_all(bytes)?;
+        file.sync_all()?;
+    }
+    std::fs::rename(&temporary, path)?;
+    sync_parent(path);
+    Ok(())
+}
+
+/// Makes a rename in `path`'s directory durable (best effort: not every platform syncs
+/// directories).
+fn sync_parent(path: &Path) {
+    if let Some(dir) = path.parent().filter(|dir| !dir.as_os_str().is_empty()) {
+        if let Ok(dir) = std::fs::File::open(dir) {
+            let _ = dir.sync_all();
+        }
+    }
+}
+
+fn with_suffix(path: &Path, suffix: &str) -> PathBuf {
+    let mut name = path.as_os_str().to_owned();
+    name.push(suffix);
+    PathBuf::from(name)
+}
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+
+    #[test]
+    fn writes_reads_and_counts() {
+        let dir = std::env::temp_dir().join(format!("continuum-files-{}", std::process::id()));
+        std::fs::create_dir_all(&dir).unwrap();
+        let path = dir.join("state.json");
+        let files = Files::default();
+
+        assert_eq!(files.read(FileKind::State, &path).unwrap(), None);
+        files.write_atomic(FileKind::State, &path, b"one").unwrap();
+        files
+            .write_atomic(FileKind::State, &path, b"three")
+            .unwrap();
+        assert_eq!(
+            files.read(FileKind::State, &path).unwrap().as_deref(),
+            Some(&b"three"[..])
+        );
+        assert!(!with_suffix(&path, ".tmp").exists());
+
+        let (kind, state) = files.counts()[0];
+        assert_eq!(kind, FileKind::State);
+        assert_eq!((state.writes, state.written_bytes), (2, 8));
+        assert_eq!((state.reads, state.read_bytes, state.errors), (1, 5, 0));
+        assert_eq!(files.counts()[1].1.reads, 0);
+
+        std::fs::remove_dir_all(&dir).unwrap();
+    }
+}
diff --git a/src/main.rs b/src/main.rs
index 57c26b5..76e5ff0 100644
--- a/src/main.rs
+++ b/src/main.rs
@@ -18,6 +18,7 @@ mod api;
 mod backoff;
 mod config;
 mod document;
+mod files;
 mod logging;
 mod remake;
 mod state;
@@ -86,10 +87,12 @@ async fn main() -> Result<()> {
         "Using Mycelium REST API"
     );
 
+    let files = Arc::new(files::Files::default());
     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());
+            let (file, remembered) =
+                swarm::state_file::StateFile::open(path.clone(), Arc::clone(&files));
             (Some(Arc::new(file)), remembered)
         }
         None => (None, None),
@@ -117,17 +120,12 @@ async fn main() -> Result<()> {
     let (change_events, _) = broadcast::channel(256);
 
     let recovered = match &args.recover_from {
-        Some(path) => Some(bootstrap::load_backup(path)?),
+        Some(path) => Some(bootstrap::load_backup(path, &files)?),
         None => None,
     };
-    let (initial_doc, recovered_epoch, recovered_swarm, backup_bytes) = match recovered {
-        Some(backup) => (
-            backup.doc,
-            Some(backup.epoch),
-            Some(backup.swarm_id),
-            Some(backup.bytes),
-        ),
-        None => (AutoCommit::new(), None, None, None),
+    let (initial_doc, recovered_epoch, recovered_swarm) = match recovered {
+        Some(backup) => (backup.doc, Some(backup.epoch), Some(backup.swarm_id)),
+        None => (AutoCommit::new(), None, None),
     };
     let initial_epoch = match (recovered_epoch, args.debug_epoch) {
         (Some(epoch), _) => epoch,
@@ -179,6 +177,7 @@ async fn main() -> Result<()> {
         blobs: Arc::new(swarm::blob::Blobs::new(args.blob_chunk_kib * 1024)),
         protocol_writers: Arc::new(args.protocol_writers.iter().cloned().collect()),
         state_file,
+        files,
         config: Arc::new(config.report()),
         neighbours: Default::default(),
         pulling: Default::default(),
@@ -193,14 +192,6 @@ async fn main() -> Result<()> {
 
     trace!("Application state initialized");
 
-    // Files read before the state existed.
-    let read = backup_bytes
-        .into_iter()
-        .chain(state.state_file.as_ref().and_then(|file| file.read_bytes()));
-    for bytes in read {
-        state.stats.disk_read(bytes);
-    }
-
     // Join the remembered swarm and no other; HELLOs claim it. A backup to recover from
     // names its swarm itself, which wins.
     let start = match recovered_swarm {
diff --git a/src/state.rs b/src/state.rs
index 37bde4a..15652ab 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>>,
+    /// This node's own files, and how much it read and wrote them.
+    pub(crate) files: Arc<crate::files::Files>,
     /// 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
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,
         }
     }
 }