add swarm stats

9 files changed, 238 insertions, 1 deletions

diff --git a/src/main.rs b/src/main.rs
index 8b6f823..622f755 100644
--- a/src/main.rs
+++ b/src/main.rs
@@ -191,6 +191,7 @@ async fn main() -> Result<()> {
         )),
         firewall,
         traffic,
+        stats: Default::default(),
         sync_events: sync_tx,
         change_tx,
         initial_sync_done: Arc::new(AtomicBool::new(false)),
diff --git a/src/state.rs b/src/state.rs
index c3b23de..aae33f5 100644
--- a/src/state.rs
+++ b/src/state.rs
@@ -519,6 +519,8 @@ pub(crate) struct AppState {
     pub(crate) firewall: Arc<Firewall>,
     /// Messages and bytes exchanged, per peer (`transport::traffic`).
     pub(crate) traffic: Arc<transport::traffic::Traffic>,
+    /// Node-wide figures for /admin/stats (`swarm::stats`).
+    pub(crate) stats: Arc<crate::swarm::stats::NodeStats>,
 
     pub(crate) sync_events: mpsc::UnboundedSender<SyncEvent>,
 
diff --git a/src/swarm/mod.rs b/src/swarm/mod.rs
index ee0de20..83e2d2d 100644
--- a/src/swarm/mod.rs
+++ b/src/swarm/mod.rs
@@ -10,5 +10,6 @@ pub(crate) mod relay;
 pub(crate) mod rtt;
 pub(crate) mod source;
 pub(crate) mod stability;
+pub(crate) mod stats;
 pub(crate) mod sync;
 pub(crate) mod topology;
diff --git a/src/swarm/peer.rs b/src/swarm/peer.rs
index 3ceea3c..7a3eed8 100644
--- a/src/swarm/peer.rs
+++ b/src/swarm/peer.rs
@@ -492,6 +492,7 @@ pub(crate) async fn run_peer_worker(state: AppState) {
             apply_forget_records(&state).await;
             crate::swarm::members::ensure_registered(&state).await;
             crate::swarm::stability::tick(&state).await;
+            crate::swarm::stats::measure_saved_size(&state);
             {
                 let known = state.peer_states.read().await;
                 state.traffic.retain(|peer| known.contains_key(peer));
@@ -725,6 +726,7 @@ pub(crate) async fn handle_message(
     // we were down) says nothing about now; the peer answers our current HELLO.
     if let transport::ParsedMessage::HelloReply(reply) = &message {
         if reply.run_id != state.run_id {
+            state.stats.stale_hello_reply();
             debug!(peer = %peer, "Ignoring a HELLO_REPLY to a HELLO from an earlier run");
             return Ok(());
         }
diff --git a/src/swarm/source.rs b/src/swarm/source.rs
index 36a333e..4025f0e 100644
--- a/src/swarm/source.rs
+++ b/src/swarm/source.rs
@@ -111,6 +111,7 @@ pub(crate) fn defer(state: &AppState, peer: &str, peer_state: &PeerState) -> Opt
         ),
         None => debug!(peer = %peer, rtt_ms, "Catching up from this peer first"),
     }
+    state.stats.catch_up_source_chosen();
     *source = Some(CatchUpSource {
         peer: peer.to_owned(),
         last_heard: now,
diff --git a/src/swarm/stability.rs b/src/swarm/stability.rs
index 31ba4b8..9d79b17 100644
--- a/src/swarm/stability.rs
+++ b/src/swarm/stability.rs
@@ -135,6 +135,13 @@ fn lock(state: &AppState) -> std::sync::MutexGuard<'_, Stability> {
         .unwrap_or_else(|poisoned| poisoned.into_inner())
 }
 
+/// The fraction of the observed time (up to a day) we had a live peer, and how much
+/// time was observed.
+pub(crate) fn measured(state: &AppState) -> (f64, Duration) {
+    let stability = lock(state);
+    (stability.fraction(), stability.observed())
+}
+
 /// A HELLO_REPLY told us how long ago that peer last heard from us before this run.
 pub(crate) fn note_away(state: &AppState, away: Option<Duration>) {
     if let Some(away) = away {
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);
+    });
+}
diff --git a/src/swarm/sync.rs b/src/swarm/sync.rs
index 383cdaf..adf8146 100644
--- a/src/swarm/sync.rs
+++ b/src/swarm/sync.rs
@@ -305,6 +305,7 @@ pub(crate) async fn handle_sync(
     let applied = {
         let doc = Arc::clone(&state.doc);
         let peer_state_mutex = Arc::clone(&peer_state_mutex);
+        let stats = Arc::clone(&state.stats);
         let check_caught_up = !state.initial_sync_done.load(Ordering::Acquire);
         let epoch = envelope.epoch;
         let message = envelope.message;
@@ -334,7 +335,9 @@ pub(crate) async fn handle_sync(
             peer_state.sync.reply_settled();
 
             resend_if_needed(&peer, &mut peer_state.sync.state);
-            let changed = replica.doc.get_heads() != heads_before;
+            let heads_after = replica.doc.get_heads();
+            stats.heads_seen(heads_after.len());
+            let changed = heads_after != heads_before;
 
             Ok(Some((
                 // Only from a ready peer: a peer still catching up has empty heads, which
@@ -367,6 +370,9 @@ pub(crate) async fn handle_sync(
 
     if caught_up && !state.initial_sync_done.swap(true, Ordering::AcqRel) {
         info!(peer = %peer, "Initial sync complete");
+        state
+            .stats
+            .caught_up(state.traffic.total().received.sync.bytes);
         source::finished(state, peer);
     }
 
diff --git a/src/swarm/topology.rs b/src/swarm/topology.rs
index 8e25a2a..c5a02f1 100644
--- a/src/swarm/topology.rs
+++ b/src/swarm/topology.rs
@@ -48,6 +48,44 @@ pub(crate) fn is_full_mesh(state: &AppState) -> bool {
         .is_none_or(|view| view.full_mesh)
 }
 
+/// The current view in numbers, once known.
+#[derive(Debug, Clone, Copy)]
+pub(crate) struct Summary {
+    /// Registered members, us included.
+    pub(crate) members: usize,
+    /// Members other than us not currently judged unreachable.
+    pub(crate) reachable: usize,
+    pub(crate) stable: usize,
+    pub(crate) neighbours: usize,
+    pub(crate) stable_neighbours: usize,
+    pub(crate) full_mesh: bool,
+}
+
+pub(crate) fn summary(state: &AppState) -> Option<Summary> {
+    let local = state.local_id();
+    state
+        .neighbours
+        .read()
+        .unwrap_or_else(|poisoned| poisoned.into_inner())
+        .as_ref()
+        .map(|view| Summary {
+            members: view.members.len(),
+            reachable: view
+                .members
+                .iter()
+                .filter(|m| m.as_str() != local && !view.unreachable.contains(*m))
+                .count(),
+            stable: view.stable.len(),
+            neighbours: view.neighbours.len(),
+            stable_neighbours: view
+                .neighbours
+                .iter()
+                .filter(|m| view.stable.contains(*m))
+                .count(),
+            full_mesh: view.full_mesh,
+        })
+}
+
 /// Whether we initiate contact with `peer`: HELLOs, keepalives, dormancy checks, sync
 /// openings. Everyone is a neighbour until the view is known (while joining, or before
 /// any member registered), so joining works as in a full mesh. We always *answer*