plugins/native/znippy-plugin-git/src/memguard.rs
//! **The indexer must never take the box down.**
//!
//! MEASURED 2026-08-29 21:10 UTC on the gunnar appliance (16 GB, 4 vCPU, no
//! swap, PID 1 = gunnar-server): a `push --mirror` of torvalds/linux (8.12 GiB
//! pack, 13 948 244 objects, 11 215 872 deltas) was appended and acked, then the
//! post-ack indexer grew past 16 GB and the kernel said
//! `Kernel panic - not syncing: System is deadlocked on memory`. PID 1 was the
//! only process, so the OOM killer had nothing to kill — and a kernel that cannot
//! kill anything panics.
//!
//! This module is the guard that makes that impossible by construction rather
//! than by hoping the pack is small: **before every batch and every few thousand
//! objects inside one**, the worker reads `MemAvailable` out of `/proc/meminfo`
//! and, below a budget, *flushes what it holds and waits*. The kernel's own
//! estimate is used because it already accounts for reclaimable page cache —
//! the mmapped pack ([`crate::archive_map`]) costs nothing against it, only the
//! anonymous heap this process really owns does.
//!
//! | knob | default | what |
//! |---|---:|---|
//! | `ZNIPPY_GIT_MEM_BUDGET_PCT` | `20` | pause when `MemAvailable` < this % of `MemTotal` |
//! | `ZNIPPY_GIT_MEM_BUDGET_BYTES` | — | absolute override of the same floor |
//! | `ZNIPPY_GIT_MEMINFO` | `/proc/meminfo` | the file read — a test hands in a fake |
//!
//! The pause is **one WARN line** when it starts (`index.paused`, which the
//! gunnar dashboard renders as `index paused: 2.9G available < 3.2G budget`),
//! a re-check every 250 ms, and **one INFO line** when it resumes. Nothing is
//! logged per poll.
//!
//! What "flush what it holds" means is the caller's: [`MemWatchdog::pause_until_ok`]
//! takes the flush as a closure, because the resolver's droppable state (its
//! held delta bases, [`crate::resolve`]) and the sink's buffered rows are not
//! this module's to know about.
use std::path::{Path, PathBuf};
use std::sync::atomic::{AtomicBool, AtomicU64, Ordering};
use std::sync::{Arc, OnceLock, RwLock};
use std::time::{Duration, Instant};
use anyhow::Result;
/// The default floor: pause when less than this share of `MemTotal` is available.
pub const DEFAULT_BUDGET_PCT: u64 = 20;
/// Re-check cadence while paused.
pub const POLL: Duration = Duration::from_millis(250);
/// Objects between two reads of `/proc/meminfo` inside one pack. A read is a
/// few microseconds; 4096 objects of inflate+hash is milliseconds.
pub const CHECK_EVERY_OBJECTS: u64 = 4096;
/// While paused, the flush is run again this often: what a reader rebuilt in
/// the meantime (its scanned pack, the exploded oid tree) is the owner's to
/// drop again. MEASURED on the 8 GB rig 2026-08-30: a pause that flushed once
/// sat for two hours at 92 % used, holding a 2 GB table a reader built AFTER
/// the flush.
pub const REFLUSH_EVERY: Duration = Duration::from_secs(10);
/// Resume once this much MORE than the floor is available — **an absolute
/// margin, not a share of the budget**.
///
/// It was `budget / 8` and that is the livelock lane P2 measured on the rig
/// 2026-08-30. The pause opened on a **75 MB (2.2 %)** shortfall at 16 GB and
/// a **1.65 MB (0.1 %)** shortfall at 8 GB, and then demanded 12.5 % of the
/// floor — 410 MB and 200 MB — before it would leave. It never left: 1 pause
/// and 0 resumes in 31m44s and 1h45m41s, and on the 8 GB box `MemAvailable`
/// had FALLEN by the time the repository was deleted out from under it.
///
/// The margin exists to stop flapping (14 pause/resume pairs in one minute on
/// the 8 GB rig), and 64 MiB stops that just as well as an eighth of the floor
/// does, without making the exit condition scale with the box.
pub const RESUME_MARGIN_BYTES: u64 = 64 << 20;
/// **A pause has a deadline.** [`MemWatchdog::pause_until_ok`] had none, which
/// is why the two rig boxes parked until somebody deleted the repository. Past
/// this the wait ends with [`PauseEnd::Deadline`] and the caller must act —
/// proceed, shed, or fail the pack BY NAME — rather than sleep again.
pub const PAUSE_DEADLINE: Duration = Duration::from_secs(300);
/// A pause whose flush frees **nothing** this many times running gives up
/// before the deadline. `index.spilled` fired 190 and 627 times with
/// `scanned_rows_dropped=0` on every single one: the spill had nothing to give,
/// because the RAM was held by the resolver's live O(pack) structures. More
/// waiting cannot fix that, and pretending otherwise is what cost two hours.
pub const GIVE_UP_AFTER_EMPTY_FLUSHES: u32 = 3;
/// `(MemTotal, MemAvailable)` in bytes out of `/proc/meminfo` text.
pub fn parse_meminfo(text: &str) -> Option<(u64, u64)> {
let mut total = None;
let mut avail = None;
for line in text.lines() {
let Some((key, rest)) = line.split_once(':') else { continue };
let Some(kib) = rest.split_ascii_whitespace().next().and_then(|v| v.parse::<u64>().ok())
else {
continue;
};
match key {
"MemTotal" => total = Some(kib * 1024),
"MemAvailable" => avail = Some(kib * 1024),
_ => {}
}
if total.is_some() && avail.is_some() {
break;
}
}
Some((total?, avail?))
}
/// `1.5G` / `512M` / `4K` — the dashboard's spelling, so the WARN row and the
/// header agree on what a number looks like.
pub fn fmt_bytes(n: u64) -> String {
const K: u64 = 1024;
if n >= K * K * K {
format!("{:.1}G", n as f64 / (K * K * K) as f64)
} else if n >= K * K {
format!("{}M", n / (K * K))
} else if n >= K {
format!("{}K", n / K)
} else {
format!("{n}B")
}
}
/// The watchdog: a meminfo path, a floor, and the counters that say what it did.
pub struct MemWatchdog {
meminfo: PathBuf,
total: u64,
budget: u64,
poll: Duration,
/// Times a pause began.
pauses: AtomicU64,
/// Milliseconds spent paused, summed.
paused_ms: AtomicU64,
/// `true` between the WARN and the INFO.
paused_now: AtomicBool,
/// The absolute resume margin (see [`RESUME_MARGIN_BYTES`]).
resume_margin: u64,
/// The ceiling on one pause; `Duration::ZERO` disables it.
deadline: Duration,
/// How often the flush is run again while paused ([`REFLUSH_EVERY`]).
reflush: Duration,
/// Pauses that ended because they ran out of time or out of things to drop.
give_ups: AtomicU64,
}
/// **How a pause ended**, which is the fact the caller has to branch on.
///
/// Before 2026-08-30 there was only one way out — memory came back — and no way
/// to say "it did not". Lane P2's two rig boxes took the missing branch for
/// 31m44s and 1h45m41s each and the clients never got an ack.
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub enum PauseEnd {
/// `MemAvailable` climbed back over the floor plus the margin.
Recovered,
/// The cancel flag was raised (the repository is being deleted).
Cancelled,
/// [`MemWatchdog::pause_deadline`] expired.
Deadline,
/// The flush freed nothing [`GIVE_UP_AFTER_EMPTY_FLUSHES`] times running
/// and memory did not move: there is nothing left to give on this box.
NothingToGive,
}
impl PauseEnd {
pub fn as_str(self) -> &'static str {
match self {
PauseEnd::Recovered => "recovered",
PauseEnd::Cancelled => "cancelled",
PauseEnd::Deadline => "deadline",
PauseEnd::NothingToGive => "nothing-to-give",
}
}
/// Did the wait end **without** the memory coming back? Then the caller
/// must not go back to sleep: it proceeds, sheds, or fails the pack by name.
pub fn gave_up(self) -> bool {
matches!(self, PauseEnd::Deadline | PauseEnd::NothingToGive)
}
}
/// What one pause did — milliseconds, how it ended, and where memory stood.
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub struct Pause {
pub ms: u64,
pub end: PauseEnd,
pub available: u64,
pub budget: u64,
/// Flushes during this pause that freed nothing at all.
pub empty_flushes: u32,
}
impl MemWatchdog {
/// From the environment: the real `/proc/meminfo` and the knobs above.
pub fn from_env() -> Self {
let meminfo = crate::arms::read_env(crate::arms::ENV_MEMINFO)
.map(PathBuf::from)
.unwrap_or_else(|| PathBuf::from("/proc/meminfo"));
let total = std::fs::read_to_string(&meminfo)
.ok()
.and_then(|t| parse_meminfo(&t))
.map(|(t, _)| t)
.unwrap_or(0);
let budget = match crate::arms::read_env(crate::arms::ENV_MEM_BUDGET_BYTES)
.and_then(|v| v.parse::<u64>().ok())
{
Some(b) => b,
None => {
let pct = crate::arms::read_env(crate::arms::ENV_MEM_BUDGET_PCT)
.and_then(|v| v.parse::<u64>().ok())
.unwrap_or(DEFAULT_BUDGET_PCT)
.min(100);
total / 100 * pct
}
};
let margin = crate::arms::read_env(crate::arms::ENV_MEM_RESUME_MARGIN_BYTES)
.and_then(|v| v.parse::<u64>().ok())
.unwrap_or(RESUME_MARGIN_BYTES);
let deadline = crate::arms::read_env(crate::arms::ENV_MEM_PAUSE_MAX_MS)
.and_then(|v| v.parse::<u64>().ok())
.map(Duration::from_millis)
.unwrap_or(PAUSE_DEADLINE);
Self::with(&meminfo, budget, POLL).bounded(margin, deadline)
}
/// An explicit watchdog: `meminfo` is what is read, `budget` the floor in
/// bytes, `poll` the re-check cadence. What a test builds.
pub fn with(meminfo: &Path, budget: u64, poll: Duration) -> Self {
let total = std::fs::read_to_string(meminfo)
.ok()
.and_then(|t| parse_meminfo(&t))
.map(|(t, _)| t)
.unwrap_or(0);
Self {
meminfo: meminfo.to_path_buf(),
total,
budget,
poll,
pauses: AtomicU64::new(0),
paused_ms: AtomicU64::new(0),
paused_now: AtomicBool::new(false),
resume_margin: RESUME_MARGIN_BYTES,
deadline: PAUSE_DEADLINE,
reflush: REFLUSH_EVERY,
give_ups: AtomicU64::new(0),
}
}
/// The two bounds that make a pause end: the absolute resume margin and the
/// deadline. `Duration::ZERO` disables the deadline — the pre-2026-08-30
/// shape, which only a test asks for.
pub fn bounded(mut self, resume_margin: u64, deadline: Duration) -> Self {
self.resume_margin = resume_margin;
self.deadline = deadline;
self
}
/// How often the flush runs again during one pause. A test shortens it so
/// that the empty-flush arm can be asserted without 20 s of wall clock;
/// production leaves it at [`REFLUSH_EVERY`].
pub fn with_reflush(mut self, every: Duration) -> Self {
self.reflush = every;
self
}
/// A watchdog that never pauses (budget 0). For callers that opt out.
pub fn disabled() -> Self {
Self::with(Path::new("/nonexistent"), 0, POLL)
}
pub fn budget(&self) -> u64 {
self.budget
}
pub fn total(&self) -> u64 {
self.total
}
/// `MemAvailable` right now, or `None` if the file cannot be read — and
/// `None` never pauses: a box with no `/proc` is not a box short of memory.
pub fn available(&self) -> Option<u64> {
std::fs::read_to_string(&self.meminfo)
.ok()
.and_then(|t| parse_meminfo(&t))
.map(|(_, a)| a)
}
/// `Some((available, budget))` when the floor is breached.
pub fn breached(&self) -> Option<(u64, u64)> {
if self.budget == 0 {
return None;
}
let avail = self.available()?;
(avail < self.budget).then_some((avail, self.budget))
}
/// The absolute resume margin this watchdog uses — [`RESUME_MARGIN_BYTES`],
/// or `ZNIPPY_GIT_MEM_RESUME_MARGIN_BYTES`. Never more than the floor
/// itself: a margin bigger than the budget would be the old bug in a new
/// spelling.
pub fn resume_margin(&self) -> u64 {
self.resume_margin.min(self.budget.max(1))
}
/// The ceiling on one pause — [`PAUSE_DEADLINE`], or
/// `ZNIPPY_GIT_MEM_PAUSE_MAX_MS`. `0` means "no deadline", which is what a
/// test asks for when it wants the old shape on purpose.
pub fn pause_deadline(&self) -> Duration {
self.deadline
}
/// Is there room to resume — the floor plus the margin?
fn recovered(&self) -> bool {
match self.available() {
None => true,
Some(avail) => avail >= self.budget.saturating_add(self.resume_margin()),
}
}
/// How many times a pause began.
pub fn pauses(&self) -> u64 {
self.pauses.load(Ordering::Relaxed)
}
/// Milliseconds spent paused in total.
pub fn paused_ms(&self) -> u64 {
self.paused_ms.load(Ordering::Relaxed)
}
/// Is a pause in progress on some worker?
pub fn is_paused(&self) -> bool {
self.paused_now.load(Ordering::Relaxed)
}
/// Pauses that ended on the deadline or with nothing left to drop.
pub fn give_ups(&self) -> u64 {
self.give_ups.load(Ordering::Relaxed)
}
/// **The check.** Returns immediately when the floor holds. Otherwise runs
/// `flush` (the caller drops what it can, and says how much it dropped),
/// then waits — re-reading every `poll`, re-flushing every
/// [`REFLUSH_EVERY`] — until one of the four things in [`PauseEnd`]
/// happens.
///
/// `flush` returns **what it freed**: rows, bytes, entries, whatever the
/// caller counts — the number is only ever compared with zero. Three zeros
/// in a row ([`GIVE_UP_AFTER_EMPTY_FLUSHES`]) with memory no better than it
/// started means there is nothing on this box left to give, and the wait
/// ends with [`PauseEnd::NothingToGive`] instead of continuing. So does
/// [`MemWatchdog::pause_deadline`] expiring.
///
/// **A pause that cannot end is worse than a pack that is refused**, which
/// is the whole lesson of lane P2's two rig boxes: they parked for 31m44s
/// and 1h45m41s, emitted 190 and 627 spills that freed nothing, and never
/// resumed. The caller decides what to do with a give-up; this decides that
/// there IS one.
///
/// `label` names the worker in the log lines (`alice/linux`).
pub fn pause_until_ok(
&self,
label: &str,
cancel: Option<&AtomicBool>,
flush: &mut dyn FnMut() -> Result<u64>,
) -> Result<Pause> {
let Some((avail, budget)) = self.breached() else {
return Ok(Pause {
ms: 0,
end: PauseEnd::Recovered,
available: self.available().unwrap_or(0),
budget: self.budget,
empty_flushes: 0,
});
};
self.pauses.fetch_add(1, Ordering::Relaxed);
self.paused_now.store(true, Ordering::Relaxed);
let margin = self.resume_margin();
tracing::warn!(
target: "gunnar::index",
event = "index.paused",
repo = label,
available = avail,
budget = budget,
resume_at = budget.saturating_add(margin),
deadline_ms = self.deadline.as_millis() as u64,
"index paused: {} available < {} budget; resumes at {}, gives up after {} s",
fmt_bytes(avail),
fmt_bytes(budget),
fmt_bytes(budget.saturating_add(margin)),
self.deadline.as_secs()
);
let started = Instant::now();
// Drop what can be dropped BEFORE waiting: the wait is for the rest of
// the box, the flush is our own share of the problem. And AGAIN every
// `REFLUSH_EVERY` while still short — what a reader rebuilt since is
// ours to drop as well.
let mut empty_flushes = 0u32;
let mut flush_err: Option<anyhow::Error> = None;
match flush() {
Ok(0) => empty_flushes = 1,
Ok(_) => {}
Err(e) => flush_err = Some(e),
}
let mut last_flush = Instant::now();
let end = loop {
if cancel.is_some_and(|c| c.load(Ordering::Relaxed)) {
break PauseEnd::Cancelled;
}
if self.recovered() {
break PauseEnd::Recovered;
}
if !self.deadline.is_zero() && started.elapsed() >= self.deadline {
break PauseEnd::Deadline;
}
if empty_flushes >= GIVE_UP_AFTER_EMPTY_FLUSHES
&& self.available().is_some_and(|a| a <= avail)
{
break PauseEnd::NothingToGive;
}
if last_flush.elapsed() >= self.reflush {
match flush() {
Ok(0) => empty_flushes += 1,
Ok(_) => empty_flushes = 0,
Err(e) => {
if flush_err.is_none() {
flush_err = Some(e);
}
}
}
last_flush = Instant::now();
}
std::thread::sleep(self.poll);
};
let ms = started.elapsed().as_millis() as u64;
self.paused_ms.fetch_add(ms, Ordering::Relaxed);
self.paused_now.store(false, Ordering::Relaxed);
if end.gave_up() {
self.give_ups.fetch_add(1, Ordering::Relaxed);
}
let now = self.available().unwrap_or(0);
if end.gave_up() {
tracing::warn!(
target: "gunnar::index",
event = "index.gave_up",
repo = label,
available = now,
budget = budget,
paused_ms = ms,
empty_flushes = empty_flushes,
reason = end.as_str(),
"index gave up waiting for memory after {ms} ms ({}): {} available, {} needed. \
The caller decides — proceed, shed, or refuse the pack by name — but nobody \
waits here again.",
end.as_str(),
fmt_bytes(now),
fmt_bytes(budget.saturating_add(margin))
);
} else {
tracing::info!(
target: "gunnar::index",
event = "index.resumed",
repo = label,
available = now,
budget = budget,
paused_ms = ms,
reason = end.as_str(),
"index resumed after {ms} ms ({}): {} available",
end.as_str(),
fmt_bytes(now)
);
}
if let Some(e) = flush_err {
return Err(e);
}
Ok(Pause { ms, end, available: now, budget, empty_flushes })
}
}
fn global_slot() -> &'static RwLock<Arc<MemWatchdog>> {
static G: OnceLock<RwLock<Arc<MemWatchdog>>> = OnceLock::new();
G.get_or_init(|| RwLock::new(Arc::new(MemWatchdog::from_env())))
}
/// The process-wide watchdog, built from the environment on first use. One per
/// process because the floor is a property of the box, not of a repository.
/// An `Arc` so a worker mid-pack keeps the one it started with if a test swaps
/// the global underneath it.
pub fn global() -> Arc<MemWatchdog> {
global_slot().read().unwrap_or_else(|e| e.into_inner()).clone()
}
/// Replace the process-wide watchdog — what a test does to put a fake
/// `/proc/meminfo` under a real store. Returns the previous one so the test
/// can put it back.
pub fn set_global(dog: Arc<MemWatchdog>) -> Arc<MemWatchdog> {
std::mem::replace(&mut *global_slot().write().unwrap_or_else(|e| e.into_inner()), dog)
}
/// This process's peak resident set (`VmHWM` off `/proc/self/status`) in
/// bytes, or 0 when unreadable. The audit line carries it: on the appliance
/// the server is PID 1, so its high-water mark is the box's.
pub fn peak_rss_bytes() -> u64 {
std::fs::read_to_string("/proc/self/status")
.ok()
.and_then(|t| {
t.lines().find_map(|l| {
l.strip_prefix("VmHWM:")
.and_then(|r| r.split_ascii_whitespace().next())
.and_then(|v| v.parse::<u64>().ok())
.map(|kib| kib * 1024)
})
})
.unwrap_or(0)
}
/// `MemTotal` in bytes, or 0 when unreadable. What byte budgets expressed as a
/// share of the box are derived from.
pub fn mem_total() -> u64 {
global().total()
}
#[cfg(test)]
mod tests {
use super::*;
use std::sync::Arc;
use std::sync::atomic::AtomicU64;
fn fake_meminfo(dir: &Path, total_kib: u64, avail_kib: u64) -> PathBuf {
let p = dir.join("meminfo");
std::fs::write(
&p,
format!(
"MemTotal: {total_kib} kB\nMemFree: {avail_kib} kB\nMemAvailable: {avail_kib} kB\nBuffers: 0 kB\n"
),
)
.unwrap();
p
}
fn scratch(name: &str) -> PathBuf {
let d = std::env::temp_dir().join(format!("memguard-{}-{name}", std::process::id()));
let _ = std::fs::remove_dir_all(&d);
std::fs::create_dir_all(&d).unwrap();
d
}
#[test]
fn parses_the_two_fields_and_ignores_the_rest() {
let (t, a) = parse_meminfo(
"MemTotal: 16000000 kB\nMemFree: 1 kB\nMemAvailable: 3000000 kB\nSwapTotal: 0 kB\n",
)
.unwrap();
assert_eq!(t, 16_000_000 * 1024);
assert_eq!(a, 3_000_000 * 1024);
assert!(parse_meminfo("MemTotal: 1 kB\n").is_none(), "no MemAvailable = no answer");
assert!(parse_meminfo("garbage").is_none());
}
#[test]
fn above_the_floor_costs_nothing_and_flushes_nothing() {
let d = scratch("above");
let p = fake_meminfo(&d, 16_000_000, 8_000_000);
let w = MemWatchdog::with(&p, 3_200_000 * 1024, Duration::from_millis(1));
let mut flushed = 0;
let p = w.pause_until_ok("t", None, &mut || {
flushed += 1;
Ok(1)
})
.unwrap();
assert_eq!(p.ms, 0);
assert_eq!(p.end, PauseEnd::Recovered);
assert_eq!(flushed, 0);
assert_eq!(w.pauses(), 0);
assert!(w.breached().is_none());
}
/// The property the panic needs: under the floor the worker FLUSHES ONCE
/// and then WAITS until the kernel says memory is back — and it does wait,
/// which is asserted by a fake meminfo that is only raised by a second
/// thread after a measurable delay.
#[test]
fn below_the_floor_flushes_once_then_waits_for_memory_to_come_back() {
let d = scratch("below");
let p = fake_meminfo(&d, 16_000_000, 2_900_000);
let w = Arc::new(MemWatchdog::with(&p, 3_200_000 * 1024, Duration::from_millis(5)));
assert_eq!(w.breached(), Some((2_900_000 * 1024, 3_200_000 * 1024)));
let raise_at = Arc::new(AtomicU64::new(0));
let (p2, r2) = (p.clone(), raise_at.clone());
let t = std::thread::spawn(move || {
std::thread::sleep(Duration::from_millis(120));
r2.store(1, Ordering::SeqCst);
fake_meminfo(p2.parent().unwrap(), 16_000_000, 9_000_000);
});
let mut flushed = 0;
let paused = w
.pause_until_ok("t", None, &mut || {
flushed += 1;
Ok(1)
})
.unwrap();
let ms = paused.ms;
t.join().unwrap();
assert_eq!(paused.end, PauseEnd::Recovered);
assert_eq!(flushed, 1, "flush runs exactly once per pause");
assert_eq!(raise_at.load(Ordering::SeqCst), 1, "the wait outlasted the raise");
assert!(ms >= 100, "waited {ms} ms; must have waited for the raise (~120 ms)");
assert_eq!(w.pauses(), 1);
assert!(w.paused_ms() >= 100);
assert!(!w.is_paused());
}
#[test]
fn a_cancel_ends_the_wait_without_memory_coming_back() {
let d = scratch("cancel");
let p = fake_meminfo(&d, 16_000_000, 100);
let w = Arc::new(MemWatchdog::with(&p, 3_200_000 * 1024, Duration::from_millis(5)));
let cancel = Arc::new(AtomicBool::new(false));
let c2 = cancel.clone();
let t = std::thread::spawn(move || {
std::thread::sleep(Duration::from_millis(60));
c2.store(true, Ordering::SeqCst);
});
let paused = w.pause_until_ok("t", Some(&cancel), &mut || Ok(1)).unwrap();
t.join().unwrap();
assert_eq!(paused.end, PauseEnd::Cancelled);
assert!(
(50..2000).contains(&paused.ms),
"ended by the cancel, not by memory: {} ms",
paused.ms
);
}
/// While paused the flush runs again every `REFLUSH_EVERY`, and the pause
/// does not end a few bytes over the floor: the fake sits at floor + 1 %
/// (still paused, margin not met) until it is raised past the margin.
#[test]
fn a_long_pause_reflushes_and_resumes_only_past_the_margin() {
let d = scratch("reflush");
let p = fake_meminfo(&d, 16_000_000, 1_000_000);
let budget = 3_200_000 * 1024;
let w = Arc::new(MemWatchdog::with(&p, budget, Duration::from_millis(5)));
let p2 = p.clone();
let t = std::thread::spawn(move || {
std::thread::sleep(Duration::from_millis(60));
// Just over the floor: not enough.
fake_meminfo(p2.parent().unwrap(), 16_000_000, 3_232_000);
std::thread::sleep(Duration::from_millis(120));
fake_meminfo(p2.parent().unwrap(), 16_000_000, 9_000_000);
});
let mut flushes = 0;
let paused = w
.pause_until_ok("t", None, &mut || {
flushes += 1;
Ok(1)
})
.unwrap();
t.join().unwrap();
let ms = paused.ms;
assert_eq!(paused.end, PauseEnd::Recovered);
assert!(
ms >= 170,
"resumed 32 MB over the floor after {ms} ms; the margin is {}",
fmt_bytes(w.resume_margin())
);
assert!(flushes >= 1);
// REFLUSH_EVERY is 10 s in production; this pause was shorter, so
// exactly the one flush at the start — the cadence itself is a constant.
assert_eq!(flushes, 1);
}
/// **The livelock, as a test.** Lane P2's 16 GB rig paused on a 75 MB
/// shortfall — 2.2 % — and the old `budget / 8` then demanded 410 MB of
/// headroom to leave. The margin is absolute now, so the same shortfall
/// clearing to 100 MB over the floor RESUMES.
#[test]
fn a_two_percent_shortfall_does_not_demand_twelve_percent_of_headroom() {
let d = scratch("livelock");
let budget = 3_361_581_460u64; // the rig's own 16 GB floor, to the byte
// available = budget - 75 MB, which is where the rig paused.
let p = fake_meminfo(&d, 16_000_000, (budget - 75_000_000) / 1024);
let w = Arc::new(MemWatchdog::with(&p, budget, Duration::from_millis(5)));
assert!(w.breached().is_some());
assert_eq!(w.resume_margin(), RESUME_MARGIN_BYTES);
assert!(
budget / 8 > 400_000_000,
"the old margin really was 410 MB on this box: {}",
fmt_bytes(budget / 8)
);
let p2 = p.clone();
let t = std::thread::spawn(move || {
std::thread::sleep(Duration::from_millis(40));
// 100 MB over the floor: under the OLD eighth, over the new margin.
fake_meminfo(p2.parent().unwrap(), 16_000_000, (budget + 100_000_000) / 1024);
});
let paused = w.pause_until_ok("t", None, &mut || Ok(0)).unwrap();
t.join().unwrap();
assert_eq!(
paused.end,
PauseEnd::Recovered,
"100 MB over a 3.1G floor is a resume; an eighth of the floor would still be waiting"
);
assert_eq!(w.give_ups(), 0);
}
/// **A pause has a deadline.** Memory never comes back, the flush frees
/// something every time (so the nothing-to-give arm cannot fire), and the
/// wait still ends — which it never did before 2026-08-30.
#[test]
fn a_pause_that_never_recovers_ends_on_its_deadline() {
let d = scratch("deadline");
let p = fake_meminfo(&d, 16_000_000, 100);
let w = MemWatchdog::with(&p, 3_200_000 * 1024, Duration::from_millis(5))
.bounded(RESUME_MARGIN_BYTES, Duration::from_millis(120));
let started = Instant::now();
let paused = w.pause_until_ok("t", None, &mut || Ok(4096)).unwrap();
assert_eq!(paused.end, PauseEnd::Deadline);
assert!(paused.end.gave_up());
assert!(started.elapsed() >= Duration::from_millis(100));
assert!(paused.ms < 5_000, "the deadline is the bound, not a suggestion");
assert_eq!(w.give_ups(), 1);
}
/// **A spill that frees nothing is not a reason to keep waiting.** This is
/// `scanned_rows_dropped=0`, 627 times, on the 8 GB rig: the caller had
/// nothing droppable left, so the pause ends BY NAME — and long before the
/// deadline, which here is ten minutes.
#[test]
fn three_flushes_that_free_nothing_end_the_pause_by_name() {
let d = scratch("nothing");
let p = fake_meminfo(&d, 16_000_000, 100);
let w = MemWatchdog::with(&p, 3_200_000 * 1024, Duration::from_millis(1))
.bounded(RESUME_MARGIN_BYTES, Duration::from_secs(600))
.with_reflush(Duration::from_millis(5));
let mut flushes = 0u32;
let started = Instant::now();
let paused = w
.pause_until_ok("t", None, &mut || {
flushes += 1;
Ok(0)
})
.unwrap();
assert_eq!(paused.end, PauseEnd::NothingToGive);
assert!(paused.end.gave_up());
assert_eq!(paused.empty_flushes, GIVE_UP_AFTER_EMPTY_FLUSHES);
assert_eq!(flushes, GIVE_UP_AFTER_EMPTY_FLUSHES, "one at the top, then the re-flushes");
assert!(started.elapsed() < Duration::from_secs(30), "the deadline was ten minutes");
assert_eq!(w.give_ups(), 1);
}
/// ...and a flush that DOES free something keeps the pause alive, because
/// the caller is still making the box better. Same fake, same never-moving
/// memory; only the flush's answer differs, and this one ends on the clock.
#[test]
fn a_flush_that_frees_something_is_not_a_give_up_by_name() {
let d = scratch("something");
let p = fake_meminfo(&d, 16_000_000, 100);
let w = MemWatchdog::with(&p, 3_200_000 * 1024, Duration::from_millis(1))
.bounded(RESUME_MARGIN_BYTES, Duration::from_millis(200))
.with_reflush(Duration::from_millis(5));
let paused = w.pause_until_ok("t", None, &mut || Ok(1)).unwrap();
assert_eq!(paused.end, PauseEnd::Deadline);
assert_eq!(paused.empty_flushes, 0);
}
#[test]
fn a_zero_budget_never_pauses_and_an_unreadable_file_never_pauses() {
let w = MemWatchdog::disabled();
assert!(w.breached().is_none());
let d = scratch("unreadable");
let w = MemWatchdog::with(&d.join("missing"), 1 << 40, Duration::from_millis(1));
assert!(w.breached().is_none(), "no /proc = no opinion, never a stall");
}
#[test]
fn the_floor_is_a_share_of_memtotal_by_default() {
let d = scratch("pct");
let p = fake_meminfo(&d, 16_000_000, 8_000_000);
// Same arithmetic `from_env` uses, checked against the fake.
let total = parse_meminfo(&std::fs::read_to_string(&p).unwrap()).unwrap().0;
assert_eq!(total / 100 * DEFAULT_BUDGET_PCT, 3_276_800_000);
assert_eq!(fmt_bytes(3_276_800_000), "3.1G");
assert_eq!(fmt_bytes(2_900_000 * 1024), "2.8G");
assert_eq!(fmt_bytes(512 << 20), "512M");
}
}