zc2 0.0.32

//! What each worker is like, right now.
//!
//! Every distribution here is chosen for one property: **its sufficient
//! statistics are additive**. A cell's summary is then the exact sum of its
//! members', so a mesh of millions of workers rolls up losslessly and no
//! broker ever has to hold — or ship — per-worker state for anyone else's
//! workers. Anything needing the original samples (raw histograms, medians)
//! is excluded by that rule.
//!
//! - service time → log-normal, statistics `(n, Σ ln x, Σ ln²x)`
//! - success, trust → Beta, statistics `(α, β)`
//!
//! Observations decay by a discount factor, so the estimate follows recent
//! behaviour rather than averaging over a worker's whole history.

/// Additive sufficient statistics for one worker (or one cell).
#[derive(Debug, Clone, Copy, PartialEq)]
pub struct Stats {
    /// Observations, discounted — fractional, hence f64.
    n: f64,
    sum_log: f64,
    sum_log_sq: f64,
    successes: f64,
    failures: f64,
    verified: f64,
    unverified: f64,
}

impl Default for Stats {
    fn default() -> Self {
        Self::new()
    }
}

impl Stats {
    pub fn new() -> Self {
        Stats {
            n: 0.0,
            sum_log: 0.0,
            sum_log_sq: 0.0,
            successes: 0.0,
            failures: 0.0,
            verified: 0.0,
            unverified: 0.0,
        }
    }

    /// Record one completed request. `duration_ms` must be positive; a
    /// non-positive or non-finite duration is not a measurement and is
    /// ignored rather than poisoning the log-sums with NaN.
    pub fn observe(&mut self, duration_ms: f64, success: bool) {
        if success {
            self.successes += 1.0;
        } else {
            self.failures += 1.0;
        }
        if duration_ms.is_finite() && duration_ms > 0.0 {
            let l = duration_ms.ln();
            self.n += 1.0;
            self.sum_log += l;
            self.sum_log_sq += l * l;
        }
    }

    /// Record whether a returned result could be verified.
    pub fn observe_trust(&mut self, verified: bool) {
        match verified {
            true => self.verified += 1.0,
            false => self.unverified += 1.0,
        }
    }

    /// Fold another summary in. This is what makes cell rollup exact: the
    /// merge of two summaries equals the summary of the two sample sets.
    pub fn merge(&mut self, other: &Stats) {
        self.n += other.n;
        self.sum_log += other.sum_log;
        self.sum_log_sq += other.sum_log_sq;
        self.successes += other.successes;
        self.failures += other.failures;
        self.verified += other.verified;
        self.unverified += other.unverified;
    }

    /// Age every observation by `factor` (0 < factor ≤ 1). One knob for how
    /// fast the estimate forgets: a worker that was slow an hour ago and is
    /// fast now should read as fast.
    pub fn decay(&mut self, factor: f64) {
        let f = factor.clamp(0.0, 1.0);
        self.n *= f;
        self.sum_log *= f;
        self.sum_log_sq *= f;
        self.successes *= f;
        self.failures *= f;
        self.verified *= f;
        self.unverified *= f;
    }

    /// How many observations back this, after discounting.
    pub fn weight(&self) -> f64 {
        self.n
    }

    /// `None` until something has been measured — unknown is not zero, and
    /// treating it as zero is what made an unmeasured worker rank as the
    /// fastest one on the mesh.
    pub fn median_service_ms(&self) -> Option<f64> {
        (self.n > 0.0).then(|| (self.sum_log / self.n).exp())
    }

    /// The spread of log service time; `None` until two observations exist,
    /// since one point has no spread to speak of.
    pub fn log_sd(&self) -> Option<f64> {
        if self.n < 2.0 {
            return None;
        }
        let mean = self.sum_log / self.n;
        let var = (self.sum_log_sq / self.n - mean * mean).max(0.0);
        Some(var.sqrt())
    }

    /// Expected service time: the log-normal mean, `exp(μ + σ²/2)`, which sits
    /// above the median because the distribution has a long right tail.
    pub fn expected_service_ms(&self) -> Option<f64> {
        let median = self.median_service_ms()?;
        let sd = self.log_sd().unwrap_or(0.0);
        Some(median * (sd * sd / 2.0).exp())
    }

    /// Posterior mean of the success rate, with a Beta(1,1) prior so a worker
    /// with no history reads as 0.5 rather than as certain either way.
    pub fn success_rate(&self) -> f64 {
        (self.successes + 1.0) / (self.successes + self.failures + 2.0)
    }

    /// Posterior mean of the share of results that could be verified.
    pub fn trust(&self) -> f64 {
        (self.verified + 1.0) / (self.verified + self.unverified + 2.0)
    }

    /// Equal up to floating-point rounding. Merging is exact in real
    /// arithmetic, but summing in a different order rounds differently, and
    /// over a large fleet those roundings accumulate — which is why the
    /// statistics are kept as sums rather than as running means.
    #[cfg(test)]
    pub(crate) fn close_to(&self, other: &Stats) -> bool {
        let near = |a: f64, b: f64| (a - b).abs() <= 1e-9 * a.abs().max(b.abs()).max(1.0);
        near(self.n, other.n)
            && near(self.sum_log, other.sum_log)
            && near(self.sum_log_sq, other.sum_log_sq)
            && near(self.successes, other.successes)
            && near(self.failures, other.failures)
            && near(self.verified, other.verified)
            && near(self.unverified, other.unverified)
    }
}

#[cfg(test)]
mod tests {
    use super::*;

    fn stats_of(samples: &[(f64, bool)]) -> Stats {
        let mut s = Stats::new();
        for (ms, ok) in samples {
            s.observe(*ms, *ok);
        }
        s
    }

    /// The property the whole rollup rests on: a cell's summary must equal the
    /// summary of its members' samples. Without it, "millions of workers" needs
    /// per-worker state shipped across the mesh, which is the thing being
    /// avoided.
    #[test]
    fn a_merged_summary_equals_the_summary_of_the_merged_samples() {
        let left = [(10.0, true), (20.0, true), (40.0, false)];
        let right = [(5.0, true), (80.0, true)];

        let mut merged = stats_of(&left);
        merged.merge(&stats_of(&right));

        let together: Vec<(f64, bool)> = left.iter().chain(right.iter()).copied().collect();
        let direct = stats_of(&together);

        // Exact in real arithmetic; in floating point, addition is not
        // associative, so the two differ by rounding (observed: 2 ULP). What
        // must hold is that no information is lost, not that the bits match.
        assert!(merged.close_to(&direct), "{merged:?} vs {direct:?}");
        assert_eq!(merged.weight(), direct.weight());
        assert_eq!(
            merged.median_service_ms().map(|m| (m * 1e9).round()),
            direct.median_service_ms().map(|m| (m * 1e9).round())
        );
    }

    /// Hand-checkable: the log-normal's median is the geometric mean, and the
    /// geometric mean of 1, 2, 4 and 8 is 64^(1/4) = 2.828…
    #[test]
    fn the_median_is_the_geometric_mean_of_what_was_seen() {
        let s = stats_of(&[(1.0, true), (2.0, true), (4.0, true), (8.0, true)]);
        let median = s.median_service_ms().expect("four observations");
        assert!(
            (median - 8f64.sqrt()).abs() < 1e-9,
            "got {median}, want {}",
            8f64.sqrt()
        );
        // The mean sits above the median: the distribution has a long tail.
        assert!(s.expected_service_ms().unwrap() > median);
    }

    /// Unknown is not zero. Reading an unmeasured worker as 0 ms is what made
    /// one that had never answered outrank a worker measured at 5 ms.
    #[test]
    fn nothing_measured_reads_as_unknown_not_as_instant() {
        let empty = Stats::new();
        assert_eq!(empty.median_service_ms(), None);
        assert_eq!(empty.expected_service_ms(), None);
        assert_eq!(empty.log_sd(), None);
        assert_eq!(empty.weight(), 0.0);
        // A coin-flip prior, not a verdict.
        assert!((empty.success_rate() - 0.5).abs() < 1e-12);
        assert!((empty.trust() - 0.5).abs() < 1e-12);
    }

    /// A worker that was slow and is now fast must read as fast, or the
    /// estimate describes a machine that no longer exists.
    #[test]
    fn recent_behaviour_outweighs_old_behaviour() {
        let mut s = Stats::new();
        for _ in 0..50 {
            s.observe(1000.0, true);
        }
        let before = s.median_service_ms().unwrap();

        // The node got fast; each new observation ages the old ones.
        for _ in 0..50 {
            s.decay(0.8);
            s.observe(10.0, true);
        }
        let after = s.median_service_ms().unwrap();

        assert!(before > 900.0, "started slow: {before}");
        assert!(after < 20.0, "followed the change: {after} (from {before})");
    }

    /// Decay must not quietly erase the record, or a worker would look
    /// unmeasured again and be re-explored from nothing on every tick.
    #[test]
    fn decay_keeps_the_shape_of_what_was_learned() {
        let mut s = stats_of(&[(10.0, true), (10.0, true), (10.0, false)]);
        let median = s.median_service_ms().unwrap();
        let rate = s.success_rate();
        s.decay(0.5);
        assert!((s.median_service_ms().unwrap() - median).abs() < 1e-9);
        assert!(s.weight() > 0.0, "still measured");
        // The rate moves toward the prior as evidence ages, but not past it.
        assert!(s.success_rate() > 0.5 && s.success_rate() < rate + 1e-9);
    }

    /// A failure with no timing (a timeout) still counts against the success
    /// rate, and must not poison the service-time statistics.
    #[test]
    fn a_timeout_counts_against_success_without_corrupting_the_timing() {
        let mut s = stats_of(&[(10.0, true)]);
        s.observe(f64::NAN, false);
        s.observe(-1.0, false);
        assert_eq!(s.weight(), 1.0, "only the real measurement timed");
        assert!((s.median_service_ms().unwrap() - 10.0).abs() < 1e-9);
        assert!(s.success_rate() < 0.5, "two failures against one success");
    }

    /// Small enough that a cell of 10k workers is well under a megabyte.
    #[test]
    fn a_worker_summary_stays_small() {
        assert!(
            std::mem::size_of::<Stats>() <= 64,
            "{} bytes",
            std::mem::size_of::<Stats>()
        );
    }
}

#[cfg(test)]
mod scale_tests {
    use super::*;

    /// The target is millions of workers. A cell's summary must cost the same
    /// to read whether it covers ten workers or a million, and the per-worker
    /// state must stay small enough to hold a whole cell in memory.
    #[test]
    fn a_million_workers_roll_up_into_one_summary_of_constant_size() {
        let per_worker = std::mem::size_of::<Stats>();
        assert!(per_worker <= 64, "{per_worker} bytes per worker");

        // One cell of 10k workers, each with a handful of observations.
        let mut cell = Stats::new();
        for w in 0..10_000u32 {
            let mut worker = Stats::new();
            let ms = 5.0 + (w % 50) as f64;
            worker.observe(ms, w % 97 != 0);
            worker.observe(ms * 1.1, true);
            cell.merge(&worker);
        }
        assert_eq!(cell.weight(), 20_000.0);
        assert_eq!(
            std::mem::size_of_val(&cell),
            per_worker,
            "still one summary"
        );

        // A hundred such cells is the million, and merging them is a hundred
        // additions — not a million.
        let mut mesh = Stats::new();
        for _ in 0..100 {
            mesh.merge(&cell);
        }
        assert_eq!(mesh.weight(), 2_000_000.0);
        assert_eq!(std::mem::size_of_val(&mesh), per_worker);
        let median = mesh.median_service_ms().expect("measured");
        assert!(
            (5.0..=60.0).contains(&median),
            "the mesh-wide median is still meaningful: {median}"
        );
    }
}