#![forbid(unsafe_code)]
#![warn(missing_docs)]
use std::fmt;
pub trait Sampler {
fn sample(&self) -> Option<u64>;
fn kind(&self) -> &'static str;
}
#[derive(Debug, Default, Clone, Copy)]
pub struct RssSampler;
impl Sampler for RssSampler {
fn sample(&self) -> Option<u64> {
memory_stats::memory_stats().map(|m| m.physical_mem as u64)
}
fn kind(&self) -> &'static str {
"rss"
}
}
#[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};
let _ = epoch::advance();
stats::allocated::read().ok().map(|v| v as u64)
}
fn kind(&self) -> &'static str {
"jemalloc/allocated"
}
}
pub type DefaultSampler = RssSampler;
#[derive(Debug, Clone, Copy, PartialEq)]
pub struct Fit {
pub slope: f64,
pub intercept: f64,
pub r2: f64,
}
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;
let r2 = if syy == 0.0 {
1.0
} else {
(sxy * sxy) / (sxx * syy)
};
Fit {
slope,
intercept,
r2: r2.clamp(0.0, 1.0),
}
}
#[derive(Debug, Clone, Copy)]
pub struct SoakConfig {
pub iterations: u64,
pub sample_every: u64,
pub warmup_frac: f64,
pub max_bytes: Option<u64>,
pub slope_bytes_per_sample: f64,
pub min_r2_for_growth: f64,
pub min_movement_bytes: u64,
}
impl SoakConfig {
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, min_movement_bytes: 0, }
}
#[must_use]
pub fn slope_budget(mut self, bytes_per_sample: f64) -> Self {
self.slope_bytes_per_sample = bytes_per_sample;
self
}
#[must_use]
pub fn require_movement(mut self, bytes: u64) -> Self {
self.min_movement_bytes = bytes;
self
}
#[must_use]
pub fn max_bytes(mut self, cap: u64) -> Self {
self.max_bytes = Some(cap);
self
}
#[must_use]
pub fn sample_every(mut self, every: u64) -> Self {
self.sample_every = every.max(1);
self
}
#[must_use]
pub fn warmup_frac(mut self, frac: f64) -> Self {
self.warmup_frac = frac.clamp(0.0, 0.95);
self
}
#[must_use]
pub fn min_r2(mut self, r2: f64) -> Self {
self.min_r2_for_growth = r2.clamp(0.0, 1.0);
self
}
}
#[derive(Debug, Clone, Copy, PartialEq)]
pub enum Verdict {
Pass,
ExceededCap {
peak: u64,
cap: u64,
},
StillGrowing {
slope: f64,
budget: f64,
},
InsufficientSamples {
reason: &'static str,
},
}
#[derive(Debug, Clone)]
pub struct SoakReport {
pub samples: Vec<(u64, u64)>,
pub baseline: u64,
pub peak: u64,
pub moved_bytes: u64,
pub back_half_slope: f64,
pub back_half_r2: f64,
pub sampler_kind: &'static str,
pub verdict: Verdict,
pub trust_warnings: Vec<String>,
}
impl SoakReport {
pub fn passed(&self) -> bool {
matches!(self.verdict, Verdict::Pass)
}
pub fn assert(&self) {
assert!(self.passed(), "navian-memcheck: {}", self.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())
}
}
pub fn soak(cfg: &SoakConfig, work: impl FnMut(u64)) -> SoakReport {
soak_with(cfg, DefaultSampler::default(), work)
}
pub fn assert_plateau(iterations: u64, work: impl FnMut(u64)) {
soak(&SoakConfig::iterations(iterations), work).assert();
}
#[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);
if (i + 1) % every == 0 {
if let Some(bytes) = sampler.sample() {
samples.push((i + 1, bytes));
}
}
}
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())
}
pub fn report_from_samples(
cfg: &SoakConfig,
samples: Vec<(u64, u64)>,
sampler_kind: &'static str,
) -> SoakReport {
finalize(cfg, samples, sampler_kind)
}
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);
let moved_bytes = peak.saturating_sub(trough);
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(),
};
}
let budget = if cfg.slope_bytes_per_sample.is_finite() {
cfg.slope_bytes_per_sample
} else {
0.0 };
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
};
let n = samples.len();
let start = ((n as f64) * warmup).floor() as usize;
let start = start.min(n.saturating_sub(2)); let tail = &samples[start..];
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;
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
};
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,
}
}
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();
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),
));
}
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),
));
}
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.",
));
}
w
}
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}"
);
}
}
pub fn fit_growth(points: &[(u64, u64)]) -> Fit {
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)
}
pub fn assert_linear_in(points: &[(u64, u64)], max_bytes_per_unit: f64) -> Fit {
assert!(
points.len() >= 2,
"navian-memcheck: assert_linear_in needs at least 2 points, got {}",
points.len()
);
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]; 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() {
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() {
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() {
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() {
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() {
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");
assert!(r.trust_warnings.iter().any(|w| w.contains("moved only")));
}
#[test]
fn earned_pass_has_no_trust_warnings() {
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() {
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() {
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); 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() {
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() {
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() {
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() {
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() {
assert_linear_in(&[(7, 0), (7, 1_000_000_000)], 0.0);
}
#[test]
fn flat_after_warmup_is_a_plateau() {
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() {
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() {
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() {
assert_bounded(500, (0..1000u64).step_by(100), |d| d);
}
}