1 files changed, 284 insertions, 0 deletions
diff --git a/src/swarm/stats.rs b/src/swarm/stats.rs
new file mode 100644
index 0000000..eb34276
--- /dev/null
+++ b/src/swarm/stats.rs
@@ -0,0 +1,284 @@
+//! 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::document::replica::Epoch;
+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, and only after a change.
+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 version (`AppState::change_tx`) and epoch the saved size is of.
+    measured_version: Option<(u64, Epoch)>,
+    propagation: Propagation,
+}
+
+/// Upper bounds (inclusive, milliseconds) of the propagation delay buckets; a last
+/// bucket takes the rest.
+const PROPAGATION_BUCKETS_MS: [u64; 11] = [
+    100, 250, 500, 1_000, 2_500, 5_000, 10_000, 30_000, 60_000, 300_000, 3_600_000,
+];
+
+/// How long changes written elsewhere took to reach us: from the writer's commit (its
+/// clock) to applying them here (ours), so clock skew between the two is part of it.
+#[derive(Debug, Clone, Copy, Default)]
+pub(crate) struct Propagation {
+    /// Per bucket of `PROPAGATION_BUCKETS_MS`, then the rest; not cumulative.
+    pub(crate) buckets: [u64; PROPAGATION_BUCKETS_MS.len() + 1],
+    pub(crate) count: u64,
+    pub(crate) sum_ms: u64,
+    /// Changes stamped later than our clock said on arrival (counted in the first
+    /// bucket): the writer's clock is ahead of ours.
+    pub(crate) ahead: u64,
+}
+
+impl Propagation {
+    fn record(&mut self, delay_ms: i64) {
+        if delay_ms < 0 {
+            self.ahead += 1;
+        }
+        let delay_ms = delay_ms.max(0) as u64;
+        let bucket = PROPAGATION_BUCKETS_MS
+            .iter()
+            .position(|&bound| delay_ms <= bound)
+            .unwrap_or(PROPAGATION_BUCKETS_MS.len());
+        self.buckets[bucket] += 1;
+        self.count += 1;
+        self.sum_ms += delay_ms;
+    }
+
+    /// Each bucket's upper bound (None for the last) and count.
+    pub(crate) fn buckets(&self) -> impl Iterator<Item = (Option<u64>, u64)> + '_ {
+        PROPAGATION_BUCKETS_MS
+            .iter()
+            .copied()
+            .map(Some)
+            .chain([None])
+            .zip(self.buckets.iter().copied())
+    }
+}
+
+/// 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,
+    pub(crate) propagation: Propagation,
+}
+
+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);
+    }
+
+    /// Changes from elsewhere were applied, `delays_ms` after they were committed.
+    pub(crate) fn propagated(&self, delays_ms: impl IntoIterator<Item = i64>) {
+        let mut inner = self.lock();
+        for delay_ms in delays_ms {
+            inner.propagation.record(delay_ms);
+        }
+    }
+
+    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 of the document at `version` is due; if so, it counts as
+    /// started.
+    fn start_measuring(&self, now: Instant, version: (u64, Epoch)) -> bool {
+        let mut inner = self.lock();
+        let due = !inner.measuring
+            && inner.measured_version != Some(version)
+            && 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
+    }
+
+    /// The document at `version` was saved for another reason (an export): a
+    /// measurement for free.
+    pub(crate) fn saved_size_measured(&self, bytes: u64, took: Duration, version: (u64, Epoch)) {
+        let mut inner = self.lock();
+        inner.saved_size = Some(SavedSize {
+            bytes,
+            at: Instant::now(),
+            took,
+        });
+        inner.measured_version = Some(version);
+    }
+
+    fn finish_measuring(&self, measured: Option<(SavedSize, (u64, Epoch))>) {
+        let mut inner = self.lock();
+        inner.measuring = false;
+        if let Some((measured, version)) = measured {
+            inner.saved_size = Some(measured);
+            inner.measured_version = Some(version);
+        }
+    }
+
+    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,
+            propagation: inner.propagation,
+        }
+    }
+}
+
+/// The document's version for measuring: bumped by every write and applied sync
+/// message (`AppState::notify_store_changed`), together with the epoch (a switch
+/// replaces the document).
+pub(crate) fn document_version(state: &AppState) -> (u64, Epoch) {
+    (*state.change_tx.borrow(), state.local_epoch())
+}
+
+/// Measures the document's saved size in the background when due: the first time, then
+/// only after the document changed, 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
+/// at most every 200 s). An idle swarm measures once. 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) {
+    // Taken before the copy: a change during it makes another measurement due.
+    let version = document_version(state);
+    if !state.stats.start_measuring(Instant::now(), version) {
+        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,
+                    },
+                    version,
+                ))
+            }
+            Err(error) => {
+                warn!(error = %error, "Measuring the saved document size failed");
+                None
+            }
+        };
+        state.stats.finish_measuring(measured);
+    });
+}
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+
+    #[test]
+    fn propagation_buckets() {
+        let mut propagation = Propagation::default();
+        for delay_ms in [-20, 100, 101, 7_200_000] {
+            propagation.record(delay_ms);
+        }
+        let buckets: Vec<_> = propagation.buckets().filter(|(_, n)| *n > 0).collect();
+        assert_eq!(buckets, [(Some(100), 2), (Some(250), 1), (None, 1)]);
+        assert_eq!(propagation.ahead, 1);
+        assert_eq!(propagation.count, 4);
+        assert_eq!(propagation.sum_ms, 7_200_201);
+    }
+}