znippy-plugin-git 0.2.2

//! **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");
    }
}