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::locks::MutexExt;
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,
    relay: Relay,
    blob_requests_dropped: u64,
    largest_sync_received: 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 {
    pub(crate) 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;
    }

    /// The bound of the bucket by which `quantile` of the delays came: None if beyond
    /// the last bound, or none came.
    pub(crate) fn within(&self, quantile: f64) -> Option<u64> {
        let wanted = (self.count as f64 * quantile).ceil() as u64;
        let mut seen = 0;
        for (bound, count) in self.buckets() {
            seen += count;
            if seen >= wanted && seen > 0 {
                return bound;
            }
        }
        None
    }

    /// 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,
}

/// Passing changes on (`swarm::relay`), since start.
#[derive(Debug, Clone, Copy, Default)]
pub(crate) struct Relay {
    /// Announcement rounds (each a STATUS to the neighbours).
    pub(crate) announce_rounds: u64,
    /// Of those, quick rounds to all neighbours, as some change came with no writer's
    /// tree to go along (pulled, a delta, from a node with no part in it), and slow rounds
    /// after changes that did (`swarm::relay`).
    pub(crate) treeless_rounds: u64,
    pub(crate) slow_rounds: u64,
    /// Pulls started.
    pub(crate) pulls: u64,
    /// Announcements of heads we lacked that had to wait for the pull in progress.
    pub(crate) pulls_queued: u64,
    /// Changes passed on along writers' trees (`swarm::tree`): a sync message asked for
    /// per child.
    pub(crate) tree_forwards: u64,
    /// Sync messages sent taken as delivered (`SyncSession::assume_delivered`), and their
    /// payload's bytes.
    pub(crate) delivered: u64,
    pub(crate) delivered_bytes: u64,
    /// Sync messages sent (not offers), the bytes the envelope added to their
    /// Automerge messages, and of those the epoch's.
    pub(crate) framed: u64,
    pub(crate) framing_bytes: u64,
    pub(crate) framing_id_bytes: u64,
    /// Sync messages received that carried changes, all of which we had already, and
    /// their payload's bytes: changes that came twice. Of those, how many came along a
    /// writer's tree (`SyncEnvelope::writer`), and how many were taken as delivered by
    /// their sender (`SyncEnvelope::delivered`).
    pub(crate) nothing_new: u64,
    pub(crate) nothing_new_bytes: u64,
    pub(crate) nothing_new_along_trees: u64,
    pub(crate) nothing_new_delivered: u64,
    /// Sync messages received that carried changes, the bytes of those changes, and of
    /// their bloom filters; and the same of those with nothing new.
    pub(crate) carrying: u64,
    pub(crate) carrying_change_bytes: u64,
    pub(crate) carrying_bloom_bytes: u64,
    pub(crate) nothing_new_change_bytes: u64,
    pub(crate) nothing_new_bloom_bytes: u64,
    /// The changes new to us that sync messages brought (beside what they carried that
    /// we had): with `carrying` less `nothing_new`, how many a message with news brings.
    pub(crate) new_changes: u64,
    /// Changes peers asked for that we had counted as theirs, as another parent in
    /// their writer's tree brings them (`SyncSession::known_to_peer`): within 2 s of
    /// counting them so (so they weren't what we passed on built on), and later.
    pub(crate) asked_known_lately: u64,
    pub(crate) asked_known_later: u64,
    /// Changes from elsewhere that came not along a writer's tree (pulled, with an
    /// answer, sent again, fetched as a delta), and their delays summed (as `Propagation`).
    pub(crate) off_tree: u64,
    pub(crate) off_tree_ms: u64,
}

/// 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) relay: Relay,
    pub(crate) blob_requests_dropped: u64,
    /// The largest sync message received (its Automerge part, in bytes).
    pub(crate) largest_sync_received: 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_or_recover()
    }

    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;
    }

    /// An announcement round: `treeless` (quick, to all), `slow`, or else quick to
    /// those writers' trees leave out.
    pub(crate) fn announce_round(&self, treeless: bool, slow: bool) {
        let mut inner = self.lock();
        inner.relay.announce_rounds += 1;
        if slow {
            inner.relay.slow_rounds += 1;
        } else if treeless {
            inner.relay.treeless_rounds += 1;
        }
    }

    pub(crate) fn pull_started(&self) {
        self.lock().relay.pulls += 1;
    }

    pub(crate) fn pull_queued(&self) {
        self.lock().relay.pulls_queued += 1;
    }

    pub(crate) fn tree_forward(&self) {
        self.lock().relay.tree_forwards += 1;
    }

    /// A sync message went out whose envelope added `bytes` to its Automerge message,
    /// `ids` of them the epoch.
    pub(crate) fn sync_framing(&self, bytes: usize, ids: usize) {
        let mut inner = self.lock();
        inner.relay.framed += 1;
        inner.relay.framing_bytes += bytes as u64;
        inner.relay.framing_id_bytes += ids as u64;
    }

    /// A sync message taken as delivered went out, of `bytes` payload.
    pub(crate) fn delivered_sent(&self, bytes: usize) {
        let mut inner = self.lock();
        inner.relay.delivered += 1;
        inner.relay.delivered_bytes += bytes as u64;
    }

    /// A sync message of `bytes` payload carried changes, all of which we had already;
    /// it came `along_tree`, or was `delivered`; `sizes`: of its changes and blooms.
    pub(crate) fn nothing_new(
        &self,
        bytes: usize,
        along_tree: bool,
        delivered: bool,
        sizes: (usize, usize),
    ) {
        let mut inner = self.lock();
        inner.relay.nothing_new += 1;
        inner.relay.nothing_new_bytes += bytes as u64;
        inner.relay.nothing_new_along_trees += u64::from(along_tree);
        inner.relay.nothing_new_delivered += u64::from(delivered);
        inner.relay.nothing_new_change_bytes += sizes.0 as u64;
        inner.relay.nothing_new_bloom_bytes += sizes.1 as u64;
    }

    /// Changes from elsewhere were applied, `delays_ms` after they were committed, that
    /// came not along a writer's tree.
    pub(crate) fn off_tree(&self, delays_ms: &[i64]) {
        let mut inner = self.lock();
        inner.relay.off_tree += delays_ms.len() as u64;
        inner.relay.off_tree_ms += delays_ms.iter().map(|ms| ms.max(&0)).sum::<i64>() as u64;
    }

    /// A sync message carrying changes arrived; `sizes`: of its changes and blooms.
    pub(crate) fn carrying(&self, sizes: (usize, usize)) {
        let mut inner = self.lock();
        inner.relay.carrying += 1;
        inner.relay.carrying_change_bytes += sizes.0 as u64;
        inner.relay.carrying_bloom_bytes += sizes.1 as u64;
    }

    /// A peer asked for changes we had counted as its (`SyncSession::resend_if_needed`).
    pub(crate) fn asked_known_elsewhere(&self, (lately, later): (usize, usize)) {
        if lately + later > 0 {
            let mut inner = self.lock();
            inner.relay.asked_known_lately += lately as u64;
            inner.relay.asked_known_later += later as u64;
        }
    }

    /// A sync message brought `count` changes new to us.
    pub(crate) fn new_changes(&self, count: usize) {
        self.lock().relay.new_changes += count as u64;
    }

    /// A sync message of `bytes` arrived.
    pub(crate) fn sync_received(&self, bytes: usize) {
        let mut inner = self.lock();
        inner.largest_sync_received = inner.largest_sync_received.max(bytes as u64);
    }

    /// A blob request went unanswered: too many were being answered.
    pub(crate) fn blob_request_dropped(&self) {
        self.lock().blob_requests_dropped += 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,
            relay: inner.relay,
            blob_requests_dropped: inner.blob_requests_dropped,
            largest_sync_received: inner.largest_sync_received,
            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 copy = doc.blocking_read().doc.clone();
            crate::document::save(&copy).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);
    }
}