pixelpass/src/host/audit/metrics.rs

//! O5 measurement, pure (impl plan §5.2).
//!
//! v3.4 §6.4 asserts "a full recompute per graph event is fine for v1". The
//! impl plan closes O5 by refusing to let that rest on a node count: what has
//! to be recorded is the **graph-event rate**, the **recompute duration
//! distribution and maximum**, and **whether events queue behind recompute or
//! logging**.
//!
//! Everything here is arithmetic over samples the caller supplies. The clock
//! reads live at the I/O edge ([`super::sink`]), which is what keeps the
//! statistics unit-testable: a test feeds a hand-written sample sequence and
//! asserts the summary exactly, with no timing flake.
//!
//! **The queueing measure is a proxy, and a one-directional one.** libpipewire
//! dispatches registry callbacks serially on its own loop thread and exposes no
//! queue depth, so nothing here can read a backlog directly. What it can see is
//! that the observer thread was *continuously busy*: if an event begins being
//! handled within [`QUEUE_THRESHOLD_US`] of the previous sample's completion,
//! it was almost certainly already waiting while that recompute ran. That makes
//! [`Summary::queued_events`] a **lower bound** — a genuine backlog always shows
//! up in it, but a burst that happens to arrive exactly as the loop goes idle is
//! counted as un-queued. Combined with [`Summary::busy_fraction`] (which needs
//! no inference at all) it is enough to answer O5 in the direction that matters:
//! a low busy fraction with zero queued events is headroom, and anything else is
//! a number to argue about rather than an assumption to inherit.

use serde::Serialize;

use crate::host::observer::EventKind;

/// An event beginning this close behind the previous sample's completion is
/// counted as having queued. Deliberately tight: the cost of being wrong in the
/// generous direction is a metric that overstates backlog and sends a later
/// round chasing a non-problem.
pub const QUEUE_THRESHOLD_US: u64 = 100;

/// Upper bounds of the duration histogram, microseconds. A twelfth (overflow)
/// bucket catches everything at or above the last bound. Log-ish spacing: the
/// interesting question is which order of magnitude a recompute lands in, not
/// its exact microsecond.
pub const BUCKET_BOUNDS_US: [u64; 11] = [
    50, 100, 250, 500, 1_000, 2_500, 5_000, 10_000, 25_000, 50_000, 100_000,
];

/// Human labels for the histogram buckets, parallel to [`BUCKET_BOUNDS_US`]
/// plus the overflow bucket.
pub const BUCKET_LABELS: [&str; 12] = [
    "<50us", "<100us", "<250us", "<500us", "<1ms", "<2.5ms", "<5ms", "<10ms", "<25ms", "<50ms",
    "<100ms", ">=100ms",
];

/// A bucketed duration distribution with exact count, sum and maximum.
///
/// Bounded memory by construction — the audit runs for as long as a share does,
/// and keeping every sample to compute an exact percentile would grow without
/// limit. The maximum, which is the number O5 actually cares about, is kept
/// exactly; percentiles are reported as the bucket they fall in.
#[derive(Clone, Debug, Default, PartialEq, Eq)]
pub struct Histogram {
    buckets: [u64; 12],
    count: u64,
    sum_us: u64,
    max_us: u64,
}

impl Histogram {
    pub fn record(&mut self, us: u64) {
        let index = BUCKET_BOUNDS_US
            .iter()
            .position(|&bound| us < bound)
            .unwrap_or(BUCKET_BOUNDS_US.len());
        self.buckets[index] += 1;
        self.count += 1;
        self.sum_us = self.sum_us.saturating_add(us);
        self.max_us = self.max_us.max(us);
    }

    pub fn count(&self) -> u64 {
        self.count
    }

    pub fn max_us(&self) -> u64 {
        self.max_us
    }

    pub fn sum_us(&self) -> u64 {
        self.sum_us
    }

    pub fn mean_us(&self) -> Option<u64> {
        (self.count > 0).then(|| self.sum_us / self.count)
    }

    /// The label of the bucket the `q`-quantile falls in (`q` in `0.0..=1.0`),
    /// or `None` when nothing has been recorded.
    ///
    /// Uses the *nearest-rank* definition: the bucket containing the
    /// `ceil(q · count)`-th sample in ascending order. Reported as a bucket
    /// rather than a number because interpolating inside a bucket would invent
    /// precision the histogram does not have.
    pub fn quantile_bucket(&self, q: f64) -> Option<&'static str> {
        if self.count == 0 {
            return None;
        }
        let q = q.clamp(0.0, 1.0);
        // Rank is 1-based; q = 0 still names the bucket holding the smallest
        // sample rather than degenerating to "no samples".
        let rank = ((q * self.count as f64).ceil() as u64).max(1);
        let mut cumulative = 0u64;
        for (index, &n) in self.buckets.iter().enumerate() {
            cumulative += n;
            if cumulative >= rank {
                return Some(BUCKET_LABELS[index]);
            }
        }
        // Unreachable while `count` is the sum of the buckets, but returning the
        // top bucket is the fail-loud answer rather than a panic in a metric.
        Some(BUCKET_LABELS[BUCKET_LABELS.len() - 1])
    }

    /// Non-empty buckets as `(label, count)`, ascending. Empty buckets are
    /// dropped so a summary line stays readable.
    pub fn distribution(&self) -> Vec<(&'static str, u64)> {
        self.buckets
            .iter()
            .enumerate()
            .filter(|&(_, &n)| n > 0)
            .map(|(index, &n)| (BUCKET_LABELS[index], n))
            .collect()
    }
}

/// One handled event, as timed by the I/O edge.
///
/// Ticks are the AEC validator's clock, not graph changes, so [`Metrics`] counts
/// them separately — folding them into the event rate would inflate it by a
/// constant 4 Hz and hide the real graph churn.
#[derive(Clone, Copy, Debug, PartialEq, Eq)]
pub struct Sample {
    /// Monotonic microseconds (since observer start) at which handling began.
    pub at_us: u64,
    /// Microseconds between the previous sample's completion and `at_us`. Zero
    /// for the first sample.
    pub gap_us: u64,
    /// Time spent in the AEC observe + taint recompute.
    pub recompute_us: u64,
    /// Time spent serialising and writing the record, zero when nothing was
    /// emitted. Separate from `recompute_us` because O5 asks about queueing
    /// behind recompute **or logging** — and if logging turns out to dominate,
    /// that is a fixable problem of a different kind.
    pub emit_us: u64,
    pub kind: EventKind,
}

/// Rolling O5 state. Fold samples in with [`Metrics::record`]; read with
/// [`Metrics::summary`].
#[derive(Clone, Debug, Default)]
pub struct Metrics {
    graph_events: u64,
    tick_events: u64,
    emitted_records: u64,
    recompute: Histogram,
    emit: Histogram,
    busy_us: u64,
    queued_events: u64,
    first_event_us: Option<u64>,
    last_completion_us: u64,
}

impl Metrics {
    pub fn record(&mut self, sample: Sample) {
        match sample.kind {
            EventKind::Graph => self.graph_events += 1,
            EventKind::Tick => self.tick_events += 1,
        }
        self.recompute.record(sample.recompute_us);
        if sample.emit_us > 0 {
            self.emitted_records += 1;
            self.emit.record(sample.emit_us);
        }
        self.busy_us = self
            .busy_us
            .saturating_add(sample.recompute_us)
            .saturating_add(sample.emit_us);

        // The first sample has no predecessor to have queued behind.
        if self.first_event_us.is_some() && sample.gap_us <= QUEUE_THRESHOLD_US {
            self.queued_events += 1;
        }
        self.first_event_us.get_or_insert(sample.at_us);
        self.last_completion_us = sample
            .at_us
            .saturating_add(sample.recompute_us)
            .saturating_add(sample.emit_us);
    }

    pub fn summary(&self) -> Summary {
        let span_us = self
            .first_event_us
            .map(|first| self.last_completion_us.saturating_sub(first))
            .unwrap_or(0);
        // A rate needs a span to divide by; one event in zero elapsed time has
        // no rate, and reporting a made-up one is worse than reporting none.
        let graph_events_per_sec = (span_us > 0)
            .then(|| self.graph_events as f64 * 1_000_000.0 / span_us as f64)
            .map(round_2);
        let busy_fraction = (span_us > 0).then(|| round_4(self.busy_us as f64 / span_us as f64));

        Summary {
            graph_events: self.graph_events,
            tick_events: self.tick_events,
            emitted_records: self.emitted_records,
            span_us,
            graph_events_per_sec,
            recompute_max_us: self.recompute.max_us(),
            recompute_mean_us: self.recompute.mean_us(),
            recompute_p50: self.recompute.quantile_bucket(0.50),
            recompute_p90: self.recompute.quantile_bucket(0.90),
            recompute_p99: self.recompute.quantile_bucket(0.99),
            recompute_distribution: self.recompute.distribution(),
            emit_max_us: self.emit.max_us(),
            emit_mean_us: self.emit.mean_us(),
            emit_distribution: self.emit.distribution(),
            busy_us: self.busy_us,
            busy_fraction,
            queued_events: self.queued_events,
            queue_threshold_us: QUEUE_THRESHOLD_US,
        }
    }
}

/// The O5 answer, as emitted.
#[derive(Clone, Debug, PartialEq, Serialize)]
pub struct Summary {
    pub graph_events: u64,
    pub tick_events: u64,
    pub emitted_records: u64,
    /// First event to last completion, microseconds.
    pub span_us: u64,
    pub graph_events_per_sec: Option<f64>,
    pub recompute_max_us: u64,
    pub recompute_mean_us: Option<u64>,
    pub recompute_p50: Option<&'static str>,
    pub recompute_p90: Option<&'static str>,
    pub recompute_p99: Option<&'static str>,
    pub recompute_distribution: Vec<(&'static str, u64)>,
    pub emit_max_us: u64,
    pub emit_mean_us: Option<u64>,
    pub emit_distribution: Vec<(&'static str, u64)>,
    /// Total observer-thread time spent recomputing and logging.
    pub busy_us: u64,
    /// `busy_us / span_us` — the share of wall time the observer thread could
    /// not be servicing PipeWire. Needs no inference, unlike `queued_events`.
    pub busy_fraction: Option<f64>,
    /// Events that began within `queue_threshold_us` of the previous sample's
    /// completion — a **lower bound** on backlog, see the module header.
    pub queued_events: u64,
    pub queue_threshold_us: u64,
}

/// Keep the JSON readable: a rate to two decimals and a fraction to four are
/// well past the precision any of this is good to.
fn round_2(value: f64) -> f64 {
    (value * 100.0).round() / 100.0
}

fn round_4(value: f64) -> f64 {
    (value * 10_000.0).round() / 10_000.0
}