gunnar-ssh 3.0.0

//! **Space admission: the disk floor, counted against the bytes still arriving.**
//!
//! The disk-headroom gate ([`crate::headroom`]) answers ONE question before a
//! push's first byte: is the data set under its floor right now. It cannot
//! answer the question that matters when eight pushes arrive at once, because a
//! pack header carries an object count and no byte count, and every one of the
//! eight is admitted against the same measurement.
//!
//! # The defect this closes (cov-13, the UpCloud mockup, 2026-10-06)
//!
//! Eight concurrent 512 MiB pushes into gunnar main's 19.9 GiB set, 2 GiB floor.
//! Each was admitted at the door with room to spare; together they ran /data out
//! of space while still receiving, and the store failed them MID-RECEIVE as
//! `uring: op … failed: Operation canceled` — an ENOSPC that does not say so, a
//! client told "the repository is unavailable; try again", and the floor the
//! gate exists to keep simply crossed.
//!
//! # What one pushed byte costs on /data (MEASURED, `tests/push_space_admission.rs`)
//!
//! A push's bytes land three times before one of the copies is let go:
//!
//! 1. the **spool** — a pack of 64 MiB or more is spooled to
//!    `<repo>/gunnar-git.znippy/incoming-*.pack` while it arrives
//!    (`gunnar-server::storage::SPOOL_ABOVE`), so RAM never holds it;
//! 2. the **transaction log** — the pack's exact bytes are a `Pack` frame before
//!    the store sees one of them (the twin's log; raw, not compressed);
//! 3. the **store** — the archive appends the pack verbatim.
//!
//! Then the spool is deleted. So the PEAK is [`PUSH_COPIES`] = 3 × the pushed
//! bytes and the steady state 2 × until the log is trimmed (the log's byte
//! bound, `gunnar-server::retention`). The ledger reserves the peak and gives
//! the copies back as they actually land on the disk, so a byte is counted
//! once: as a promise until it is written, as `statvfs` after.
//!
//! # The rule
//!
//! One server-wide [`SpaceLedger`]: the bytes promised to transfers still
//! landing. A transfer takes a [`SpaceHold`] and asks it, chunk by chunk,
//! BEFORE the chunk is read off the wire:
//!
//! ```text
//!     admit c  ⇔  free − floor − promised − (lane, for a big push)  ≥  3.5 × c
//!
//! (3 copies, each with the filesystem's slack: XFS counts what it preallocates past a growing
//! file's end — MEASURED at start, [`FsSlack`]; never under [`PUSH_FS_SLACK_PERMILLE`].)
//! ```
//!
//! measured with one `statvfs` per ask under the ledger's lock, so two pushes
//! can never both be granted the same byte. A chunk that does not fit is
//! granted in part; a push that is granted nothing is REFUSED — that push, not
//! the others — with [`FULL_NOW`]'s line: its reservation is given back at
//! once, its spool is deleted by its own failing ingest, and nothing of it was
//! framed or stored. The refusal is final for that push and transient for the
//! data set: try again later.
//!
//! # Fair: one huge push must not starve everything
//!
//! * **The small-push lane.** A push that has sent more than
//!   [`SMALL_PUSH_BYTES`] (16 MiB) may not take the last [`SMALL_PUSH_LANE`]
//!   (256 MiB) above the floor. A big push is therefore refused while there is
//!   still room for ordinary pushes, and a 40 KiB commit keeps landing beside a
//!   512 MiB import that does not fit — the import is the one told "full".
//! * **Whoever cannot be granted is refused, and gives its room back.** N equal
//!   pushes into room for M < N of them do not all stall at a shared line: the
//!   first one short is refused and its promise (and spool) returns to the
//!   others, so the remaining ones finish. Bytes are granted in [`LEASE_BYTES`]
//!   steps, so a big push is never granted the room ahead of the push beside it.
//!
//! # What stays OUT of the ledger, on purpose
//!
//! `Admin.Capacity` (and so monetize's emergency-growth trigger, which reads
//! `used = total − servable`) is the RAW `statvfs` — what is on the disk. A
//! promise is not a byte used: a burst that is refused would otherwise read as
//! a full set for the seconds it lasted and buy a growth nobody needed. The
//! bytes that DO land move `used` as they land, which is when the trigger
//! should see them.
//!
//! What a push writes AFTER its hold is given back — the indexer's rows,
//! journal pages, a `Refs` frame — is not reserved: it is small beside the pack
//! and it is what the floor itself is kept for.

use std::path::{Path, PathBuf};
use std::sync::{Arc, Mutex, MutexGuard};

use crate::headroom::{human, DiskFloor};

const MIB: u64 = 1024 * 1024;

/// **How many bytes of /data one pushed byte occupies at its peak**: spool,
/// transaction-log frame, store. See the module header for the measurement.
pub const PUSH_COPIES: u64 = 3;

/// **The filesystem's slack on top of the copies, ‰** (r5.1, 2026-10-07): what the data set's
/// filesystem counts as used beyond the bytes written while a file grows. XFS — the appliance's
/// /data — preallocates past a growing file's end and counts it used until the file is closed
/// (or trimmed): MEASURED on XFS, one 96 MiB push peaked at 3.24× its bytes on the data set and
/// held 2.16× after it landed, against the three copies (tmpfs: under 3.15×). 1/6 on top of
/// every copy: a promise of 3.5× per wire byte, given back when the hold drops.
pub const PUSH_FS_SLACK_PERMILLE: u64 = 167;

/// **What the ledger promises per wire byte of a framed push, ‰**, at the least: the
/// [`PUSH_COPIES`] with the floor of the filesystem's slack ([`PUSH_FS_SLACK_PERMILLE`]). A
/// ledger built with a measured [`FsSlack`] promises more where the filesystem preallocates more.
pub const PUSH_RESERVE_PERMILLE: u64 = PUSH_COPIES * (1000 + PUSH_FS_SLACK_PERMILLE);

// ── the filesystem's preallocation, measured at start (round 5.2, 2026-10-07) ─────────────────
//
// MEASURED: XFS preallocates past a growing file's end by DOUBLING — a file written in 1 MiB
// steps holds 16 MiB allocated at 8 MiB, 32 at 16, 64 at 32, 128 at 64 — and counts it used until
// the file is closed: right after a power of two a growing file occupies 2× its bytes (dwarf, a
// 256 KiB stripe; oden's 1 MiB stripe the same). The pushed bytes' three copies are such growing
// files (the spool, the transaction log, the store). One push of 96 MiB peaked at 3.24× its bytes
// on dwarf and 3.58× on oden — over the fixed 3.5× this ledger used to promise. tmpfs and ext4
// preallocate nothing that `st_blocks` counts. So the slack is measured on the data set itself.

/// **The probe's spool-sized write** ([`measure_fs_slack`]): grown in 1 MiB steps past 32 MiB,
/// so a doubling filesystem shows its worst (right after a power of two) inside it.
pub const PROBE_BYTES: u64 = 33 * MIB;

/// The probe measures from here on: a push this size or bigger is a big push
/// ([`SMALL_PUSH_BYTES`]); what a smaller file holds over its bytes is small beside the floor.
pub const PROBE_FROM: u64 = SMALL_PUSH_BYTES;

/// **The margin on top of what was measured, ‰.**
pub const SLACK_MARGIN_PERMILLE: u64 = 50;

/// **The slack when the filesystem could not be measured, ‰**: XFS's doubling (1000 ‰) and the
/// margin — the most a growing file was seen to occupy. Conservative on purpose: it only refuses a
/// big push sooner near the floor.
pub const UNMEASURED_SLACK_PERMILLE: u64 = 1000 + SLACK_MARGIN_PERMILLE;

/// The probe needs this much free above nothing to run (its file and what the filesystem
/// preallocates for it, twice over); less, and the slack is [`UNMEASURED_SLACK_PERMILLE`].
pub const PROBE_NEEDS_FREE: u64 = 4 * PROBE_BYTES;

/// The probe's file names (`<root>/.gunnar-space-probe-<pid>-<n>`): one a crash left behind is
/// deleted by the next probe.
pub const PROBE_FILE_PREFIX: &str = ".gunnar-space-probe-";

/// **The filesystem's slack as this data set's ledger reserves it.**
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub struct FsSlack {
    /// What a growing spool-sized file occupied beyond its bytes, ‰ of them, at worst — `None`
    /// when it could not be measured.
    pub measured_permille: Option<u64>,
    /// What the ledger reserves on top of every copy, ‰: the measurement and
    /// [`SLACK_MARGIN_PERMILLE`], never under [`PUSH_FS_SLACK_PERMILLE`];
    /// [`UNMEASURED_SLACK_PERMILLE`] unmeasured.
    pub slack_permille: u64,
}

impl Default for FsSlack {
    /// The floor: what a ledger built without a measurement reserves (a test's, a tool's).
    fn default() -> Self {
        FsSlack {
            measured_permille: None,
            slack_permille: PUSH_FS_SLACK_PERMILLE,
        }
    }
}

impl FsSlack {
    /// The slack from a measurement (`None`: not measured).
    pub fn from_measured(measured_permille: Option<u64>) -> FsSlack {
        let slack_permille = match measured_permille {
            Some(m) => m
                .saturating_add(SLACK_MARGIN_PERMILLE)
                .max(PUSH_FS_SLACK_PERMILLE),
            None => UNMEASURED_SLACK_PERMILLE,
        };
        FsSlack {
            measured_permille,
            slack_permille,
        }
    }

    /// **Measure `root`'s filesystem** ([`measure_fs_slack`]) and say why when it cannot be.
    pub fn of(root: &Path) -> (FsSlack, Option<String>) {
        match measure_fs_slack(root) {
            Ok(m) => (FsSlack::from_measured(Some(m)), None),
            Err(e) => (FsSlack::from_measured(None), Some(e.to_string())),
        }
    }

    /// What one wire byte of a transfer of `copies` copies is promised, ‰.
    pub fn reserve_permille(&self, copies: u64) -> u64 {
        copies.max(1).saturating_mul(1000 + self.slack_permille)
    }

    /// **The start-up line, in plain words**: what the filesystem was measured to preallocate and
    /// what every push therefore reserves.
    pub fn words(&self, root: &Path, unmeasured_because: Option<&str>) -> String {
        let reserve = times(self.reserve_permille(PUSH_COPIES));
        match self.measured_permille {
            Some(m) => format!(
                "push space: the filesystem of {root} held {held} a growing file's bytes at worst while it grew (measured at start with a {probe} write); every push reserves {reserve} its bytes until it lands — its {PUSH_COPIES} copies (spool, transaction log, store), each with room for that preallocation and a {margin} ‰ margin, never under {floor}",
                root = root.display(),
                held = times(1000 + m),
                probe = human(PROBE_BYTES),
                margin = SLACK_MARGIN_PERMILLE,
                floor = times(PUSH_RESERVE_PERMILLE),
            ),
            None => format!(
                "push space: the filesystem of {root} could not be measured ({why}); every push reserves {reserve} its bytes until it lands — its {PUSH_COPIES} copies, each with room for a filesystem that doubles a growing file (XFS) and a {margin} ‰ margin",
                root = root.display(),
                why = unmeasured_because.unwrap_or("no reason given"),
                margin = SLACK_MARGIN_PERMILLE,
            ),
        }
    }
}

/// `3500` ‰ as `3.50×`.
fn times(permille: u64) -> String {
    format!("{}.{:02}×", permille / 1000, (permille % 1000) / 10)
}

/// **What `dir`'s filesystem occupies beyond a growing file's bytes, ‰ of them, at worst**: a
/// [`PROBE_BYTES`] file is written in 1 MiB steps and its `st_blocks` read after each step from
/// [`PROBE_FROM`] on and once more after an `fdatasync`, all while it is open — the way a pushed
/// pack's spool, frame and store grow. The file is deleted before this returns; one a crash left
/// is deleted by the next. Refused when less than [`PROBE_NEEDS_FREE`] is free.
pub fn measure_fs_slack(dir: &Path) -> std::io::Result<u64> {
    use std::io::Write as _;
    use std::os::unix::fs::MetadataExt as _;
    if let Ok(entries) = std::fs::read_dir(dir) {
        for e in entries.flatten() {
            if e.file_name()
                .to_string_lossy()
                .starts_with(PROBE_FILE_PREFIX)
            {
                let _ = std::fs::remove_file(e.path());
            }
        }
    }
    let (free, _) = statvfs_bytes(dir)?;
    if free < PROBE_NEEDS_FREE {
        return Err(std::io::Error::other(format!(
            "{} free, and the probe needs {}",
            human(free),
            human(PROBE_NEEDS_FREE)
        )));
    }
    static SEQ: std::sync::atomic::AtomicU64 = std::sync::atomic::AtomicU64::new(0);
    let path = dir.join(format!(
        "{PROBE_FILE_PREFIX}{}-{}",
        std::process::id(),
        SEQ.fetch_add(1, std::sync::atomic::Ordering::Relaxed)
    ));
    struct Gone(PathBuf);
    impl Drop for Gone {
        fn drop(&mut self) {
            let _ = std::fs::remove_file(&self.0);
        }
    }
    let _gone = Gone(path.clone());
    let mut file = std::fs::OpenOptions::new()
        .write(true)
        .create_new(true)
        .open(&path)?;
    let chunk: Vec<u8> = (0..MIB as usize)
        .map(|i| (i.wrapping_mul(2_654_435_761) >> 13) as u8)
        .collect();
    let over = |file: &std::fs::File, written: u64| -> std::io::Result<u64> {
        let held = file.metadata()?.blocks().saturating_mul(512);
        Ok(held.saturating_sub(written).saturating_mul(1000) / written.max(1))
    };
    let (mut written, mut worst) = (0u64, 0u64);
    while written < PROBE_BYTES {
        file.write_all(&chunk)?;
        written += MIB;
        if written >= PROBE_FROM {
            worst = worst.max(over(&file, written)?);
        }
    }
    file.sync_data()?;
    worst = worst.max(over(&file, written)?);
    Ok(worst)
}

/// A push that has sent more than this is a BIG push, held out of the
/// [`SMALL_PUSH_LANE`]. 16 MiB is an ordinary commit's pack many times over.
pub const SMALL_PUSH_BYTES: u64 = 16 * MIB;

/// The room above the floor a big push may never take, kept for small pushes:
/// at [`PUSH_COPIES`] that is five pushes of [`SMALL_PUSH_BYTES`] at once, or
/// thousands of ordinary ones.
pub const SMALL_PUSH_LANE: u64 = 256 * MIB;

/// The most wire bytes one ask grants: one `statvfs` per 4 MiB a push sends.
pub const LEASE_BYTES: u64 = 4 * MIB;

/// Less room than this above the floor and the promises — one 64 KiB chunk at
/// [`PUSH_COPIES`] — and the door says full before the first byte.
pub const MIN_ROOM: u64 = PUSH_RESERVE_PERMILLE * 64 * 1024 / 1000;

/// **The start of every refusal for room**, so a client, a kit and a test can
/// recognise it. Deliberately NOT the floor gate's phrase
/// ([`crate::headroom::AT_FLOOR`]): that one says the set is AT its floor and stays
/// so until space is reclaimed or grown — final — and tools stop on it. This one
/// says this push does not fit beside what is already arriving — try again.
pub const FULL_NOW: &str = "the data set is full right now";

/// What the ledger measures: `(free, total)` in bytes, `free` as an
/// unprivileged writer sees it (`f_bavail`).
pub trait SpaceProbe: Send + Sync + std::fmt::Debug {
    /// One measurement. An error is a refusal, never room.
    fn measure(&self) -> std::io::Result<(u64, u64)>;
}

/// The probe that ships: one `statvfs` on the data root.
#[derive(Debug, Clone)]
pub struct Statvfs(pub PathBuf);

impl SpaceProbe for Statvfs {
    fn measure(&self) -> std::io::Result<(u64, u64)> {
        statvfs_bytes(&self.0)
    }
}

/// `(f_bavail, f_blocks)` in bytes for the filesystem `root` lives on.
pub fn statvfs_bytes(root: &Path) -> std::io::Result<(u64, u64)> {
    let vfs = rustix::fs::statvfs(root).map_err(std::io::Error::from)?;
    let frsize = if vfs.f_frsize == 0 {
        vfs.f_bsize
    } else {
        vfs.f_frsize
    };
    Ok((
        vfs.f_bavail.saturating_mul(frsize),
        vfs.f_blocks.saturating_mul(frsize),
    ))
}

/// One reading of the data set and of what is promised on it.
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub struct Standing {
    /// Bytes free now (`statvfs`).
    pub free: u64,
    /// The filesystem's size.
    pub total: u64,
    /// The floor in force for that size.
    pub floor: u64,
    /// Bytes promised to transfers still landing and not on the disk yet.
    pub reserved: u64,
}

impl Standing {
    /// What may still be promised: free, less the floor, less the promises.
    pub fn room(&self) -> u64 {
        self.free
            .saturating_sub(self.floor)
            .saturating_sub(self.reserved)
    }
}

#[derive(Debug, Default)]
struct LedgerState {
    /// Σ over live holds of their `held`.
    reserved: u64,
    /// Live holds.
    holds: usize,
    /// Transfers refused for room since the process started.
    refused: u64,
}

/// **The server-wide account of bytes promised to transfers still landing.**
/// One per data root; `gunnar serve` builds it with the disk-headroom gate and
/// hands the same `Arc` to every door that writes (push, mirror, LFS).
#[derive(Debug)]
pub struct SpaceLedger {
    root: PathBuf,
    floor: DiskFloor,
    probe: Arc<dyn SpaceProbe>,
    lane: u64,
    /// The filesystem's slack every promise carries ([`FsSlack`]).
    slack: FsSlack,
    state: Mutex<LedgerState>,
}

impl SpaceLedger {
    /// The ledger for the data set `root` lives on, measured with `statvfs`.
    pub fn new(root: impl Into<PathBuf>, floor: DiskFloor) -> Self {
        let root = root.into();
        let probe = Arc::new(Statvfs(root.clone()));
        Self::with_probe(root, floor, probe)
    }

    /// The same, measured by `probe` — a test's fake disk.
    pub fn with_probe(
        root: impl Into<PathBuf>,
        floor: DiskFloor,
        probe: Arc<dyn SpaceProbe>,
    ) -> Self {
        SpaceLedger {
            root: root.into(),
            floor,
            probe,
            lane: SMALL_PUSH_LANE,
            slack: FsSlack::default(),
            state: Mutex::new(LedgerState::default()),
        }
    }

    /// **Promise the filesystem's slack as measured** ([`FsSlack::of`]) instead of the floor
    /// ([`PUSH_FS_SLACK_PERMILLE`]). `gunnar serve` measures its data set once at start.
    pub fn with_fs_slack(mut self, slack: FsSlack) -> Self {
        self.slack = slack;
        self
    }

    /// The filesystem's slack in force.
    pub fn fs_slack(&self) -> FsSlack {
        self.slack
    }

    /// Keep `bytes` above the floor for small pushes instead of
    /// [`SMALL_PUSH_LANE`]. `0` turns the lane off.
    pub fn with_small_lane(mut self, bytes: u64) -> Self {
        self.lane = bytes;
        self
    }

    /// The data root this ledger speaks for (named in its refusals).
    pub fn root(&self) -> &Path {
        &self.root
    }

    /// The floor it keeps.
    pub fn floor(&self) -> DiskFloor {
        self.floor
    }

    /// The small-push lane in force.
    pub fn small_lane(&self) -> u64 {
        self.lane
    }

    /// The probe it measures with — the disk gate's `headroom` reads the same.
    pub fn probe(&self) -> &Arc<dyn SpaceProbe> {
        &self.probe
    }

    fn lock(&self) -> MutexGuard<'_, LedgerState> {
        self.state
            .lock()
            .unwrap_or_else(|poisoned| poisoned.into_inner())
    }

    /// Bytes promised right now and not on the disk yet.
    pub fn reserved(&self) -> u64 {
        self.lock().reserved
    }

    /// Transfers holding a promise right now.
    pub fn holds(&self) -> usize {
        self.lock().holds
    }

    /// Transfers refused for room since this ledger was built.
    pub fn refused(&self) -> u64 {
        self.lock().refused
    }

    /// One measurement, with the promises beside it.
    pub fn standing(&self) -> std::io::Result<Standing> {
        let reserved = self.reserved();
        self.measure_with(reserved)
    }

    fn measure_with(&self, reserved: u64) -> std::io::Result<Standing> {
        let (free, total) = self.probe.measure()?;
        Ok(Standing {
            free,
            total,
            floor: self.floor.bytes_for(total),
            reserved,
        })
    }

    /// **A promise for one transfer**, held to `copies` bytes of /data per byte
    /// it sends ([`PUSH_COPIES`] for a framed push). Everything it still holds
    /// is given back when it drops — landed, refused, cut or crashed.
    pub fn hold(self: &Arc<Self>, copies: u64) -> SpaceHold {
        self.lock().holds += 1;
        SpaceHold {
            ledger: Arc::clone(self),
            copies: copies.max(1),
            state: Mutex::new(HoldState::default()),
        }
    }

    /// **The up-front line, before any byte**: `Some` when what is promised
    /// already covers everything above the floor. The floor gate's own line
    /// (under the floor with nothing promised) is the gate's, not this.
    ///
    /// The pusher is told in plain words; the numbers go to the journal
    /// (wiring tracer B 2026-10-08: a customer never sees room left).
    pub fn full_before_the_first_byte(&self) -> Option<String> {
        let standing = self.standing().ok()?;
        if standing.free <= standing.floor || standing.reserved == 0 || standing.room() >= MIN_ROOM
        {
            return None;
        }
        tracing::warn!(
            target: "gunnar::space",
            event = "push.space.full_up_front",
            root = %self.root.display(),
            free = %human(standing.free),
            floor = %human(standing.floor),
            disk_floor = %self.floor,
            promised = %human(standing.reserved),
            "a push was refused before its first byte: everything above the disk floor is promised to pushes already arriving; reclaim space (`gunnar control repo delete --purge`, `gunnar gc`) or grow the storage"
        );
        Some(format!(
            "{FULL_NOW}: everything free is promised to pushes already arriving, so nothing was \
             received. Try again in a while; the storage is grown when it fills"
        ))
    }

    /// The refusal for a transfer that cannot be granted its next chunk: the
    /// pusher's plain line; what was measured goes to the journal.
    fn full_line(
        &self,
        standing: &Standing,
        own_held: u64,
        sent: u64,
        copies: u64,
        lane: u64,
    ) -> String {
        let others = standing.reserved.saturating_sub(own_held);
        let prealloc = match self.slack.measured_permille {
            Some(m) => format!(
                "measured at start to hold {} a growing file's bytes",
                times(1000 + m)
            ),
            None => "not measured, so a doubling is assumed".to_owned(),
        };
        tracing::warn!(
            target: "gunnar::space",
            event = "push.space.refused",
            root = %self.root.display(),
            free = %human(standing.free),
            floor = %human(standing.floor),
            disk_floor = %self.floor,
            small_lane = %human(lane),
            promised_to_others = %human(others),
            sent = %human(sent),
            copies,
            reserve = %times(self.slack.reserve_permille(copies)),
            filesystem = %prealloc,
            "a push did not fit beside the others while it landed: nothing of it was kept; reclaim space (`gunnar control repo delete --purge`, `gunnar gc`) or grow the storage"
        );
        format!(
            "{FULL_NOW}: this push does not fit beside the pushes already arriving, so nothing of it \
             was kept. Try again in a while; the storage is grown when it fills"
        )
    }

    fn unmeasured_line(&self, err: &std::io::Error) -> String {
        tracing::error!(
            target: "gunnar::space",
            event = "push.space.unmeasured",
            root = %self.root.display(),
            %err,
            "the data set could not be measured: a push was refused rather than guessing there is room"
        );
        format!(
            "{FULL_NOW}: Gunnar could not check its storage, so this push was refused rather than \
             risk it. Nothing of it was kept; try again in a while"
        )
    }
}

#[derive(Debug, Default)]
struct HoldState {
    /// Bytes of /data promised to this transfer and not on the disk yet.
    held: u64,
    /// Wire bytes granted so far.
    admitted: u64,
    /// Set when it was refused; every later ask is the same refusal.
    refusal: Option<String>,
}

/// **One transfer's promise.** Asked before each chunk ([`Self::admit`]),
/// told as its copies land ([`Self::realize`]), given back when it drops.
#[derive(Debug)]
pub struct SpaceHold {
    ledger: Arc<SpaceLedger>,
    copies: u64,
    state: Mutex<HoldState>,
}

impl SpaceHold {
    fn lock(&self) -> MutexGuard<'_, HoldState> {
        self.state
            .lock()
            .unwrap_or_else(|poisoned| poisoned.into_inner())
    }

    /// **May this transfer take up to `want` more wire bytes?** `Ok(n)` with
    /// `0 < n ≤ want` grants `n` (promising `copies × n` of /data); `Err(line)`
    /// refuses the transfer — for good: every later ask is the same `Err`, and
    /// everything it held is given back now, before its spool is even gone.
    pub fn admit(&self, want: u64) -> Result<u64, String> {
        let mut hold = self.lock();
        if let Some(line) = &hold.refusal {
            return Err(line.clone());
        }
        if want == 0 {
            return Ok(0);
        }
        let ledger = &self.ledger;
        let mut state = ledger.lock();
        let standing = match ledger.measure_with(state.reserved) {
            Ok(standing) => standing,
            Err(err) => {
                let line = ledger.unmeasured_line(&err);
                Self::refuse(&mut hold, &mut state, line.clone());
                return Err(line);
            }
        };
        let big = hold.admitted.saturating_add(want) > SMALL_PUSH_BYTES;
        let lane = if big { ledger.lane } else { 0 };
        let granted =
            want.min(standing.room().saturating_sub(lane).saturating_mul(1000) / self.per_mille());
        if granted == 0 {
            let line = ledger.full_line(&standing, hold.held, hold.admitted, self.copies, lane);
            Self::refuse(&mut hold, &mut state, line.clone());
            return Err(line);
        }
        let promise = self.promise_for(granted);
        state.reserved = state.reserved.saturating_add(promise);
        hold.held = hold.held.saturating_add(promise);
        hold.admitted = hold.admitted.saturating_add(granted);
        Ok(granted)
    }

    /// **All of `bytes`, or a refusal** — a transfer that declares its size
    /// up front (an LFS object). Partial room is no room here.
    pub fn admit_all(&self, bytes: u64) -> Result<(), String> {
        let mut left = bytes;
        while left > 0 {
            // `admit` grants what fits; asking again for the rest either gets
            // it (nothing changed) or refuses (and gives everything back).
            let granted = self.admit(left)?;
            if granted < left {
                let mut hold = self.lock();
                let mut state = self.ledger.lock();
                let standing = self
                    .ledger
                    .measure_with(state.reserved)
                    .unwrap_or(Standing {
                        free: 0,
                        total: 0,
                        floor: 0,
                        reserved: state.reserved,
                    });
                let lane = if bytes > SMALL_PUSH_BYTES {
                    self.ledger.lane
                } else {
                    0
                };
                let line = self
                    .ledger
                    .full_line(&standing, hold.held, bytes, self.copies, lane);
                Self::refuse(&mut hold, &mut state, line.clone());
                return Err(line);
            }
            left -= granted;
        }
        Ok(())
    }

    fn refuse(hold: &mut HoldState, state: &mut LedgerState, line: String) {
        state.reserved = state.reserved.saturating_sub(hold.held);
        state.refused += 1;
        hold.held = 0;
        hold.refusal = Some(line);
    }

    /// **`bytes` of what this transfer promised are on the disk now**, where
    /// `statvfs` counts them — so they stop being counted twice. Call it after
    /// the write returns, never before: a promise released early is a byte
    /// nobody counts.
    pub fn realize(&self, bytes: u64) {
        let mut hold = self.lock();
        let given = bytes.min(hold.held);
        if given == 0 {
            return;
        }
        hold.held -= given;
        let mut state = self.ledger.lock();
        state.reserved = state.reserved.saturating_sub(given);
    }

    /// Bytes of /data this transfer still has promised.
    pub fn held(&self) -> u64 {
        self.lock().held
    }

    /// Wire bytes granted so far.
    pub fn admitted(&self) -> u64 {
        self.lock().admitted
    }

    /// The refusal, once refused.
    pub fn refusal(&self) -> Option<String> {
        self.lock().refusal.clone()
    }

    /// The copies of /data per wire byte this hold promises.
    pub fn copies(&self) -> u64 {
        self.copies
    }

    /// What this hold promises per wire byte, ‰: its copies with the filesystem's slack
    /// ([`SpaceLedger::fs_slack`]; the floor [`PUSH_FS_SLACK_PERMILLE`] unless measured).
    pub fn per_mille(&self) -> u64 {
        self.ledger.slack.reserve_permille(self.copies)
    }

    /// The bytes of /data `wire` wire bytes are promised: [`Self::per_mille`], rounded up. What
    /// lands is given back by [`Self::realize`]; the slack is given back when the hold drops.
    pub fn promise_for(&self, wire: u64) -> u64 {
        wire.saturating_mul(self.per_mille()).div_ceil(1000)
    }

    /// **The line for a store that ran out of room anyway** — an ENOSPC the
    /// admission did not foresee (a writer outside the ledger). Said in the
    /// same words, with the numbers as they are now, so a client is never told
    /// `uring: op failed: Operation canceled`.
    pub fn full_now(&self) -> String {
        let mut hold = self.lock();
        if let Some(line) = &hold.refusal {
            return line.clone();
        }
        let mut state = self.ledger.lock();
        let line = match self.ledger.measure_with(state.reserved) {
            Ok(standing) => {
                self.ledger
                    .full_line(&standing, hold.held, hold.admitted, self.copies, 0)
            }
            Err(err) => self.ledger.unmeasured_line(&err),
        };
        Self::refuse(&mut hold, &mut state, line.clone());
        line
    }

    /// Was the data set at its floor when this transfer failed — free now,
    /// less the `freed` bytes its own failure just gave back (its spool), at
    /// or under the floor? An unmeasurable disk counts as yes. What a store
    /// failure that does not SAY "no space" (io_uring's `Operation canceled`
    /// for a linked chain whose head hit ENOSPC) is judged by.
    pub fn at_the_floor(&self, freed: u64) -> bool {
        match self.ledger.probe.measure() {
            Ok((free, total)) => {
                free.saturating_sub(freed)
                    <= self.ledger.floor.bytes_for(total).saturating_add(MIN_ROOM)
            }
            Err(_) => true,
        }
    }
}

impl Drop for SpaceHold {
    fn drop(&mut self) {
        let held = std::mem::take(&mut self.lock().held);
        let mut state = self.ledger.lock();
        state.reserved = state.reserved.saturating_sub(held);
        state.holds = state.holds.saturating_sub(1);
    }
}

/// **Is this error the disk being full?** `ENOSPC`, `EDQUOT`, or the
/// `StorageFull`/`QuotaExceeded` kinds std maps them to — found anywhere in
/// the chain an `anyhow` or `io::Error` carries. `Operation canceled` (125) is
/// how io_uring reports the tail of a linked chain whose head hit ENOSPC, so it
/// counts only beside a disk that measures at its floor ([`SpaceHold::at_the_floor`]).
pub fn is_out_of_space(err: &std::io::Error) -> bool {
    use std::io::ErrorKind;
    matches!(
        err.kind(),
        ErrorKind::StorageFull | ErrorKind::QuotaExceeded
    ) || matches!(err.raw_os_error(), Some(28) | Some(122))
}

#[cfg(test)]
mod tests {
    use super::*;
    use std::sync::atomic::{AtomicI64, AtomicU64, Ordering};

    const GIB: u64 = 1024 * MIB;

    /// **A disk that is only numbers**: free goes down when a test writes and
    /// up when it deletes, and the lowest it ever read is kept — the floor
    /// assertion is about that, not about a sample.
    #[derive(Debug)]
    struct FakeDisk {
        free: AtomicI64,
        total: u64,
        lowest: AtomicI64,
        fail: std::sync::atomic::AtomicBool,
    }

    impl FakeDisk {
        fn new(total: u64, free: u64) -> Arc<FakeDisk> {
            Arc::new(FakeDisk {
                free: AtomicI64::new(free as i64),
                total,
                lowest: AtomicI64::new(free as i64),
                fail: Default::default(),
            })
        }
        fn write(&self, bytes: u64) {
            let now = self.free.fetch_sub(bytes as i64, Ordering::SeqCst) - bytes as i64;
            self.lowest.fetch_min(now, Ordering::SeqCst);
        }
        fn delete(&self, bytes: u64) {
            self.free.fetch_add(bytes as i64, Ordering::SeqCst);
        }
        fn lowest(&self) -> i64 {
            self.lowest.load(Ordering::SeqCst)
        }
    }

    impl SpaceProbe for FakeDisk {
        fn measure(&self) -> std::io::Result<(u64, u64)> {
            if self.fail.load(Ordering::SeqCst) {
                return Err(std::io::Error::other("the mount went away"));
            }
            Ok((self.free.load(Ordering::SeqCst).max(0) as u64, self.total))
        }
    }

    /// ★ **The slack is what the filesystem preallocates, plus a margin, never under a sixth**
    /// (round 5.2, 2026-10-07): measured nothing (tmpfs, ext4) is the floor, 3.5×; XFS's
    /// doubling (1000 ‰) is 6.15×; not measured is the doubling assumed. The ledger promises it,
    /// says it at start in plain words, and a big push refused near the floor is told that room is
    /// reserved for the filesystem's preallocation.
    #[test]
    fn the_slack_is_the_measured_preallocation_with_a_margin_and_never_under_a_sixth() {
        assert_eq!(
            FsSlack::from_measured(Some(0)).slack_permille,
            PUSH_FS_SLACK_PERMILLE
        );
        assert_eq!(
            FsSlack::from_measured(Some(0)).reserve_permille(PUSH_COPIES),
            PUSH_RESERVE_PERMILLE
        );
        assert_eq!(FsSlack::from_measured(Some(1000)).slack_permille, 1050);
        assert_eq!(
            FsSlack::from_measured(Some(1000)).reserve_permille(PUSH_COPIES),
            6150
        );
        assert_eq!(
            FsSlack::from_measured(None).slack_permille,
            UNMEASURED_SLACK_PERMILLE
        );
        assert_eq!(
            FsSlack::default().reserve_permille(PUSH_COPIES),
            PUSH_RESERVE_PERMILLE,
            "a ledger nobody measured promises the floor"
        );
        let said = FsSlack::from_measured(Some(1000)).words(Path::new("/data"), None);
        assert!(
            said.contains("held 2.00× a growing file's bytes")
                && said.contains("every push reserves 6.15× its bytes"),
            "{said}"
        );
        assert!(!said.contains("  "), "one line, single-spaced: {said}");
        let said = FsSlack::from_measured(None).words(Path::new("/data"), Some("12 MiB free"));
        assert!(
            said.contains("could not be measured (12 MiB free)") && said.contains("6.15×"),
            "{said}"
        );

        let disk = FakeDisk::new(10 * GIB, 2 * GIB);
        let ledger = Arc::new(
            SpaceLedger::with_probe(
                "/data",
                DiskFloor {
                    percent: 0,
                    bytes: GIB,
                },
                Arc::clone(&disk) as Arc<dyn SpaceProbe>,
            )
            .with_small_lane(0)
            .with_fs_slack(FsSlack::from_measured(Some(1000))),
        );
        let hold = ledger.hold(PUSH_COPIES);
        assert_eq!(
            hold.promise_for(100 * MIB),
            615 * MIB,
            "6.15× promised per wire byte"
        );
        let granted = hold.admit(GIB).expect("some of it fits");
        assert_eq!(
            granted,
            (GIB * 1000) / 6150,
            "a measured slack grants less near the floor"
        );
        let line = hold.admit(GIB).unwrap_err();
        assert!(
            line.starts_with(FULL_NOW) && line.contains("does not fit"),
            "the pusher's line is plain; the 2.00x slack is the journal's: {line}"
        );
    }

    /// ★ **The pusher is told the storage is full, in plain words — no
    /// number, no path, no flag, no internal word** (wiring tracer B
    /// 2026-10-08; owner: a customer never sees caps or room left). What was
    /// free, the floor, the promises and the measured slack are the
    /// operator's: they go to the journal, not to `git push`'s stderr.
    #[test]
    fn the_pusher_is_told_full_in_plain_words_and_the_numbers_go_to_the_journal() {
        let plain = |line: &str| {
            assert!(line.starts_with(FULL_NOW), "{line}");
            assert!(
                line.contains("try again") || line.contains("Try again"),
                "{line}"
            );
            for internal in [
                "/data",
                "--disk-floor",
                "transaction log",
                "spool",
                "preallocation",
                "gunnar gc",
                "repo delete",
                "×",
                "MiB",
                "GiB",
                "promised to pushes already arriving, and this push",
            ] {
                assert!(
                    !line.contains(internal),
                    "{internal:?} reaches the pusher: {line}"
                );
            }
            assert!(
                !line.chars().any(|c| c.is_ascii_digit()),
                "a number reaches the pusher: {line}"
            );
        };
        let disk = FakeDisk::new(10 * GIB, 2 * GIB);
        let ledger = Arc::new(
            SpaceLedger::with_probe(
                "/data",
                DiskFloor {
                    percent: 0,
                    bytes: GIB,
                },
                Arc::clone(&disk) as Arc<dyn SpaceProbe>,
            )
            .with_small_lane(0)
            .with_fs_slack(FsSlack::from_measured(Some(1000))),
        );
        let hold = ledger.hold(PUSH_COPIES);
        hold.admit(GIB).expect("some of it fits");
        plain(&hold.admit(GIB).unwrap_err());
        let second = ledger.hold(PUSH_COPIES);
        let up_front = ledger.full_before_the_first_byte();
        drop(second);
        if let Some(line) = up_front {
            plain(&line);
        }
        plain(&ledger.unmeasured_line(&std::io::Error::other("the mount went away")));
    }

    /// ★ **The probe sees what the filesystem holds for a growing file — XFS's doubling where
    /// there is XFS — and leaves nothing behind** (round 5.2). Where the test's temp dir is XFS
    /// (dwarf, oden), the worst is the doubling; elsewhere, whatever it is, it is measured.
    #[test]
    fn the_probe_sees_the_preallocation_of_a_growing_file_and_leaves_nothing_behind() {
        let dir =
            std::env::temp_dir().join(format!("gunnar-space-probe-test-{}", std::process::id()));
        let _ = std::fs::remove_dir_all(&dir);
        std::fs::create_dir_all(&dir).unwrap();
        std::fs::write(
            dir.join(format!("{PROBE_FILE_PREFIX}1-7")),
            b"a crash left me",
        )
        .unwrap();
        let measured = measure_fs_slack(&dir).expect("measured");
        let left: Vec<_> = std::fs::read_dir(&dir)
            .unwrap()
            .flatten()
            .map(|e| e.file_name())
            .collect();
        assert!(
            left.is_empty(),
            "the probe leaves nothing, and takes what a crash left: {left:?}"
        );
        let xfs = rustix::fs::statfs(&dir)
            .map(|s| s.f_type as u64 == 0x5846_5342)
            .unwrap_or(false);
        eprintln!(
            "MEASURED probe on {} ({}): a growing file held {} its bytes at worst",
            dir.display(),
            if xfs { "xfs" } else { "not xfs" },
            times(1000 + measured)
        );
        if xfs {
            assert!(
                measured >= 900,
                "XFS doubles a growing file's allocation; the probe saw only {} more",
                times(measured)
            );
        }
        let _ = std::fs::remove_dir_all(&dir);
    }

    fn ledger(disk: &Arc<FakeDisk>, floor: u64, lane: u64) -> Arc<SpaceLedger> {
        Arc::new(
            SpaceLedger::with_probe(
                "/data",
                DiskFloor {
                    percent: 0,
                    bytes: floor,
                },
                Arc::clone(disk) as Arc<dyn SpaceProbe>,
            )
            .with_small_lane(lane),
        )
    }

    /// One push the way the receive path drives it: ask a lease, spool what
    /// was granted (on the disk at once), and once the pack is whole, frame
    /// it and store it (two more copies), then drop the spool. `Err` is the
    /// refusal, after the push's spool was deleted as its ingest would.
    fn push(
        disk: &FakeDisk,
        ledger: &Arc<SpaceLedger>,
        bytes: u64,
        chunk: u64,
    ) -> Result<(), String> {
        let hold = ledger.hold(PUSH_COPIES);
        let mut spooled = 0;
        while spooled < bytes {
            match hold.admit(chunk.min(bytes - spooled)) {
                Ok(granted) => {
                    disk.write(granted);
                    hold.realize(granted);
                    spooled += granted;
                }
                Err(line) => {
                    disk.delete(spooled);
                    return Err(line);
                }
            }
        }
        disk.write(bytes); // the transaction log's frame
        hold.realize(bytes);
        disk.write(bytes); // the store
        hold.realize(bytes);
        disk.delete(spooled); // the spool, dropped
        drop(hold);
        Ok(())
    }

    /// ★ cov-13: a push that would cross the floor is refused BY NAME, the line
    /// says what is free, what the floor is and what this push had sent, and
    /// the refused push gives back everything it promised.
    #[test]
    fn a_push_that_would_cross_the_floor_is_refused_by_name_and_gives_its_promise_back() {
        let disk = FakeDisk::new(10 * GIB, 3 * GIB);
        let ledger = ledger(&disk, 2 * GIB, 0);
        let hold = ledger.hold(PUSH_COPIES);
        assert_eq!(
            hold.admit(256 * MIB),
            Ok(256 * MIB),
            "896 MiB (3× + the filesystem's slack) of 1 GiB room fits"
        );
        assert_eq!(ledger.reserved(), hold.promise_for(256 * MIB));
        assert_eq!(
            hold.promise_for(256 * MIB),
            (256 * MIB * PUSH_RESERVE_PERMILLE).div_ceil(1000)
        );
        let partial = hold
            .admit(256 * MIB)
            .expect("what is left is granted in part");
        assert!(partial < 256 * MIB && partial > 0, "{partial}");
        let line = hold.admit(MIB).expect_err("no room left: refused");
        assert!(line.starts_with(FULL_NOW), "{line}");
        // The pusher's words are plain (wiring tracer B 2026-10-08); what was free, the floor
        // and what was sent are the journal's (`push.space.refused`).
        for part in ["Try again", "nothing of it was kept"] {
            assert!(line.contains(part), "the line must say {part:?}: {line}");
        }
        for internal in ["/data", "--disk-floor"] {
            assert!(
                !line.contains(internal),
                "the pusher is not told {internal:?}: {line}"
            );
        }
        assert!(
            !line.starts_with(crate::headroom::AT_FLOOR),
            "the transient refusal must not read as the floor gate's FINAL one (kits stop on that): {line}"
        );
        assert_eq!(ledger.reserved(), 0, "a refused push holds nothing");
        assert_eq!(
            hold.admit(1),
            Err(line.clone()),
            "refused is refused for good"
        );
        assert_eq!(ledger.refused(), 1);
    }

    /// ★ cov-13's eight pushes on a fake disk, measured in sequence: the room
    /// is SHARED, so pushes that each fit alone cannot all be granted it, the
    /// first short of room is refused and its room goes back, and the others
    /// land — never all eight stalled at one line.
    #[test]
    fn eight_pushes_into_room_for_three_land_three_or_more_and_refuse_the_rest() {
        // 4 GiB above the floor; 8 × 384 MiB at 3.5× (3 copies + the filesystem's slack) is
        // 10.5 GiB of peak.
        let disk = FakeDisk::new(20 * GIB, 6 * GIB);
        let ledger = ledger(&disk, 2 * GIB, 0);
        let holds: Vec<_> = (0..8).map(|_| ledger.hold(PUSH_COPIES)).collect();
        let size = 384 * MIB;
        let mut spooled = [0u64; 8];
        let mut done = [false; 8];
        let mut refused = [false; 8];
        // Round-robin, 4 MiB at a time: eight pushes arriving at one speed.
        for _ in 0..100_000 {
            let mut moved = false;
            for i in 0..8 {
                if done[i] || refused[i] {
                    continue;
                }
                moved = true;
                if spooled[i] == size {
                    for _ in 0..2 {
                        disk.write(size);
                        holds[i].realize(size);
                    }
                    disk.delete(size);
                    holds[i].realize(u64::MAX);
                    done[i] = true;
                    continue;
                }
                match holds[i].admit(LEASE_BYTES.min(size - spooled[i])) {
                    Ok(g) => {
                        disk.write(g);
                        holds[i].realize(g);
                        spooled[i] += g;
                    }
                    Err(line) => {
                        assert!(line.starts_with(FULL_NOW), "{line}");
                        disk.delete(spooled[i]);
                        refused[i] = true;
                    }
                }
                assert!(
                    disk.free.load(Ordering::SeqCst) - ledger.reserved() as i64 >= (2 * GIB) as i64,
                    "promised + written crossed the floor"
                );
            }
            if !moved {
                break;
            }
        }
        let landed = done.iter().filter(|d| **d).count();
        assert!(
            landed >= 3,
            "room for {} at the peak, {} landed (refused {refused:?})",
            4 * GIB / holds[0].promise_for(size),
            landed
        );
        assert!(
            refused.iter().any(|r| *r),
            "8 × 1.3 GiB cannot fit in 4 GiB: something must be refused"
        );
        assert!(
            disk.lowest() >= (2 * GIB) as i64,
            "the floor was crossed: lowest free {}",
            disk.lowest()
        );
        drop(holds);
        assert_eq!((ledger.reserved(), ledger.holds()), (0, 0));
    }

    /// ★ The floor is never crossed by REAL concurrency: sixteen threads, each
    /// pushing 160 MiB at its own pace into room for a handful, every write
    /// on the fake disk the moment it is granted.
    #[test]
    fn concurrent_pushes_never_take_the_disk_under_the_floor() {
        let disk = FakeDisk::new(20 * GIB, 4 * GIB);
        let ledger = ledger(&disk, 2 * GIB, 64 * MIB);
        let landed = Arc::new(AtomicU64::new(0));
        let refused = Arc::new(AtomicU64::new(0));
        std::thread::scope(|s| {
            for i in 0..16u64 {
                let (disk, ledger, landed, refused) = (&disk, &ledger, &landed, &refused);
                s.spawn(move || {
                    match push(disk, ledger, 160 * MIB, MIB * (1 + i % 3)) {
                        Ok(()) => landed.fetch_add(1, Ordering::SeqCst),
                        Err(_) => refused.fetch_add(1, Ordering::SeqCst),
                    };
                });
            }
        });
        let (landed, refused) = (
            landed.load(Ordering::SeqCst),
            refused.load(Ordering::SeqCst),
        );
        assert_eq!(landed + refused, 16);
        assert!(
            landed >= 1 && refused >= 1,
            "landed {landed}, refused {refused}"
        );
        assert!(
            disk.lowest() >= (2 * GIB) as i64,
            "the floor was crossed: lowest free {} B",
            disk.lowest()
        );
        assert_eq!(
            (ledger.reserved(), ledger.holds()),
            (0, 0),
            "every promise came back"
        );
    }

    /// ★ Fair: a big push is cut at the small-push lane, and a small push
    /// beside it lands in the room the lane kept.
    #[test]
    fn a_small_push_lands_beside_a_big_one_cut_at_the_lane() {
        let disk = FakeDisk::new(10 * GIB, 3 * GIB);
        let ledger = ledger(&disk, 2 * GIB, 256 * MIB);
        let big = ledger.hold(PUSH_COPIES);
        let mut sent = 0;
        let mut cut = None;
        for _ in 0..10_000 {
            match big.admit(LEASE_BYTES) {
                Ok(g) => {
                    disk.write(g);
                    big.realize(g);
                    sent += g;
                }
                Err(line) => {
                    cut = Some(line);
                    break;
                }
            }
        }
        let line = cut.expect("a big push into 1 GiB of room is cut");
        assert!(
            line.starts_with(FULL_NOW) && line.contains("does not fit"),
            "{line}"
        );
        let room = disk.free.load(Ordering::SeqCst) as u64 - 2 * GIB - ledger.reserved();
        assert!(
            room >= 250 * MIB,
            "the big push left the lane: {} free above the floor after {} sent",
            human(room),
            human(sent)
        );
        let small = ledger.hold(PUSH_COPIES);
        let mut got = 0;
        for _ in 0..1_000 {
            if got == 8 * MIB {
                break;
            }
            got += small
                .admit(8 * MIB - got)
                .expect("a small push lands in the lane");
        }
        assert_eq!(got, 8 * MIB);
    }

    /// A cancelled transfer — the client hung up, the ingest failed — gives
    /// back its whole promise when its hold drops.
    #[test]
    fn a_dropped_hold_gives_back_what_it_promised() {
        let disk = FakeDisk::new(10 * GIB, 5 * GIB);
        let ledger = ledger(&disk, 2 * GIB, 0);
        let hold = ledger.hold(PUSH_COPIES);
        hold.admit(100 * MIB).unwrap();
        hold.realize(40 * MIB);
        assert_eq!(
            (ledger.reserved(), ledger.holds()),
            (hold.promise_for(100 * MIB) - 40 * MIB, 1)
        );
        drop(hold);
        assert_eq!((ledger.reserved(), ledger.holds()), (0, 0));
    }

    /// The up-front line counts the promises: a door that would let a push
    /// in when everything above the floor is already promised says so before
    /// the first byte.
    #[test]
    fn the_up_front_check_says_full_when_the_room_is_all_promised() {
        let disk = FakeDisk::new(10 * GIB, 3 * GIB);
        let ledger = ledger(&disk, 2 * GIB, 0);
        assert_eq!(ledger.full_before_the_first_byte(), None);
        let hold = ledger.hold(PUSH_COPIES);
        hold.admit(250 * MIB).unwrap();
        assert_eq!(
            ledger.full_before_the_first_byte(),
            None,
            "~149 MiB of room left"
        );
        hold.admit(LEASE_BYTES).unwrap();
        let rest = hold.admit(GIB).unwrap();
        assert!(rest > 0);
        let line = ledger
            .full_before_the_first_byte()
            .expect("all of it promised");
        assert!(
            line.starts_with(FULL_NOW) && line.contains("promised to pushes already arriving"),
            "{line}"
        );
        drop(hold);
        assert_eq!(ledger.full_before_the_first_byte(), None);
    }

    /// The door asks the same ledger: under the floor it is the floor's own
    /// (final) line, above it with the room all promised it is [`FULL_NOW`]'s,
    /// and the pushes' promises given back open it again.
    #[test]
    fn the_disk_gate_refuses_up_front_when_the_room_is_all_promised() {
        use crate::plan::{PlanGate, PlanVerdict};
        let disk = FakeDisk::new(10 * GIB, 3 * GIB);
        let ledger = ledger(&disk, 2 * GIB, 0);
        let gate = crate::headroom::DiskHeadroomGate::over(Arc::clone(&ledger));
        assert_eq!(gate.pack_quota("team"), PlanVerdict::Allow);
        let hold = ledger.hold(PUSH_COPIES);
        // Bounded, so a ledger that never says no fails here rather than spins.
        for _ in 0..10_000 {
            if !hold.admit(LEASE_BYTES).is_ok_and(|g| g == LEASE_BYTES) {
                break;
            }
        }
        match gate.pack_quota("team") {
            PlanVerdict::Refuse(line) => assert!(line.starts_with(FULL_NOW), "{line}"),
            other => panic!("everything above the floor is promised: {other:?}"),
        }
        assert_eq!(
            gate.headroom().unwrap().free,
            3 * GIB,
            "Admin.Capacity reads the disk, not the promises"
        );
        drop(hold);
        assert_eq!(
            gate.pack_quota("team"),
            PlanVerdict::Allow,
            "the promise came back"
        );
        disk.write(GIB + 1);
        match gate.pack_quota("team") {
            PlanVerdict::Refuse(line) => {
                assert!(line.starts_with(crate::headroom::AT_FLOOR), "{line}")
            }
            other => panic!("under the floor: {other:?}"),
        }
    }

    /// A disk that cannot be measured refuses — mid-stream too.
    #[test]
    fn an_unmeasurable_disk_refuses_the_next_chunk() {
        let disk = FakeDisk::new(10 * GIB, 5 * GIB);
        let ledger = ledger(&disk, 2 * GIB, 0);
        let hold = ledger.hold(PUSH_COPIES);
        hold.admit(MIB).unwrap();
        disk.fail.store(true, Ordering::SeqCst);
        let line = hold.admit(MIB).unwrap_err();
        assert!(line.contains("could not check its storage"), "{line}");
        assert_eq!(ledger.reserved(), 0);
    }

    /// An LFS object declares its size: all of it, or a refusal.
    #[test]
    fn a_declared_size_is_admitted_whole_or_refused() {
        let disk = FakeDisk::new(10 * GIB, 3 * GIB);
        let ledger = ledger(&disk, 2 * GIB, 0);
        let fits = ledger.hold(PUSH_COPIES);
        fits.admit_all(200 * MIB)
            .expect("700 MiB (3× + the filesystem's slack) of 1 GiB");
        let no = ledger.hold(PUSH_COPIES);
        let line = no
            .admit_all(200 * MIB)
            .expect_err("700 MiB more does not fit in 324 MiB");
        assert!(line.starts_with(FULL_NOW), "{line}");
        assert_eq!(
            ledger.reserved(),
            fits.promise_for(200 * MIB),
            "the refused one holds nothing, the other its whole promise"
        );
    }

    #[test]
    fn enospc_and_edquot_are_out_of_space_and_a_plain_cancel_is_not() {
        assert!(is_out_of_space(&std::io::Error::from_raw_os_error(28)));
        assert!(is_out_of_space(&std::io::Error::from_raw_os_error(122)));
        assert!(is_out_of_space(&std::io::Error::from(
            std::io::ErrorKind::StorageFull
        )));
        assert!(!is_out_of_space(&std::io::Error::from_raw_os_error(125)));
    }
}