navian_memcheck/

lib.rs

//! # navian-memcheck
//!
//! **Resource-leak / soak testing as a `cargo test` assertion.**
//!
//! The bug that took down a 52 GB process in production was not a leak in the
//! classic sense: every byte was *reachable*. A per-session map simply grew
//! without bound because nothing ever evicted it. `LeakSanitizer` and Valgrind
//! `memcheck` are blind to this — the memory is still referenced, so to them it
//! is "in use", not "lost". A heap profiler would show it, but only if a human
//! sat and eyeballed a flamegraph.
//!
//! `navian-memcheck` asserts the *property that was actually violated*: **after
//! a warmup period, live memory PLATEAUS.** You drive a workload under sustained
//! load; the crate samples live heap on a fixed cadence, fits a line through the
//! back half of the run, and fails if the slope is still climbing or a hard cap
//! is breached. It is a pass/fail check you drop into a test — no profiler, no
//! flamegraph, no platform.
//!
//! ```no_run
//! use navian_memcheck::{soak, SoakConfig};
//!
//! # fn process_one_event(_: u64) {}
//! let report = soak(&SoakConfig::iterations(200_000), |i| {
//!     process_one_event(i); // your real per-event work
//! });
//! report.assert(); // panics with a readable summary if memory kept growing
//! ```
//!
//! ## Activation: one dependency, one test, zero code changes
//!
//! Add the crate and write one test. Nothing in your production code changes.
//!
//! ```toml
//! [dev-dependencies]
//! navian-memcheck = "0.1"
//! ```
//!
//! The default [`RssSampler`] reads the OS, so there is no allocator to install
//! and no global state to set up.
//!
//! ## Two surfaces
//!
//! - **In-process** ([`soak`], [`assert_bounded`], [`assert_linear_in`]) — drive a
//!   workload closure and watch *this* process's memory. Runs in your test suite.
//! - **Out-of-process** — the `navian-memcheck` CLI soaks *any* command's RSS over
//!   a duration and gates CI on the same plateau property, with no code at all. See
//!   the `navian-memcheck-cli` crate.
//!
//! ## Sampling precision (optional)
//!
//! The default reads process RSS — coarser (page-granular, includes allocator
//! retention) but zero-setup. For a cleaner in-process signal, enable the
//! `jemalloc` feature and pass `JemallocSampler` to [`soak_with`]; it reads
//! jemalloc `stats.allocated` (live bytes, no page noise). That requires jemalloc
//! to be the global allocator — one line, and free for services already on it:
//!
//! ```ignore
//! #[global_allocator]
//! static ALLOC: tikv_jemallocator::Jemalloc = tikv_jemallocator::Jemalloc;
//! ```
//!
//! ## Determinism
//!
//! Memory boundedness is a property, and like any property it is only trustworthy
//! if the run that checks it is reproducible. Drive your workload from a seeded
//! RNG (or under [`navian-dst`](https://github.com/TheFuturePlutus/navian-dst)) so
//! that a soak which fails on seed *N* fails again, identically, on seed *N* — and
//! so you can bisect the growth to the event that caused it.

#![forbid(unsafe_code)]
#![warn(missing_docs)]

use std::fmt;

// ─────────────────────────────────────────────────────────────────────────────
// Sampler
// ─────────────────────────────────────────────────────────────────────────────

/// A source of "live memory, in bytes, right now".
///
/// Implementors return a monotonic-ish estimate of the memory the process is
/// currently holding. Two implementations ship with the crate: [`RssSampler`]
/// (always available) and `JemallocSampler` (with the `jemalloc` feature).
pub trait Sampler {
    /// Current live memory in bytes, or `None` if the reading FAILED (the platform
    /// API was unavailable this call). `None` is NOT the same as `Some(0)`: a failed
    /// read must not be recorded as zero bytes, which would depress peak/slope and
    /// bias the run toward a false pass. A failed sample is skipped; if none ever
    /// succeed the run is reported inconclusive. Should be cheap enough to call
    /// thousands of times per run.
    fn sample(&self) -> Option<u64>;
    /// Short human label for what is being measured (shown in reports).
    fn kind(&self) -> &'static str;
}

/// Process resident-set-size sampler. Always available, no allocator requirement,
/// but coarser than jemalloc: page-granular and inflated by allocator retention.
#[derive(Debug, Default, Clone, Copy)]
pub struct RssSampler;

impl Sampler for RssSampler {
    fn sample(&self) -> Option<u64> {
        // None (not 0) when the platform RSS API is unavailable, so a failed read
        // is skipped rather than injected as a spurious zero.
        memory_stats::memory_stats().map(|m| m.physical_mem as u64)
    }
    fn kind(&self) -> &'static str {
        "rss"
    }
}

/// In-process live-heap sampler backed by jemalloc `stats.allocated`.
///
/// Requires jemalloc to be the active global allocator (see the crate docs). The
/// reading is the number of bytes currently handed out by the allocator — the
/// cleanest signal for "did my data structures stop growing".
#[cfg(feature = "jemalloc")]
#[derive(Debug, Default, Clone, Copy)]
pub struct JemallocSampler;

#[cfg(feature = "jemalloc")]
impl Sampler for JemallocSampler {
    fn sample(&self) -> Option<u64> {
        use tikv_jemalloc_ctl::{epoch, stats};
        // `allocated` is only refreshed when the epoch is advanced.
        let _ = epoch::advance();
        // None (not 0) on a ctl read error, so a failed read is skipped.
        stats::allocated::read().ok().map(|v| v as u64)
    }
    fn kind(&self) -> &'static str {
        "jemalloc/allocated"
    }
}

/// The sampler [`soak`] and friends use when you don't pass one explicitly.
///
/// This is [`RssSampler`] — it reads the OS and needs no allocator setup, so the
/// zero-config path is just `cargo add navian-memcheck` plus one soak test. For a
/// cleaner in-process signal, enable the `jemalloc` feature and pass
/// `JemallocSampler` to [`soak_with`].
pub type DefaultSampler = RssSampler;

// ─────────────────────────────────────────────────────────────────────────────
// Linear fit (ordinary least squares)
// ─────────────────────────────────────────────────────────────────────────────

/// Result of an ordinary-least-squares fit of `y = slope * x + intercept`.
#[derive(Debug, Clone, Copy, PartialEq)]
pub struct Fit {
    /// Change in `y` per unit `x`.
    pub slope: f64,
    /// `y` at `x = 0`.
    pub intercept: f64,
    /// Coefficient of determination in `[0, 1]`; `1.0` is a perfect line.
    pub r2: f64,
}

/// Least-squares fit of `ys` against `xs`. Returns a zero-slope fit when there
/// are fewer than two points or `xs` has no spread.
pub fn linear_fit(xs: &[f64], ys: &[f64]) -> Fit {
    let n = xs.len().min(ys.len());
    if n < 2 {
        return Fit {
            slope: 0.0,
            intercept: ys.first().copied().unwrap_or(0.0),
            r2: 1.0,
        };
    }
    let nf = n as f64;
    let mean_x = xs[..n].iter().sum::<f64>() / nf;
    let mean_y = ys[..n].iter().sum::<f64>() / nf;
    let mut sxx = 0.0;
    let mut sxy = 0.0;
    let mut syy = 0.0;
    for i in 0..n {
        let dx = xs[i] - mean_x;
        let dy = ys[i] - mean_y;
        sxx += dx * dx;
        sxy += dx * dy;
        syy += dy * dy;
    }
    if sxx == 0.0 {
        return Fit {
            slope: 0.0,
            intercept: mean_y,
            r2: 1.0,
        };
    }
    let slope = sxy / sxx;
    let intercept = mean_y - slope * mean_x;
    // r2 = explained / total variance; guard the flat-y case.
    let r2 = if syy == 0.0 {
        1.0
    } else {
        (sxy * sxy) / (sxx * syy)
    };
    Fit {
        slope,
        intercept,
        r2: r2.clamp(0.0, 1.0),
    }
}

// ─────────────────────────────────────────────────────────────────────────────
// Soak
// ─────────────────────────────────────────────────────────────────────────────

/// How to run a soak: how long, how often to sample, and what "still growing"
/// and "too big" mean.
#[derive(Debug, Clone, Copy)]
pub struct SoakConfig {
    /// Number of times the workload closure is invoked.
    pub iterations: u64,
    /// Sample live memory every `sample_every` iterations. Must be `>= 1`.
    pub sample_every: u64,
    /// Fraction of the run (by sample count) to *discard* as warmup before
    /// measuring the plateau slope. `0.5` fits the line through the back half.
    pub warmup_frac: f64,
    /// Optional hard ceiling on peak live bytes. `None` disables the cap check.
    pub max_bytes: Option<u64>,
    /// Plateau budget: the largest back-half slope, in **bytes per sample**, that
    /// still counts as "leveled off". Anything above this fails as `StillGrowing`.
    pub slope_bytes_per_sample: f64,
    /// Optional minimum r² of the back-half fit required before an over-budget
    /// slope is treated as growth. **Off by default (`0.0`).**
    ///
    /// Raising it suppresses false `StillGrowing` on jittery RSS — but it is *not*
    /// on by default and should be used with care: a slow leak buried in page-noise
    /// also has a low r², so a high floor can mask the very bug this tool exists to
    /// catch. Prefer tuning [`slope_bytes_per_sample`](Self::slope_bytes_per_sample)
    /// above your platform's noise floor, or use the jemalloc sampler for a clean
    /// signal, over relying on this.
    pub min_r2_for_growth: f64,
    /// Minimum bytes the live-memory reading must span (`peak - trough`) over the
    /// run before a plateau is trusted. **Off by default (`0`).**
    ///
    /// A flat series is ambiguous: it is what a bounded workload looks like, but
    /// also what a workload that never allocated — or a stuck sampler — looks like.
    /// The tool can't tell which from the series, so by default it passes a flat
    /// series and merely reports [`SoakReport::moved_bytes`]. Set this to the amount
    /// you KNOW your workload should churn (domain knowledge the tool doesn't have),
    /// and a run that moved less is reported [`Verdict::InsufficientSamples`] instead
    /// of a hollow pass.
    pub min_movement_bytes: u64,
}

impl SoakConfig {
    /// A sensible default soak of `iterations` events: 200 samples, back-half
    /// plateau check, ~4 KB/sample slope budget, no hard cap.
    pub fn iterations(iterations: u64) -> Self {
        let sample_every = (iterations / 200).max(1);
        SoakConfig {
            iterations,
            sample_every,
            warmup_frac: 0.5,
            max_bytes: None,
            slope_bytes_per_sample: 4096.0,
            min_r2_for_growth: 0.0, // r² floor off by default — slope alone decides
            min_movement_bytes: 0,  // movement enforcement off by default
        }
    }
    /// Set the plateau slope budget in bytes per sample.
    #[must_use]
    pub fn slope_budget(mut self, bytes_per_sample: f64) -> Self {
        self.slope_bytes_per_sample = bytes_per_sample;
        self
    }
    /// Require the run to move at least `bytes` (`peak - trough`) before a plateau is
    /// trusted; a run that moved less is reported [`Verdict::InsufficientSamples`].
    /// See [`min_movement_bytes`](Self::min_movement_bytes).
    #[must_use]
    pub fn require_movement(mut self, bytes: u64) -> Self {
        self.min_movement_bytes = bytes;
        self
    }
    /// Set a hard ceiling on peak live bytes.
    #[must_use]
    pub fn max_bytes(mut self, cap: u64) -> Self {
        self.max_bytes = Some(cap);
        self
    }
    /// Set how many iterations pass between samples.
    #[must_use]
    pub fn sample_every(mut self, every: u64) -> Self {
        self.sample_every = every.max(1);
        self
    }
    /// Set the warmup fraction discarded before the plateau fit. The value is
    /// clamped to `0.0..=0.95`.
    #[must_use]
    pub fn warmup_frac(mut self, frac: f64) -> Self {
        self.warmup_frac = frac.clamp(0.0, 0.95);
        self
    }
    /// Set the optional back-half r² floor required before an over-budget slope
    /// counts as growth (clamped to `0.0..=1.0`; `0.0`, the default, disables it).
    /// Advanced: see the caveat on [`SoakConfig::min_r2_for_growth`] — a high floor
    /// can mask a slow, noisy leak.
    #[must_use]
    pub fn min_r2(mut self, r2: f64) -> Self {
        self.min_r2_for_growth = r2.clamp(0.0, 1.0);
        self
    }
}

/// Why a soak passed or failed.
#[derive(Debug, Clone, Copy, PartialEq)]
pub enum Verdict {
    /// Memory plateaued and stayed under any configured cap.
    Pass,
    /// Peak live bytes exceeded the configured hard cap.
    ExceededCap {
        /// Observed peak.
        peak: u64,
        /// Configured ceiling.
        cap: u64,
    },
    /// Back-half slope exceeded the plateau budget — memory was still climbing.
    StillGrowing {
        /// Observed back-half slope, bytes per sample.
        slope: f64,
        /// Configured budget, bytes per sample.
        budget: f64,
    },
    /// The run produced too little signal to judge — fewer than two samples, or
    /// every sample read zero bytes (sampler unsupported on this platform). This
    /// is **not** a pass; treat it as a failed run to investigate.
    InsufficientSamples {
        /// Human-readable reason the run was inconclusive.
        reason: &'static str,
    },
}

/// The outcome of a [`soak`] run: the raw samples plus the computed verdict.
#[derive(Debug, Clone)]
pub struct SoakReport {
    /// `(iteration, live_bytes)` pairs, in order.
    pub samples: Vec<(u64, u64)>,
    /// Live bytes at the first sample (rough baseline).
    pub baseline: u64,
    /// Peak live bytes observed across the whole run.
    pub peak: u64,
    /// Total spread of live bytes observed: `peak - trough`. This is the evidence
    /// that the run actually exercised memory. When it is `0` every sample read the
    /// same value — the workload may never allocate (or never hit the suspected
    /// path), or the sampler may be stuck — and a flat "plateau" then proves
    /// nothing, so the verdict is [`Verdict::InsufficientSamples`], not a pass.
    pub moved_bytes: u64,
    /// Back-half slope in bytes per sample (the plateau signal).
    pub back_half_slope: f64,
    /// R² of the back-half fit (how line-like the tail was).
    pub back_half_r2: f64,
    /// What the sampler measured.
    pub sampler_kind: &'static str,
    /// Pass / fail and why.
    pub verdict: Verdict,
    /// Advisory warnings surfaced on a PASS that may be hollow — e.g. the run
    /// barely moved, the slope budget is loose enough to hide a real leak, or the
    /// plateau rests on too few samples. They do NOT change the verdict (defaults
    /// keep the tool zero-config), but each names the knob to set so a green result
    /// is earned, not assumed. Empty on a fail or a clearly-earned pass.
    pub trust_warnings: Vec<String>,
}

impl SoakReport {
    /// `true` if the verdict is [`Verdict::Pass`].
    pub fn passed(&self) -> bool {
        matches!(self.verdict, Verdict::Pass)
    }

    /// Panic with [`summary`](Self::summary) if the soak did not pass. The
    /// idiomatic last line of a `#[test]`.
    ///
    /// # Panics
    ///
    /// Panics if the verdict is anything other than [`Verdict::Pass`] — i.e. the
    /// memory kept growing, breached a cap, or the run was inconclusive.
    pub fn assert(&self) {
        assert!(self.passed(), "navian-memcheck: {}", self.summary());
    }

    /// One-line, human-readable result summary.
    pub fn summary(&self) -> String {
        let mb = |b: u64| b as f64 / (1024.0 * 1024.0);
        let verdict = match self.verdict {
            Verdict::Pass => "PASS — memory plateaued".to_string(),
            Verdict::ExceededCap { peak, cap } => {
                format!("FAIL — peak {:.1} MB exceeded cap {:.1} MB", mb(peak), mb(cap))
            }
            Verdict::StillGrowing { slope, budget } => format!(
                "FAIL — still growing: back-half slope {slope:.0} B/sample > budget {budget:.0} B/sample"
            ),
            Verdict::InsufficientSamples { reason } => {
                format!("INCONCLUSIVE — {reason}")
            }
        };
        let mut out = format!(
            "{verdict} [sampler={}, samples={}, baseline={:.1} MB, peak={:.1} MB, moved={:.1} MB, slope={:.0} B/sample, r2={:.2}]",
            self.sampler_kind,
            self.samples.len(),
            mb(self.baseline),
            mb(self.peak),
            mb(self.moved_bytes),
            self.back_half_slope,
            self.back_half_r2,
        );
        for warn in &self.trust_warnings {
            out.push_str("\n  ⚠ trust: ");
            out.push_str(warn);
        }
        out
    }
}

impl fmt::Display for SoakReport {
    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
        f.write_str(&self.summary())
    }
}

/// Run `work` for `cfg.iterations`, sampling live heap with the [`DefaultSampler`],
/// and return a [`SoakReport`]. See [`soak_with`] to supply a sampler.
pub fn soak(cfg: &SoakConfig, work: impl FnMut(u64)) -> SoakReport {
    soak_with(cfg, DefaultSampler::default(), work)
}

/// One-line convenience: soak `iterations` of `work` with default settings and
/// assert the result plateaus. Equivalent to
/// `soak(&SoakConfig::iterations(iterations), work).assert()`.
///
/// ```no_run
/// # fn process_one_event(_: u64) {}
/// navian_memcheck::assert_plateau(200_000, |i| process_one_event(i));
/// ```
///
/// # Panics
///
/// Panics if memory did not plateau (see [`SoakReport::assert`]).
pub fn assert_plateau(iterations: u64, work: impl FnMut(u64)) {
    soak(&SoakConfig::iterations(iterations), work).assert();
}

/// Like [`soak`] but with an explicit [`Sampler`].
// The sampler impls are zero-sized; taking one by value reads naturally at the
// call site (`soak_with(cfg, RssSampler, work)`) and costs nothing.
#[allow(clippy::needless_pass_by_value)]
pub fn soak_with<S: Sampler>(
    cfg: &SoakConfig,
    sampler: S,
    mut work: impl FnMut(u64),
) -> SoakReport {
    let every = cfg.sample_every.max(1);
    let mut samples: Vec<(u64, u64)> = Vec::with_capacity((cfg.iterations / every) as usize + 1);

    for i in 0..cfg.iterations {
        work(i);
        // Skip a FAILED read rather than record it as 0 — a spurious zero would
        // depress peak/slope and bias the run toward a false pass.
        if (i + 1) % every == 0 {
            if let Some(bytes) = sampler.sample() {
                samples.push((i + 1, bytes));
            }
        }
    }
    // Guarantee at least one tail sample even if iterations isn't a multiple of `every`
    // (only when the tail read succeeds).
    if cfg.iterations > 0 && samples.last().map(|(it, _)| *it) != Some(cfg.iterations) {
        if let Some(bytes) = sampler.sample() {
            samples.push((cfg.iterations, bytes));
        }
    }

    finalize(cfg, samples, sampler.kind())
}

/// Build a [`SoakReport`] from samples collected *outside* this process — e.g.
/// the CLI polling another process's RSS. Pass `(tick, bytes)` pairs with one
/// tick per sample and set `cfg.sample_every == 1` so the slope is reported in
/// bytes-per-sample.
pub fn report_from_samples(
    cfg: &SoakConfig,
    samples: Vec<(u64, u64)>,
    sampler_kind: &'static str,
) -> SoakReport {
    finalize(cfg, samples, sampler_kind)
}

/// Compute peak, back-half slope, and verdict from collected samples.
fn finalize(cfg: &SoakConfig, samples: Vec<(u64, u64)>, sampler_kind: &'static str) -> SoakReport {
    let baseline = samples.first().map_or(0, |(_, b)| *b);
    let peak = samples.iter().map(|(_, b)| *b).max().unwrap_or(0);
    let trough = samples.iter().map(|(_, b)| *b).min().unwrap_or(0);
    // Spread over the WHOLE run — the evidence the workload exercised memory at all.
    let moved_bytes = peak.saturating_sub(trough);

    // Guard the DEGENERATE runs that can't be fit at all. Everything else (cap
    // breach, growth, movement) is decided AFTER the line fit below, in priority
    // order, so the fitted slope/r² is real on every non-degenerate report and the
    // movement guard can only ever downgrade a would-be PASS — never a detected
    // leak (`StillGrowing`) or a cap breach.
    let inconclusive_reason = if samples.len() < 2 {
        Some("fewer than 2 samples collected")
    } else if peak == 0 {
        Some("every sample read 0 bytes — is the sampler supported on this platform?")
    } else {
        None
    };
    if let Some(reason) = inconclusive_reason {
        return SoakReport {
            samples,
            baseline,
            peak,
            moved_bytes,
            back_half_slope: 0.0,
            back_half_r2: 0.0,
            sampler_kind,
            verdict: Verdict::InsufficientSamples { reason },
            trust_warnings: Vec::new(),
        };
    }

    // Sanitize thresholds here (not just in the builders) so a struct-literal
    // SoakConfig can never *disable* detection with a non-finite value.
    let budget = if cfg.slope_bytes_per_sample.is_finite() {
        cfg.slope_bytes_per_sample
    } else {
        0.0 // a non-finite budget errs toward FAIL, never toward a silent pass
    };
    let min_r2 = if cfg.min_r2_for_growth.is_finite() {
        cfg.min_r2_for_growth.clamp(0.0, 1.0)
    } else {
        0.0
    };
    let warmup = if cfg.warmup_frac.is_finite() {
        cfg.warmup_frac.clamp(0.0, 0.95)
    } else {
        0.5
    };
    // Fit a line through the back half (or configured tail) of the samples.
    let n = samples.len();
    let start = ((n as f64) * warmup).floor() as usize;
    let start = start.min(n.saturating_sub(2)); // keep at least 2 points if we can
    let tail = &samples[start..];
    // Fit against sample INDEX (0,1,2,…), NOT the caller-supplied ticks. For the
    // uniform cadence of an internal soak this is identical to fitting the ticks and
    // rescaling to bytes-per-sample, but it is robust on the public sample-based API:
    // duplicate ticks (a zero x-range) would fit slope 0 and FALSE-PASS growing
    // memory, and non-monotonic ticks would underflow the unsigned offset. Index x
    // sidesteps both — and bytes-per-sample is exactly the cadence-independent metric
    // we want.
    let xs: Vec<f64> = (0..tail.len()).map(|i| i as f64).collect();
    let ys: Vec<f64> = tail.iter().map(|(_, b)| *b as f64).collect();
    let fit = linear_fit(&xs, &ys);
    let slope_per_sample = fit.slope;

    // Verdict in PRIORITY order, so a definitive failure always beats "inconclusive":
    //   1. cap breach — too much memory, a FAIL regardless of shape or movement;
    //   2. growth — slope over budget is a detected leak (a FAIL), and must NOT be
    //      downgraded to "inconclusive" just because total movement was small;
    //   3. insufficient movement — only NOW, for a would-be PASS, does the opt-in
    //      `require_movement` guard apply: a flat run that never moved enough proves
    //      nothing, so it is inconclusive rather than a hollow pass;
    //   4. otherwise the memory plateaued → PASS.
    //
    // Growth uses the optional r² floor (off by default): a slow leak buried in
    // page-noise has a low r², so gating on r² by default would mask the exact bug
    // this tool exists to catch. The budget is the noise knob; for a clean signal
    // use the jemalloc sampler.
    let verdict = if let Some(cap) = cfg.max_bytes.filter(|&c| peak > c) {
        Verdict::ExceededCap { peak, cap }
    } else if slope_per_sample > budget && fit.r2 >= min_r2 {
        Verdict::StillGrowing {
            slope: slope_per_sample,
            budget,
        }
    } else if cfg.min_movement_bytes > 0 && moved_bytes < cfg.min_movement_bytes {
        Verdict::InsufficientSamples {
            reason: "memory moved less than the required minimum — the workload may \
                     not allocate (or never hit the suspected path), or the sampler \
                     may be stuck. A flat series that never moved proves nothing.",
        }
    } else {
        Verdict::Pass
    };

    // On a PASS, flag the ways it might be hollow so a green is earned, not assumed.
    // These are advisory (the verdict stands) but each names the knob to set.
    let trust_warnings = if matches!(verdict, Verdict::Pass) {
        pass_trust_warnings(peak, moved_bytes, cfg.min_movement_bytes, budget, tail.len())
    } else {
        Vec::new()
    };

    SoakReport {
        samples,
        baseline,
        peak,
        moved_bytes,
        back_half_slope: slope_per_sample,
        back_half_r2: fit.r2,
        sampler_kind,
        verdict,
        trust_warnings,
    }
}

/// Warnings for a PASS that may not be trustworthy. Each fires on a heuristic that
/// a green could be hollow, and names the exact knob to make it meaningful. All are
/// advisory — the defaults keep the tool zero-config; these push you to earn the
/// pass. `back_half_len` is the number of samples the plateau slope was fit over.
fn pass_trust_warnings(
    peak: u64,
    moved_bytes: u64,
    min_movement_bytes: u64,
    budget: f64,
    back_half_len: usize,
) -> Vec<String> {
    let mb = |b: f64| b / (1024.0 * 1024.0);
    let mut w = Vec::new();

    // 1) Barely moved: a flat run can hide a workload that never allocated or a
    //    stuck sampler. Only nag when the caller has NOT already set the hard knob.
    if min_movement_bytes == 0 && moved_bytes.saturating_mul(20) < peak {
        w.push(format!(
            "memory moved only {:.1} MB across the run ({:.1} MB peak) — a nearly-flat \
             run can hide a workload that never exercised the leak path, or a stuck \
             sampler. Set require_movement / --min-movement to the churn you expect.",
            mb(moved_bytes as f64),
            mb(peak as f64),
        ));
    }

    // 2) Loose budget: the tolerated cumulative growth over the fitted window is as
    //    large as the whole peak, so a real leak up to that size would still pass.
    //    (Threshold is the full peak, not half, to avoid crying wolf on every pass.)
    let tolerated = budget * back_half_len as f64;
    if peak > 0 && tolerated >= peak as f64 {
        w.push(format!(
            "slope budget tolerates ~{:.1} MB of growth over this window (~{:.0}% of the \
             {:.1} MB peak) — a leak up to that size would still pass. Tighten \
             slope_budget / --slope-budget toward your platform's noise floor.",
            mb(tolerated),
            (tolerated / peak as f64) * 100.0,
            mb(peak as f64),
        ));
    }

    // 3) Too few points: a short tail fits a line trivially (2 points → r²=1) and a
    //    slow leak may not have ramped yet.
    if back_half_len < 8 {
        w.push(format!(
            "plateau rests on only {back_half_len} back-half sample(s) — too few to \
             trust; a slow leak may not have ramped yet. Soak longer or sample more \
             (more iterations / smaller --interval) so the tail has >= 8 points.",
        ));
    }

    // NOTE: a step/sawtooth leak that sits flat WITHIN the fitted tail between steps
    // is not flagged here — a clean heuristic is elusive (a legitimate plateau also
    // ends at its peak level, and ties make peak-position useless). The whole-run
    // `moved_bytes` is reported so a large spread under a flat slope is at least
    // visible for inspection; a robust monotonic-trend statistic is a future add.

    w
}

// ─────────────────────────────────────────────────────────────────────────────
// Bounded-state assertions
// ─────────────────────────────────────────────────────────────────────────────

/// Assert that a size metric never exceeds `cap` as a driver sweeps over
/// `drivers`. Use this to prove a structure is bounded regardless of input scale
/// — e.g. "no matter how many distinct sessions arrive, the session map holds at
/// most `cap` entries."
///
/// # Panics
///
/// Panics on the first driver value whose measurement exceeds `cap`.
pub fn assert_bounded<I>(cap: u64, drivers: I, mut measure: impl FnMut(u64) -> u64)
where
    I: IntoIterator<Item = u64>,
{
    for d in drivers {
        let got = measure(d);
        assert!(
            got <= cap,
            "navian-memcheck: bound violated at driver={d}: measured {got} > cap {cap}"
        );
    }
}

/// Fit `(driver, bytes)` growth points and return the [`Fit`]. `slope` is bytes
/// per unit of driver; a bounded-per-item structure has a small, stable slope and
/// high `r2`.
pub fn fit_growth(points: &[(u64, u64)]) -> Fit {
    // Offset x by the MINIMUM driver value (not the first) to preserve f64 precision
    // on large inputs AND stay panic-free on unsorted/non-monotonic drivers — a first
    // offset would underflow the unsigned subtraction (debug panic / release wrap to a
    // huge x, corrupting the fit) whenever a later driver is smaller than the first.
    let x0 = points.iter().map(|(d, _)| *d).min().unwrap_or(0);
    let xs: Vec<f64> = points.iter().map(|(d, _)| (d - x0) as f64).collect();
    let ys: Vec<f64> = points.iter().map(|(_, b)| *b as f64).collect();
    linear_fit(&xs, &ys)
}

/// Assert that memory grows *at most linearly* in a driver and no faster than
/// `max_bytes_per_unit`. Catches super-linear blowups (e.g. an accidental O(n²)
/// retained buffer) that a single-point check would miss.
///
/// # Panics
///
/// Panics if the fitted slope exceeds `max_bytes_per_unit`.
pub fn assert_linear_in(points: &[(u64, u64)], max_bytes_per_unit: f64) -> Fit {
    // A line needs ≥2 points; fewer can't measure a slope, and a non-finite or
    // negative budget would silently DISABLE the assertion. Reject both loudly rather
    // than let a misuse turn into a false pass.
    assert!(
        points.len() >= 2,
        "navian-memcheck: assert_linear_in needs at least 2 points, got {}",
        points.len()
    );
    // Distinct driver values are required, not just ≥2 points: with no x-spread the
    // least-squares slope collapses to 0 and would FALSE-PASS any growth. Reject it.
    let x_min = points.iter().map(|(d, _)| *d).min().unwrap();
    let x_max = points.iter().map(|(d, _)| *d).max().unwrap();
    assert!(
        x_max > x_min,
        "navian-memcheck: assert_linear_in needs at least two distinct driver values (all were {x_min})"
    );
    assert!(
        max_bytes_per_unit.is_finite() && max_bytes_per_unit >= 0.0,
        "navian-memcheck: max_bytes_per_unit must be finite and non-negative, got {max_bytes_per_unit}"
    );
    let fit = fit_growth(points);
    assert!(
        fit.slope <= max_bytes_per_unit,
        "navian-memcheck: growth too steep: {:.1} B/unit > budget {:.1} B/unit (r2={:.2})",
        fit.slope,
        max_bytes_per_unit,
        fit.r2
    );
    fit
}

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

    #[test]
    fn linear_fit_recovers_slope() {
        let xs = [0.0, 1.0, 2.0, 3.0, 4.0];
        let ys = [1.0, 3.0, 5.0, 7.0, 9.0]; // y = 2x + 1
        let fit = linear_fit(&xs, &ys);
        assert!((fit.slope - 2.0).abs() < 1e-9);
        assert!((fit.intercept - 1.0).abs() < 1e-9);
        assert!((fit.r2 - 1.0).abs() < 1e-9);
    }

    #[test]
    fn duplicate_ticks_do_not_hide_growth() {
        // Every sample carries the SAME tick (0). Fitting against ticks would give a
        // zero x-range → slope 0 → FALSE PASS. Fitting against sample index catches it.
        let samples: Vec<(u64, u64)> = (0..12)
            .map(|i| (0u64, 100_000_000 + i * 10_000_000))
            .collect();
        let r = report_from_samples(&SoakConfig::iterations(200), samples, "test");
        assert!(
            matches!(r.verdict, Verdict::StillGrowing { .. }),
            "duplicate ticks must not mask growth, got {:?}",
            r.verdict
        );
    }

    #[test]
    fn non_monotonic_ticks_do_not_panic_and_still_detect() {
        // Decreasing/unsorted ticks previously underflowed the unsigned offset
        // (debug panic / release wrap). Index-based fit is immune; growth by index
        // is still flagged.
        let samples: Vec<(u64, u64)> = vec![
            (100, 100_000_000),
            (50, 110_000_000),
            (200, 120_000_000),
            (10, 130_000_000),
            (150, 140_000_000),
            (5, 150_000_000),
        ];
        let r = report_from_samples(&SoakConfig::iterations(200), samples, "test");
        assert!(matches!(r.verdict, Verdict::StillGrowing { .. }));
    }

    #[test]
    fn flat_after_warmup_passes() {
        // Guard against a false FAIL: a series that RISES during warmup then levels
        // off is a genuine plateau (moved > 0, back-half slope ~0) → PASS.
        let mut samples: Vec<(u64, u64)> = vec![(0, 90_000_000)];
        samples.extend((1..12).map(|i| (i, 100_000_000)));
        let r = report_from_samples(&SoakConfig::iterations(200), samples, "test");
        assert!(matches!(r.verdict, Verdict::Pass), "flat-after-warmup must pass, got {:?}", r.verdict);
    }

    #[test]
    fn constant_series_passes_by_default_but_reports_zero_movement() {
        // By default a flat constant series PASSES (a bounded workload looks flat),
        // but its zero movement is visible so a green isn't blindly trusted.
        let samples: Vec<(u64, u64)> = (0..50).map(|i| (i, 10_000_000)).collect();
        let r = report_from_samples(&SoakConfig::iterations(50), samples.clone(), "test");
        assert!(matches!(r.verdict, Verdict::Pass), "flat passes by default, got {:?}", r.verdict);
        assert_eq!(r.moved_bytes, 0, "but zero movement is reported");
    }

    #[test]
    fn hollow_pass_raises_trust_warnings() {
        // A flat, near-zero-movement series over a short tail PASSES by default but
        // must carry advisory warnings pointing at the knobs.
        let samples: Vec<(u64, u64)> = (0..12).map(|i| (i, 10_000_000)).collect();
        let r = report_from_samples(&SoakConfig::iterations(12).sample_every(1), samples, "test");
        assert!(matches!(r.verdict, Verdict::Pass));
        assert!(!r.trust_warnings.is_empty(), "hollow pass must warn");
        // moved 0 → the movement warning fires.
        assert!(r.trust_warnings.iter().any(|w| w.contains("moved only")));
    }

    #[test]
    fn earned_pass_has_no_trust_warnings() {
        // Real movement, tight budget, plenty of samples → a clean pass, no nags.
        let mut samples: Vec<(u64, u64)> = (0..20).map(|i| (i, 5_000_000 + i * 250_000)).collect();
        samples.extend((20..80).map(|i| (i, 10_000_000)));
        let cfg = SoakConfig::iterations(80).sample_every(1).slope_budget(1024.0);
        let r = report_from_samples(&cfg, samples, "test");
        assert!(matches!(r.verdict, Verdict::Pass), "{}", r.summary());
        assert!(
            r.trust_warnings.is_empty(),
            "earned pass should not warn, got: {:?}",
            r.trust_warnings
        );
    }

    #[test]
    fn fail_carries_no_trust_warnings() {
        // Warnings are a PASS concept; a growing series just fails.
        let samples: Vec<(u64, u64)> = (0..80).map(|i| (i, 1_000_000 + i * 500_000)).collect();
        let cfg = SoakConfig::iterations(80).sample_every(1).slope_budget(4096.0);
        let r = report_from_samples(&cfg, samples, "test");
        assert!(matches!(r.verdict, Verdict::StillGrowing { .. }));
        assert!(r.trust_warnings.is_empty());
    }

    #[test]
    fn a_detected_leak_is_not_downgraded_by_require_movement() {
        // A clearly-growing series whose TOTAL spread is below the movement floor
        // must still FAIL as StillGrowing — a detected leak is never downgraded to
        // "inconclusive". (Growth is decided before the movement guard.)
        let samples: Vec<(u64, u64)> = (0..200).map(|i| (i, 1_000_000 + i * 400_000)).collect();
        let cfg = SoakConfig::iterations(200)
            .sample_every(1)
            .slope_budget(4096.0)
            .require_movement(10_000_000_000); // absurdly high floor
        let r = report_from_samples(&cfg, samples, "test");
        assert!(
            matches!(r.verdict, Verdict::StillGrowing { .. }),
            "a leak must fail, not go inconclusive; got {:?}",
            r.verdict
        );
    }

    #[test]
    fn cap_breach_carries_real_slope_and_beats_growth() {
        // A rising, over-cap series → ExceededCap (cap beats growth), and the report
        // now carries the real fitted slope (not a hard-coded 0).
        let samples: Vec<(u64, u64)> = (0..50).map(|i| (i, 20_000_000 + i * 1_000_000)).collect();
        let cfg = SoakConfig::iterations(50).sample_every(1).max_bytes(10_000_000);
        let r = report_from_samples(&cfg, samples, "test");
        assert!(matches!(r.verdict, Verdict::ExceededCap { .. }));
        assert!(r.back_half_slope > 0.0, "cap report should carry the real slope");
    }

    #[test]
    fn single_over_cap_sample_is_inconclusive_not_cap() {
        // A 1-sample run can't be fit; the degenerate guard wins over the cap check
        // (restores the pre-reorder behavior for sub-2-sample runs).
        let samples = vec![(0u64, 50_000_000u64)];
        let cfg = SoakConfig::iterations(1).max_bytes(10_000_000);
        let r = report_from_samples(&cfg, samples, "test");
        assert!(matches!(r.verdict, Verdict::InsufficientSamples { .. }), "got {:?}", r.verdict);
    }

    #[test]
    fn require_movement_makes_a_flat_run_inconclusive() {
        // Opt in: a run that moved less than required is inconclusive, not a hollow
        // pass — the domain threshold is the caller's to set.
        let samples: Vec<(u64, u64)> = (0..50).map(|i| (i, 10_000_000)).collect();
        let cfg = SoakConfig::iterations(50).require_movement(1_000_000);
        let r = report_from_samples(&cfg, samples, "test");
        assert!(
            matches!(r.verdict, Verdict::InsufficientSamples { .. }),
            "require_movement must make a zero-movement run inconclusive, got {:?}",
            r.verdict
        );
    }

    #[test]
    fn fit_growth_handles_unsorted_drivers() {
        // Must not panic on unsorted/non-monotonic drivers (min-offset, not first).
        let fit = fit_growth(&[(100, 1), (1, 2), (50, 3), (0, 4)]);
        assert!(fit.slope.is_finite());
    }

    #[test]
    #[should_panic(expected = "at least 2 points")]
    fn assert_linear_in_rejects_too_few_points() {
        assert_linear_in(&[(1, 100)], 1.0);
    }

    #[test]
    #[should_panic(expected = "finite and non-negative")]
    fn assert_linear_in_rejects_infinite_budget() {
        assert_linear_in(&[(1, 100), (2, 200)], f64::INFINITY);
    }

    #[test]
    #[should_panic(expected = "distinct driver")]
    fn assert_linear_in_rejects_no_driver_spread() {
        // Two points, same driver → zero x-spread → slope 0 → would false-pass.
        assert_linear_in(&[(7, 0), (7, 1_000_000_000)], 0.0);
    }

    #[test]
    fn flat_after_warmup_is_a_plateau() {
        // Ramps for the first 10 samples, then flat for 90 — a real plateau with
        // movement, back-half slope ~0.
        let mut samples: Vec<(u64, u64)> = (0..10).map(|i| (i, 9_000_000 + i * 100_000)).collect();
        samples.extend((10..100).map(|i| (i, 10_000_000)));
        let cfg = SoakConfig::iterations(100);
        let report = finalize(&cfg, samples, "test");
        assert!(report.passed(), "{}", report.summary());
        assert!(report.back_half_slope.abs() < 1.0);
        assert!(report.moved_bytes > 0);
    }

    #[test]
    fn rising_series_still_growing() {
        // +100 KB per sample, forever.
        let samples: Vec<(u64, u64)> = (0..100).map(|i| (i, 1_000_000 + i * 100_000)).collect();
        let cfg = SoakConfig::iterations(100)
            .sample_every(1)
            .slope_budget(4096.0);
        let report = finalize(&cfg, samples, "test");
        assert!(!report.passed());
        matches!(report.verdict, Verdict::StillGrowing { .. });
    }

    #[test]
    fn cap_breach_beats_slope_check() {
        let samples: Vec<(u64, u64)> = (0..50).map(|i| (i, 50_000_000)).collect();
        let cfg = SoakConfig::iterations(50).max_bytes(10_000_000);
        let report = finalize(&cfg, samples, "test");
        assert!(matches!(report.verdict, Verdict::ExceededCap { .. }));
    }

    #[test]
    fn assert_linear_accepts_bounded_growth() {
        // 64 bytes per entry, exactly linear.
        let pts: Vec<(u64, u64)> = (0..1000).step_by(50).map(|n| (n, n * 64)).collect();
        let fit = assert_linear_in(&pts, 128.0);
        assert!((fit.slope - 64.0).abs() < 1.0);
    }

    #[test]
    #[should_panic(expected = "bound violated")]
    fn assert_bounded_catches_unbounded() {
        // measured == driver, so it blows past a fixed cap.
        assert_bounded(500, (0..1000u64).step_by(100), |d| d);
    }
}