path: root/src/swarm/stats.rs

//! 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);
    }
}