add swarm stats

1 files changed, 179 insertions, 0 deletions

diff --git a/src/swarm/stats.rs b/src/swarm/stats.rs
new file mode 100644
index 0000000..79e5c83
--- /dev/null
+++ b/src/swarm/stats.rs
@@ -0,0 +1,179 @@
+//! Node-wide figures for /admin/stats that no other part keeps. Recorded where things
+//! happen anyway; the one exception is the document's saved size (see
+//! `measure_saved_size`).
+
+use std::sync::{Mutex, MutexGuard};
+use std::time::Duration;
+
+use tokio::time::Instant;
+use tracing::{debug, warn};
+
+use crate::state::AppState;
+
+/// Saving takes time proportional to the document; we spend at most about 1% of it on
+/// measuring, and measure at least this long apart.
+const SAVED_SIZE_MIN_INTERVAL: Duration = Duration::from_secs(60);
+const SAVED_SIZE_DUTY: u32 = 100;
+
+#[derive(Debug)]
+pub(crate) struct NodeStats {
+    started: Instant,
+    inner: Mutex<Inner>,
+}
+
+#[derive(Debug, Default)]
+struct Inner {
+    caught_up: Option<CaughtUp>,
+    catch_up_sources: u32,
+    max_heads: usize,
+    stale_hello_replies: u64,
+    saved_size: Option<SavedSize>,
+    measuring: bool,
+}
+
+/// The document's size as saved (compressed), when it was measured and how long that
+/// took (copy and save).
+#[derive(Debug, Clone, Copy)]
+pub(crate) struct SavedSize {
+    pub(crate) bytes: u64,
+    pub(crate) at: Instant,
+    pub(crate) took: Duration,
+}
+
+/// When the initial sync finished, and what it took.
+#[derive(Debug, Clone, Copy)]
+pub(crate) struct CaughtUp {
+    /// Since the process started (handshakes included).
+    pub(crate) after: Duration,
+    /// Sync payload received until then, from all peers.
+    pub(crate) sync_bytes: u64,
+}
+
+/// A copy of the figures.
+#[derive(Debug, Clone, Copy)]
+pub(crate) struct Snapshot {
+    pub(crate) uptime: Duration,
+    pub(crate) caught_up: Option<CaughtUp>,
+    pub(crate) catch_up_sources: u32,
+    pub(crate) max_heads: usize,
+    pub(crate) stale_hello_replies: u64,
+}
+
+impl Default for NodeStats {
+    fn default() -> Self {
+        Self {
+            started: Instant::now(),
+            inner: Mutex::default(),
+        }
+    }
+}
+
+impl NodeStats {
+    fn lock(&self) -> MutexGuard<'_, Inner> {
+        self.inner
+            .lock()
+            .unwrap_or_else(|poisoned| poisoned.into_inner())
+    }
+
+    pub(crate) fn caught_up(&self, sync_bytes: u64) {
+        let after = self.started.elapsed();
+        self.lock()
+            .caught_up
+            .get_or_insert(CaughtUp { after, sync_bytes });
+    }
+
+    /// A catch-up source was chosen (the first, or a replacement).
+    pub(crate) fn catch_up_source_chosen(&self) {
+        self.lock().catch_up_sources += 1;
+    }
+
+    /// Our document has `count` heads (after applying a sync message).
+    pub(crate) fn heads_seen(&self, count: usize) {
+        let mut inner = self.lock();
+        inner.max_heads = inner.max_heads.max(count);
+    }
+
+    pub(crate) fn stale_hello_reply(&self) {
+        self.lock().stale_hello_replies += 1;
+    }
+
+    /// The last measurement of the saved size, if any.
+    pub(crate) fn saved_size(&self) -> Option<SavedSize> {
+        self.lock().saved_size
+    }
+
+    /// Whether a measurement is due; if so, it counts as started.
+    fn start_measuring(&self, now: Instant) -> bool {
+        let mut inner = self.lock();
+        let due = !inner.measuring
+            && inner.saved_size.is_none_or(|last| {
+                now >= last.at + SAVED_SIZE_MIN_INTERVAL.max(last.took * SAVED_SIZE_DUTY)
+            });
+        if due {
+            inner.measuring = true;
+        }
+        due
+    }
+
+    fn finish_measuring(&self, measured: Option<SavedSize>) {
+        let mut inner = self.lock();
+        inner.measuring = false;
+        if let Some(measured) = measured {
+            inner.saved_size = Some(measured);
+        }
+    }
+
+    pub(crate) fn snapshot(&self) -> Snapshot {
+        let inner = self.lock();
+        Snapshot {
+            uptime: self.started.elapsed(),
+            caught_up: inner.caught_up,
+            catch_up_sources: inner.catch_up_sources,
+            max_heads: inner.max_heads,
+            stale_hello_replies: inner.stale_hello_replies,
+        }
+    }
+}
+
+/// Measures the document's saved size in the background when due: the first time, then
+/// at least a minute apart and spending no more than about 1% of the time on it (a
+/// document that takes 2 s to copy and save is measured every 200 s). Called by the
+/// peer worker every few seconds.
+///
+/// The copy holds the read lock (writes wait for it); the save runs without any lock.
+pub(crate) fn measure_saved_size(state: &AppState) {
+    if !state.stats.start_measuring(Instant::now()) {
+        return;
+    }
+    let state = state.clone();
+    tokio::spawn(async move {
+        let start = Instant::now();
+        let doc = std::sync::Arc::clone(&state.doc);
+        // Both steps take time proportional to the document: off the async threads.
+        let measured = match tokio::task::spawn_blocking(move || {
+            let mut copy = doc.blocking_read().doc.clone();
+            copy.save().len() as u64
+        })
+        .await
+        {
+            Ok(bytes) => {
+                let took = start.elapsed();
+                debug!(
+                    bytes,
+                    took_ms = took.as_millis() as u64,
+                    "Measured the saved document size"
+                );
+                Some(SavedSize {
+                    bytes,
+                    at: Instant::now(),
+                    took,
+                })
+            }
+            Err(error) => {
+                warn!(error = %error, "Measuring the saved document size failed");
+                None
+            }
+        };
+        state.stats.finish_measuring(measured);
+    });
+}