Skip to main content

gunnar_ssh/
space.rs

1//! **Space admission: the disk floor, counted against the bytes still arriving.**
2//!
3//! The disk-headroom gate ([`crate::headroom`]) answers ONE question before a
4//! push's first byte: is the data set under its floor right now. It cannot
5//! answer the question that matters when eight pushes arrive at once, because a
6//! pack header carries an object count and no byte count, and every one of the
7//! eight is admitted against the same measurement.
8//!
9//! # The defect this closes (cov-13, the UpCloud mockup, 2026-10-06)
10//!
11//! Eight concurrent 512 MiB pushes into gunnar main's 19.9 GiB set, 2 GiB floor.
12//! Each was admitted at the door with room to spare; together they ran /data out
13//! of space while still receiving, and the store failed them MID-RECEIVE as
14//! `uring: op … failed: Operation canceled` — an ENOSPC that does not say so, a
15//! client told "the repository is unavailable; try again", and the floor the
16//! gate exists to keep simply crossed.
17//!
18//! # What one pushed byte costs on /data (MEASURED, `tests/push_space_admission.rs`)
19//!
20//! A push's bytes land three times before one of the copies is let go:
21//!
22//! 1. the **spool** — a pack of 64 MiB or more is spooled to
23//!    `<repo>/gunnar-git.znippy/incoming-*.pack` while it arrives
24//!    (`gunnar-server::storage::SPOOL_ABOVE`), so RAM never holds it;
25//! 2. the **transaction log** — the pack's exact bytes are a `Pack` frame before
26//!    the store sees one of them (the twin's log; raw, not compressed);
27//! 3. the **store** — the archive appends the pack verbatim.
28//!
29//! Then the spool is deleted. So the PEAK is [`PUSH_COPIES`] = 3 × the pushed
30//! bytes and the steady state 2 × until the log is trimmed (the log's byte
31//! bound, `gunnar-server::retention`). The ledger reserves the peak and gives
32//! the copies back as they actually land on the disk, so a byte is counted
33//! once: as a promise until it is written, as `statvfs` after.
34//!
35//! # The rule
36//!
37//! One server-wide [`SpaceLedger`]: the bytes promised to transfers still
38//! landing. A transfer takes a [`SpaceHold`] and asks it, chunk by chunk,
39//! BEFORE the chunk is read off the wire:
40//!
41//! ```text
42//!     admit c  ⇔  free − floor − promised − (lane, for a big push)  ≥  3.5 × c
43//!
44//! (3 copies, each with the filesystem's slack: XFS counts what it preallocates past a growing
45//! file's end — MEASURED at start, [`FsSlack`]; never under [`PUSH_FS_SLACK_PERMILLE`].)
46//! ```
47//!
48//! measured with one `statvfs` per ask under the ledger's lock, so two pushes
49//! can never both be granted the same byte. A chunk that does not fit is
50//! granted in part; a push that is granted nothing is REFUSED — that push, not
51//! the others — with [`FULL_NOW`]'s line: its reservation is given back at
52//! once, its spool is deleted by its own failing ingest, and nothing of it was
53//! framed or stored. The refusal is final for that push and transient for the
54//! data set: try again later.
55//!
56//! # Fair: one huge push must not starve everything
57//!
58//! * **The small-push lane.** A push that has sent more than
59//!   [`SMALL_PUSH_BYTES`] (16 MiB) may not take the last [`SMALL_PUSH_LANE`]
60//!   (256 MiB) above the floor. A big push is therefore refused while there is
61//!   still room for ordinary pushes, and a 40 KiB commit keeps landing beside a
62//!   512 MiB import that does not fit — the import is the one told "full".
63//! * **Whoever cannot be granted is refused, and gives its room back.** N equal
64//!   pushes into room for M < N of them do not all stall at a shared line: the
65//!   first one short is refused and its promise (and spool) returns to the
66//!   others, so the remaining ones finish. Bytes are granted in [`LEASE_BYTES`]
67//!   steps, so a big push is never granted the room ahead of the push beside it.
68//!
69//! # What stays OUT of the ledger, on purpose
70//!
71//! `Admin.Capacity` (and so monetize's emergency-growth trigger, which reads
72//! `used = total − servable`) is the RAW `statvfs` — what is on the disk. A
73//! promise is not a byte used: a burst that is refused would otherwise read as
74//! a full set for the seconds it lasted and buy a growth nobody needed. The
75//! bytes that DO land move `used` as they land, which is when the trigger
76//! should see them.
77//!
78//! What a push writes AFTER its hold is given back — the indexer's rows,
79//! journal pages, a `Refs` frame — is not reserved: it is small beside the pack
80//! and it is what the floor itself is kept for.
81
82use std::path::{Path, PathBuf};
83use std::sync::{Arc, Mutex, MutexGuard};
84
85use crate::headroom::{human, DiskFloor};
86
87const MIB: u64 = 1024 * 1024;
88
89/// **How many bytes of /data one pushed byte occupies at its peak**: spool,
90/// transaction-log frame, store. See the module header for the measurement.
91pub const PUSH_COPIES: u64 = 3;
92
93/// **The filesystem's slack on top of the copies, ‰** (r5.1, 2026-10-07): what the data set's
94/// filesystem counts as used beyond the bytes written while a file grows. XFS — the appliance's
95/// /data — preallocates past a growing file's end and counts it used until the file is closed
96/// (or trimmed): MEASURED on XFS, one 96 MiB push peaked at 3.24× its bytes on the data set and
97/// held 2.16× after it landed, against the three copies (tmpfs: under 3.15×). 1/6 on top of
98/// every copy: a promise of 3.5× per wire byte, given back when the hold drops.
99pub const PUSH_FS_SLACK_PERMILLE: u64 = 167;
100
101/// **What the ledger promises per wire byte of a framed push, ‰**, at the least: the
102/// [`PUSH_COPIES`] with the floor of the filesystem's slack ([`PUSH_FS_SLACK_PERMILLE`]). A
103/// ledger built with a measured [`FsSlack`] promises more where the filesystem preallocates more.
104pub const PUSH_RESERVE_PERMILLE: u64 = PUSH_COPIES * (1000 + PUSH_FS_SLACK_PERMILLE);
105
106// ── the filesystem's preallocation, measured at start (round 5.2, 2026-10-07) ─────────────────
107//
108// MEASURED: XFS preallocates past a growing file's end by DOUBLING — a file written in 1 MiB
109// steps holds 16 MiB allocated at 8 MiB, 32 at 16, 64 at 32, 128 at 64 — and counts it used until
110// the file is closed: right after a power of two a growing file occupies 2× its bytes (dwarf, a
111// 256 KiB stripe; oden's 1 MiB stripe the same). The pushed bytes' three copies are such growing
112// files (the spool, the transaction log, the store). One push of 96 MiB peaked at 3.24× its bytes
113// on dwarf and 3.58× on oden — over the fixed 3.5× this ledger used to promise. tmpfs and ext4
114// preallocate nothing that `st_blocks` counts. So the slack is measured on the data set itself.
115
116/// **The probe's spool-sized write** ([`measure_fs_slack`]): grown in 1 MiB steps past 32 MiB,
117/// so a doubling filesystem shows its worst (right after a power of two) inside it.
118pub const PROBE_BYTES: u64 = 33 * MIB;
119
120/// The probe measures from here on: a push this size or bigger is a big push
121/// ([`SMALL_PUSH_BYTES`]); what a smaller file holds over its bytes is small beside the floor.
122pub const PROBE_FROM: u64 = SMALL_PUSH_BYTES;
123
124/// **The margin on top of what was measured, ‰.**
125pub const SLACK_MARGIN_PERMILLE: u64 = 50;
126
127/// **The slack when the filesystem could not be measured, ‰**: XFS's doubling (1000 ‰) and the
128/// margin — the most a growing file was seen to occupy. Conservative on purpose: it only refuses a
129/// big push sooner near the floor.
130pub const UNMEASURED_SLACK_PERMILLE: u64 = 1000 + SLACK_MARGIN_PERMILLE;
131
132/// The probe needs this much free above nothing to run (its file and what the filesystem
133/// preallocates for it, twice over); less, and the slack is [`UNMEASURED_SLACK_PERMILLE`].
134pub const PROBE_NEEDS_FREE: u64 = 4 * PROBE_BYTES;
135
136/// The probe's file names (`<root>/.gunnar-space-probe-<pid>-<n>`): one a crash left behind is
137/// deleted by the next probe.
138pub const PROBE_FILE_PREFIX: &str = ".gunnar-space-probe-";
139
140/// **The filesystem's slack as this data set's ledger reserves it.**
141#[derive(Debug, Clone, Copy, PartialEq, Eq)]
142pub struct FsSlack {
143    /// What a growing spool-sized file occupied beyond its bytes, ‰ of them, at worst — `None`
144    /// when it could not be measured.
145    pub measured_permille: Option<u64>,
146    /// What the ledger reserves on top of every copy, ‰: the measurement and
147    /// [`SLACK_MARGIN_PERMILLE`], never under [`PUSH_FS_SLACK_PERMILLE`];
148    /// [`UNMEASURED_SLACK_PERMILLE`] unmeasured.
149    pub slack_permille: u64,
150}
151
152impl Default for FsSlack {
153    /// The floor: what a ledger built without a measurement reserves (a test's, a tool's).
154    fn default() -> Self {
155        FsSlack {
156            measured_permille: None,
157            slack_permille: PUSH_FS_SLACK_PERMILLE,
158        }
159    }
160}
161
162impl FsSlack {
163    /// The slack from a measurement (`None`: not measured).
164    pub fn from_measured(measured_permille: Option<u64>) -> FsSlack {
165        let slack_permille = match measured_permille {
166            Some(m) => m
167                .saturating_add(SLACK_MARGIN_PERMILLE)
168                .max(PUSH_FS_SLACK_PERMILLE),
169            None => UNMEASURED_SLACK_PERMILLE,
170        };
171        FsSlack {
172            measured_permille,
173            slack_permille,
174        }
175    }
176
177    /// **Measure `root`'s filesystem** ([`measure_fs_slack`]) and say why when it cannot be.
178    pub fn of(root: &Path) -> (FsSlack, Option<String>) {
179        match measure_fs_slack(root) {
180            Ok(m) => (FsSlack::from_measured(Some(m)), None),
181            Err(e) => (FsSlack::from_measured(None), Some(e.to_string())),
182        }
183    }
184
185    /// What one wire byte of a transfer of `copies` copies is promised, ‰.
186    pub fn reserve_permille(&self, copies: u64) -> u64 {
187        copies.max(1).saturating_mul(1000 + self.slack_permille)
188    }
189
190    /// **The start-up line, in plain words**: what the filesystem was measured to preallocate and
191    /// what every push therefore reserves.
192    pub fn words(&self, root: &Path, unmeasured_because: Option<&str>) -> String {
193        let reserve = times(self.reserve_permille(PUSH_COPIES));
194        match self.measured_permille {
195            Some(m) => format!(
196                "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}",
197                root = root.display(),
198                held = times(1000 + m),
199                probe = human(PROBE_BYTES),
200                margin = SLACK_MARGIN_PERMILLE,
201                floor = times(PUSH_RESERVE_PERMILLE),
202            ),
203            None => format!(
204                "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",
205                root = root.display(),
206                why = unmeasured_because.unwrap_or("no reason given"),
207                margin = SLACK_MARGIN_PERMILLE,
208            ),
209        }
210    }
211}
212
213/// `3500` ‰ as `3.50×`.
214fn times(permille: u64) -> String {
215    format!("{}.{:02}×", permille / 1000, (permille % 1000) / 10)
216}
217
218/// **What `dir`'s filesystem occupies beyond a growing file's bytes, ‰ of them, at worst**: a
219/// [`PROBE_BYTES`] file is written in 1 MiB steps and its `st_blocks` read after each step from
220/// [`PROBE_FROM`] on and once more after an `fdatasync`, all while it is open — the way a pushed
221/// pack's spool, frame and store grow. The file is deleted before this returns; one a crash left
222/// is deleted by the next. Refused when less than [`PROBE_NEEDS_FREE`] is free.
223pub fn measure_fs_slack(dir: &Path) -> std::io::Result<u64> {
224    use std::io::Write as _;
225    use std::os::unix::fs::MetadataExt as _;
226    if let Ok(entries) = std::fs::read_dir(dir) {
227        for e in entries.flatten() {
228            if e.file_name()
229                .to_string_lossy()
230                .starts_with(PROBE_FILE_PREFIX)
231            {
232                let _ = std::fs::remove_file(e.path());
233            }
234        }
235    }
236    let (free, _) = statvfs_bytes(dir)?;
237    if free < PROBE_NEEDS_FREE {
238        return Err(std::io::Error::other(format!(
239            "{} free, and the probe needs {}",
240            human(free),
241            human(PROBE_NEEDS_FREE)
242        )));
243    }
244    static SEQ: std::sync::atomic::AtomicU64 = std::sync::atomic::AtomicU64::new(0);
245    let path = dir.join(format!(
246        "{PROBE_FILE_PREFIX}{}-{}",
247        std::process::id(),
248        SEQ.fetch_add(1, std::sync::atomic::Ordering::Relaxed)
249    ));
250    struct Gone(PathBuf);
251    impl Drop for Gone {
252        fn drop(&mut self) {
253            let _ = std::fs::remove_file(&self.0);
254        }
255    }
256    let _gone = Gone(path.clone());
257    let mut file = std::fs::OpenOptions::new()
258        .write(true)
259        .create_new(true)
260        .open(&path)?;
261    let chunk: Vec<u8> = (0..MIB as usize)
262        .map(|i| (i.wrapping_mul(2_654_435_761) >> 13) as u8)
263        .collect();
264    let over = |file: &std::fs::File, written: u64| -> std::io::Result<u64> {
265        let held = file.metadata()?.blocks().saturating_mul(512);
266        Ok(held.saturating_sub(written).saturating_mul(1000) / written.max(1))
267    };
268    let (mut written, mut worst) = (0u64, 0u64);
269    while written < PROBE_BYTES {
270        file.write_all(&chunk)?;
271        written += MIB;
272        if written >= PROBE_FROM {
273            worst = worst.max(over(&file, written)?);
274        }
275    }
276    file.sync_data()?;
277    worst = worst.max(over(&file, written)?);
278    Ok(worst)
279}
280
281/// A push that has sent more than this is a BIG push, held out of the
282/// [`SMALL_PUSH_LANE`]. 16 MiB is an ordinary commit's pack many times over.
283pub const SMALL_PUSH_BYTES: u64 = 16 * MIB;
284
285/// The room above the floor a big push may never take, kept for small pushes:
286/// at [`PUSH_COPIES`] that is five pushes of [`SMALL_PUSH_BYTES`] at once, or
287/// thousands of ordinary ones.
288pub const SMALL_PUSH_LANE: u64 = 256 * MIB;
289
290/// The most wire bytes one ask grants: one `statvfs` per 4 MiB a push sends.
291pub const LEASE_BYTES: u64 = 4 * MIB;
292
293/// Less room than this above the floor and the promises — one 64 KiB chunk at
294/// [`PUSH_COPIES`] — and the door says full before the first byte.
295pub const MIN_ROOM: u64 = PUSH_RESERVE_PERMILLE * 64 * 1024 / 1000;
296
297/// **The start of every refusal for room**, so a client, a kit and a test can
298/// recognise it. Deliberately NOT the floor gate's phrase
299/// ([`crate::headroom::AT_FLOOR`]): that one says the set is AT its floor and stays
300/// so until space is reclaimed or grown — final — and tools stop on it. This one
301/// says this push does not fit beside what is already arriving — try again.
302pub const FULL_NOW: &str = "the data set is full right now";
303
304/// What the ledger measures: `(free, total)` in bytes, `free` as an
305/// unprivileged writer sees it (`f_bavail`).
306pub trait SpaceProbe: Send + Sync + std::fmt::Debug {
307    /// One measurement. An error is a refusal, never room.
308    fn measure(&self) -> std::io::Result<(u64, u64)>;
309}
310
311/// The probe that ships: one `statvfs` on the data root.
312#[derive(Debug, Clone)]
313pub struct Statvfs(pub PathBuf);
314
315impl SpaceProbe for Statvfs {
316    fn measure(&self) -> std::io::Result<(u64, u64)> {
317        statvfs_bytes(&self.0)
318    }
319}
320
321/// `(f_bavail, f_blocks)` in bytes for the filesystem `root` lives on.
322pub fn statvfs_bytes(root: &Path) -> std::io::Result<(u64, u64)> {
323    let vfs = rustix::fs::statvfs(root).map_err(std::io::Error::from)?;
324    let frsize = if vfs.f_frsize == 0 {
325        vfs.f_bsize
326    } else {
327        vfs.f_frsize
328    };
329    Ok((
330        vfs.f_bavail.saturating_mul(frsize),
331        vfs.f_blocks.saturating_mul(frsize),
332    ))
333}
334
335/// One reading of the data set and of what is promised on it.
336#[derive(Debug, Clone, Copy, PartialEq, Eq)]
337pub struct Standing {
338    /// Bytes free now (`statvfs`).
339    pub free: u64,
340    /// The filesystem's size.
341    pub total: u64,
342    /// The floor in force for that size.
343    pub floor: u64,
344    /// Bytes promised to transfers still landing and not on the disk yet.
345    pub reserved: u64,
346}
347
348impl Standing {
349    /// What may still be promised: free, less the floor, less the promises.
350    pub fn room(&self) -> u64 {
351        self.free
352            .saturating_sub(self.floor)
353            .saturating_sub(self.reserved)
354    }
355}
356
357#[derive(Debug, Default)]
358struct LedgerState {
359    /// Σ over live holds of their `held`.
360    reserved: u64,
361    /// Live holds.
362    holds: usize,
363    /// Transfers refused for room since the process started.
364    refused: u64,
365}
366
367/// **The server-wide account of bytes promised to transfers still landing.**
368/// One per data root; `gunnar serve` builds it with the disk-headroom gate and
369/// hands the same `Arc` to every door that writes (push, mirror, LFS).
370#[derive(Debug)]
371pub struct SpaceLedger {
372    root: PathBuf,
373    floor: DiskFloor,
374    probe: Arc<dyn SpaceProbe>,
375    lane: u64,
376    /// The filesystem's slack every promise carries ([`FsSlack`]).
377    slack: FsSlack,
378    state: Mutex<LedgerState>,
379}
380
381impl SpaceLedger {
382    /// The ledger for the data set `root` lives on, measured with `statvfs`.
383    pub fn new(root: impl Into<PathBuf>, floor: DiskFloor) -> Self {
384        let root = root.into();
385        let probe = Arc::new(Statvfs(root.clone()));
386        Self::with_probe(root, floor, probe)
387    }
388
389    /// The same, measured by `probe` — a test's fake disk.
390    pub fn with_probe(
391        root: impl Into<PathBuf>,
392        floor: DiskFloor,
393        probe: Arc<dyn SpaceProbe>,
394    ) -> Self {
395        SpaceLedger {
396            root: root.into(),
397            floor,
398            probe,
399            lane: SMALL_PUSH_LANE,
400            slack: FsSlack::default(),
401            state: Mutex::new(LedgerState::default()),
402        }
403    }
404
405    /// **Promise the filesystem's slack as measured** ([`FsSlack::of`]) instead of the floor
406    /// ([`PUSH_FS_SLACK_PERMILLE`]). `gunnar serve` measures its data set once at start.
407    pub fn with_fs_slack(mut self, slack: FsSlack) -> Self {
408        self.slack = slack;
409        self
410    }
411
412    /// The filesystem's slack in force.
413    pub fn fs_slack(&self) -> FsSlack {
414        self.slack
415    }
416
417    /// Keep `bytes` above the floor for small pushes instead of
418    /// [`SMALL_PUSH_LANE`]. `0` turns the lane off.
419    pub fn with_small_lane(mut self, bytes: u64) -> Self {
420        self.lane = bytes;
421        self
422    }
423
424    /// The data root this ledger speaks for (named in its refusals).
425    pub fn root(&self) -> &Path {
426        &self.root
427    }
428
429    /// The floor it keeps.
430    pub fn floor(&self) -> DiskFloor {
431        self.floor
432    }
433
434    /// The small-push lane in force.
435    pub fn small_lane(&self) -> u64 {
436        self.lane
437    }
438
439    /// The probe it measures with — the disk gate's `headroom` reads the same.
440    pub fn probe(&self) -> &Arc<dyn SpaceProbe> {
441        &self.probe
442    }
443
444    fn lock(&self) -> MutexGuard<'_, LedgerState> {
445        self.state
446            .lock()
447            .unwrap_or_else(|poisoned| poisoned.into_inner())
448    }
449
450    /// Bytes promised right now and not on the disk yet.
451    pub fn reserved(&self) -> u64 {
452        self.lock().reserved
453    }
454
455    /// Transfers holding a promise right now.
456    pub fn holds(&self) -> usize {
457        self.lock().holds
458    }
459
460    /// Transfers refused for room since this ledger was built.
461    pub fn refused(&self) -> u64 {
462        self.lock().refused
463    }
464
465    /// One measurement, with the promises beside it.
466    pub fn standing(&self) -> std::io::Result<Standing> {
467        let reserved = self.reserved();
468        self.measure_with(reserved)
469    }
470
471    fn measure_with(&self, reserved: u64) -> std::io::Result<Standing> {
472        let (free, total) = self.probe.measure()?;
473        Ok(Standing {
474            free,
475            total,
476            floor: self.floor.bytes_for(total),
477            reserved,
478        })
479    }
480
481    /// **A promise for one transfer**, held to `copies` bytes of /data per byte
482    /// it sends ([`PUSH_COPIES`] for a framed push). Everything it still holds
483    /// is given back when it drops — landed, refused, cut or crashed.
484    pub fn hold(self: &Arc<Self>, copies: u64) -> SpaceHold {
485        self.lock().holds += 1;
486        SpaceHold {
487            ledger: Arc::clone(self),
488            copies: copies.max(1),
489            state: Mutex::new(HoldState::default()),
490        }
491    }
492
493    /// **The up-front line, before any byte**: `Some` when what is promised
494    /// already covers everything above the floor. The floor gate's own line
495    /// (under the floor with nothing promised) is the gate's, not this.
496    ///
497    /// The pusher is told in plain words; the numbers go to the journal
498    /// (wiring tracer B 2026-10-08: a customer never sees room left).
499    pub fn full_before_the_first_byte(&self) -> Option<String> {
500        let standing = self.standing().ok()?;
501        if standing.free <= standing.floor || standing.reserved == 0 || standing.room() >= MIN_ROOM
502        {
503            return None;
504        }
505        tracing::warn!(
506            target: "gunnar::space",
507            event = "push.space.full_up_front",
508            root = %self.root.display(),
509            free = %human(standing.free),
510            floor = %human(standing.floor),
511            disk_floor = %self.floor,
512            promised = %human(standing.reserved),
513            "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"
514        );
515        Some(format!(
516            "{FULL_NOW}: everything free is promised to pushes already arriving, so nothing was \
517             received. Try again in a while; the storage is grown when it fills"
518        ))
519    }
520
521    /// The refusal for a transfer that cannot be granted its next chunk: the
522    /// pusher's plain line; what was measured goes to the journal.
523    fn full_line(
524        &self,
525        standing: &Standing,
526        own_held: u64,
527        sent: u64,
528        copies: u64,
529        lane: u64,
530    ) -> String {
531        let others = standing.reserved.saturating_sub(own_held);
532        let prealloc = match self.slack.measured_permille {
533            Some(m) => format!(
534                "measured at start to hold {} a growing file's bytes",
535                times(1000 + m)
536            ),
537            None => "not measured, so a doubling is assumed".to_owned(),
538        };
539        tracing::warn!(
540            target: "gunnar::space",
541            event = "push.space.refused",
542            root = %self.root.display(),
543            free = %human(standing.free),
544            floor = %human(standing.floor),
545            disk_floor = %self.floor,
546            small_lane = %human(lane),
547            promised_to_others = %human(others),
548            sent = %human(sent),
549            copies,
550            reserve = %times(self.slack.reserve_permille(copies)),
551            filesystem = %prealloc,
552            "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"
553        );
554        format!(
555            "{FULL_NOW}: this push does not fit beside the pushes already arriving, so nothing of it \
556             was kept. Try again in a while; the storage is grown when it fills"
557        )
558    }
559
560    fn unmeasured_line(&self, err: &std::io::Error) -> String {
561        tracing::error!(
562            target: "gunnar::space",
563            event = "push.space.unmeasured",
564            root = %self.root.display(),
565            %err,
566            "the data set could not be measured: a push was refused rather than guessing there is room"
567        );
568        format!(
569            "{FULL_NOW}: Gunnar could not check its storage, so this push was refused rather than \
570             risk it. Nothing of it was kept; try again in a while"
571        )
572    }
573}
574
575#[derive(Debug, Default)]
576struct HoldState {
577    /// Bytes of /data promised to this transfer and not on the disk yet.
578    held: u64,
579    /// Wire bytes granted so far.
580    admitted: u64,
581    /// Set when it was refused; every later ask is the same refusal.
582    refusal: Option<String>,
583}
584
585/// **One transfer's promise.** Asked before each chunk ([`Self::admit`]),
586/// told as its copies land ([`Self::realize`]), given back when it drops.
587#[derive(Debug)]
588pub struct SpaceHold {
589    ledger: Arc<SpaceLedger>,
590    copies: u64,
591    state: Mutex<HoldState>,
592}
593
594impl SpaceHold {
595    fn lock(&self) -> MutexGuard<'_, HoldState> {
596        self.state
597            .lock()
598            .unwrap_or_else(|poisoned| poisoned.into_inner())
599    }
600
601    /// **May this transfer take up to `want` more wire bytes?** `Ok(n)` with
602    /// `0 < n ≤ want` grants `n` (promising `copies × n` of /data); `Err(line)`
603    /// refuses the transfer — for good: every later ask is the same `Err`, and
604    /// everything it held is given back now, before its spool is even gone.
605    pub fn admit(&self, want: u64) -> Result<u64, String> {
606        let mut hold = self.lock();
607        if let Some(line) = &hold.refusal {
608            return Err(line.clone());
609        }
610        if want == 0 {
611            return Ok(0);
612        }
613        let ledger = &self.ledger;
614        let mut state = ledger.lock();
615        let standing = match ledger.measure_with(state.reserved) {
616            Ok(standing) => standing,
617            Err(err) => {
618                let line = ledger.unmeasured_line(&err);
619                Self::refuse(&mut hold, &mut state, line.clone());
620                return Err(line);
621            }
622        };
623        let big = hold.admitted.saturating_add(want) > SMALL_PUSH_BYTES;
624        let lane = if big { ledger.lane } else { 0 };
625        let granted =
626            want.min(standing.room().saturating_sub(lane).saturating_mul(1000) / self.per_mille());
627        if granted == 0 {
628            let line = ledger.full_line(&standing, hold.held, hold.admitted, self.copies, lane);
629            Self::refuse(&mut hold, &mut state, line.clone());
630            return Err(line);
631        }
632        let promise = self.promise_for(granted);
633        state.reserved = state.reserved.saturating_add(promise);
634        hold.held = hold.held.saturating_add(promise);
635        hold.admitted = hold.admitted.saturating_add(granted);
636        Ok(granted)
637    }
638
639    /// **All of `bytes`, or a refusal** — a transfer that declares its size
640    /// up front (an LFS object). Partial room is no room here.
641    pub fn admit_all(&self, bytes: u64) -> Result<(), String> {
642        let mut left = bytes;
643        while left > 0 {
644            // `admit` grants what fits; asking again for the rest either gets
645            // it (nothing changed) or refuses (and gives everything back).
646            let granted = self.admit(left)?;
647            if granted < left {
648                let mut hold = self.lock();
649                let mut state = self.ledger.lock();
650                let standing = self
651                    .ledger
652                    .measure_with(state.reserved)
653                    .unwrap_or(Standing {
654                        free: 0,
655                        total: 0,
656                        floor: 0,
657                        reserved: state.reserved,
658                    });
659                let lane = if bytes > SMALL_PUSH_BYTES {
660                    self.ledger.lane
661                } else {
662                    0
663                };
664                let line = self
665                    .ledger
666                    .full_line(&standing, hold.held, bytes, self.copies, lane);
667                Self::refuse(&mut hold, &mut state, line.clone());
668                return Err(line);
669            }
670            left -= granted;
671        }
672        Ok(())
673    }
674
675    fn refuse(hold: &mut HoldState, state: &mut LedgerState, line: String) {
676        state.reserved = state.reserved.saturating_sub(hold.held);
677        state.refused += 1;
678        hold.held = 0;
679        hold.refusal = Some(line);
680    }
681
682    /// **`bytes` of what this transfer promised are on the disk now**, where
683    /// `statvfs` counts them — so they stop being counted twice. Call it after
684    /// the write returns, never before: a promise released early is a byte
685    /// nobody counts.
686    pub fn realize(&self, bytes: u64) {
687        let mut hold = self.lock();
688        let given = bytes.min(hold.held);
689        if given == 0 {
690            return;
691        }
692        hold.held -= given;
693        let mut state = self.ledger.lock();
694        state.reserved = state.reserved.saturating_sub(given);
695    }
696
697    /// Bytes of /data this transfer still has promised.
698    pub fn held(&self) -> u64 {
699        self.lock().held
700    }
701
702    /// Wire bytes granted so far.
703    pub fn admitted(&self) -> u64 {
704        self.lock().admitted
705    }
706
707    /// The refusal, once refused.
708    pub fn refusal(&self) -> Option<String> {
709        self.lock().refusal.clone()
710    }
711
712    /// The copies of /data per wire byte this hold promises.
713    pub fn copies(&self) -> u64 {
714        self.copies
715    }
716
717    /// What this hold promises per wire byte, ‰: its copies with the filesystem's slack
718    /// ([`SpaceLedger::fs_slack`]; the floor [`PUSH_FS_SLACK_PERMILLE`] unless measured).
719    pub fn per_mille(&self) -> u64 {
720        self.ledger.slack.reserve_permille(self.copies)
721    }
722
723    /// The bytes of /data `wire` wire bytes are promised: [`Self::per_mille`], rounded up. What
724    /// lands is given back by [`Self::realize`]; the slack is given back when the hold drops.
725    pub fn promise_for(&self, wire: u64) -> u64 {
726        wire.saturating_mul(self.per_mille()).div_ceil(1000)
727    }
728
729    /// **The line for a store that ran out of room anyway** — an ENOSPC the
730    /// admission did not foresee (a writer outside the ledger). Said in the
731    /// same words, with the numbers as they are now, so a client is never told
732    /// `uring: op failed: Operation canceled`.
733    pub fn full_now(&self) -> String {
734        let mut hold = self.lock();
735        if let Some(line) = &hold.refusal {
736            return line.clone();
737        }
738        let mut state = self.ledger.lock();
739        let line = match self.ledger.measure_with(state.reserved) {
740            Ok(standing) => {
741                self.ledger
742                    .full_line(&standing, hold.held, hold.admitted, self.copies, 0)
743            }
744            Err(err) => self.ledger.unmeasured_line(&err),
745        };
746        Self::refuse(&mut hold, &mut state, line.clone());
747        line
748    }
749
750    /// Was the data set at its floor when this transfer failed — free now,
751    /// less the `freed` bytes its own failure just gave back (its spool), at
752    /// or under the floor? An unmeasurable disk counts as yes. What a store
753    /// failure that does not SAY "no space" (io_uring's `Operation canceled`
754    /// for a linked chain whose head hit ENOSPC) is judged by.
755    pub fn at_the_floor(&self, freed: u64) -> bool {
756        match self.ledger.probe.measure() {
757            Ok((free, total)) => {
758                free.saturating_sub(freed)
759                    <= self.ledger.floor.bytes_for(total).saturating_add(MIN_ROOM)
760            }
761            Err(_) => true,
762        }
763    }
764}
765
766impl Drop for SpaceHold {
767    fn drop(&mut self) {
768        let held = std::mem::take(&mut self.lock().held);
769        let mut state = self.ledger.lock();
770        state.reserved = state.reserved.saturating_sub(held);
771        state.holds = state.holds.saturating_sub(1);
772    }
773}
774
775/// **Is this error the disk being full?** `ENOSPC`, `EDQUOT`, or the
776/// `StorageFull`/`QuotaExceeded` kinds std maps them to — found anywhere in
777/// the chain an `anyhow` or `io::Error` carries. `Operation canceled` (125) is
778/// how io_uring reports the tail of a linked chain whose head hit ENOSPC, so it
779/// counts only beside a disk that measures at its floor ([`SpaceHold::at_the_floor`]).
780pub fn is_out_of_space(err: &std::io::Error) -> bool {
781    use std::io::ErrorKind;
782    matches!(
783        err.kind(),
784        ErrorKind::StorageFull | ErrorKind::QuotaExceeded
785    ) || matches!(err.raw_os_error(), Some(28) | Some(122))
786}
787
788#[cfg(test)]
789mod tests {
790    use super::*;
791    use std::sync::atomic::{AtomicI64, AtomicU64, Ordering};
792
793    const GIB: u64 = 1024 * MIB;
794
795    /// **A disk that is only numbers**: free goes down when a test writes and
796    /// up when it deletes, and the lowest it ever read is kept — the floor
797    /// assertion is about that, not about a sample.
798    #[derive(Debug)]
799    struct FakeDisk {
800        free: AtomicI64,
801        total: u64,
802        lowest: AtomicI64,
803        fail: std::sync::atomic::AtomicBool,
804    }
805
806    impl FakeDisk {
807        fn new(total: u64, free: u64) -> Arc<FakeDisk> {
808            Arc::new(FakeDisk {
809                free: AtomicI64::new(free as i64),
810                total,
811                lowest: AtomicI64::new(free as i64),
812                fail: Default::default(),
813            })
814        }
815        fn write(&self, bytes: u64) {
816            let now = self.free.fetch_sub(bytes as i64, Ordering::SeqCst) - bytes as i64;
817            self.lowest.fetch_min(now, Ordering::SeqCst);
818        }
819        fn delete(&self, bytes: u64) {
820            self.free.fetch_add(bytes as i64, Ordering::SeqCst);
821        }
822        fn lowest(&self) -> i64 {
823            self.lowest.load(Ordering::SeqCst)
824        }
825    }
826
827    impl SpaceProbe for FakeDisk {
828        fn measure(&self) -> std::io::Result<(u64, u64)> {
829            if self.fail.load(Ordering::SeqCst) {
830                return Err(std::io::Error::other("the mount went away"));
831            }
832            Ok((self.free.load(Ordering::SeqCst).max(0) as u64, self.total))
833        }
834    }
835
836    /// ★ **The slack is what the filesystem preallocates, plus a margin, never under a sixth**
837    /// (round 5.2, 2026-10-07): measured nothing (tmpfs, ext4) is the floor, 3.5×; XFS's
838    /// doubling (1000 ‰) is 6.15×; not measured is the doubling assumed. The ledger promises it,
839    /// says it at start in plain words, and a big push refused near the floor is told that room is
840    /// reserved for the filesystem's preallocation.
841    #[test]
842    fn the_slack_is_the_measured_preallocation_with_a_margin_and_never_under_a_sixth() {
843        assert_eq!(
844            FsSlack::from_measured(Some(0)).slack_permille,
845            PUSH_FS_SLACK_PERMILLE
846        );
847        assert_eq!(
848            FsSlack::from_measured(Some(0)).reserve_permille(PUSH_COPIES),
849            PUSH_RESERVE_PERMILLE
850        );
851        assert_eq!(FsSlack::from_measured(Some(1000)).slack_permille, 1050);
852        assert_eq!(
853            FsSlack::from_measured(Some(1000)).reserve_permille(PUSH_COPIES),
854            6150
855        );
856        assert_eq!(
857            FsSlack::from_measured(None).slack_permille,
858            UNMEASURED_SLACK_PERMILLE
859        );
860        assert_eq!(
861            FsSlack::default().reserve_permille(PUSH_COPIES),
862            PUSH_RESERVE_PERMILLE,
863            "a ledger nobody measured promises the floor"
864        );
865        let said = FsSlack::from_measured(Some(1000)).words(Path::new("/data"), None);
866        assert!(
867            said.contains("held 2.00× a growing file's bytes")
868                && said.contains("every push reserves 6.15× its bytes"),
869            "{said}"
870        );
871        assert!(!said.contains("  "), "one line, single-spaced: {said}");
872        let said = FsSlack::from_measured(None).words(Path::new("/data"), Some("12 MiB free"));
873        assert!(
874            said.contains("could not be measured (12 MiB free)") && said.contains("6.15×"),
875            "{said}"
876        );
877
878        let disk = FakeDisk::new(10 * GIB, 2 * GIB);
879        let ledger = Arc::new(
880            SpaceLedger::with_probe(
881                "/data",
882                DiskFloor {
883                    percent: 0,
884                    bytes: GIB,
885                },
886                Arc::clone(&disk) as Arc<dyn SpaceProbe>,
887            )
888            .with_small_lane(0)
889            .with_fs_slack(FsSlack::from_measured(Some(1000))),
890        );
891        let hold = ledger.hold(PUSH_COPIES);
892        assert_eq!(
893            hold.promise_for(100 * MIB),
894            615 * MIB,
895            "6.15× promised per wire byte"
896        );
897        let granted = hold.admit(GIB).expect("some of it fits");
898        assert_eq!(
899            granted,
900            (GIB * 1000) / 6150,
901            "a measured slack grants less near the floor"
902        );
903        let line = hold.admit(GIB).unwrap_err();
904        assert!(
905            line.starts_with(FULL_NOW) && line.contains("does not fit"),
906            "the pusher's line is plain; the 2.00x slack is the journal's: {line}"
907        );
908    }
909
910    /// ★ **The pusher is told the storage is full, in plain words — no
911    /// number, no path, no flag, no internal word** (wiring tracer B
912    /// 2026-10-08; owner: a customer never sees caps or room left). What was
913    /// free, the floor, the promises and the measured slack are the
914    /// operator's: they go to the journal, not to `git push`'s stderr.
915    #[test]
916    fn the_pusher_is_told_full_in_plain_words_and_the_numbers_go_to_the_journal() {
917        let plain = |line: &str| {
918            assert!(line.starts_with(FULL_NOW), "{line}");
919            assert!(
920                line.contains("try again") || line.contains("Try again"),
921                "{line}"
922            );
923            for internal in [
924                "/data",
925                "--disk-floor",
926                "transaction log",
927                "spool",
928                "preallocation",
929                "gunnar gc",
930                "repo delete",
931                "×",
932                "MiB",
933                "GiB",
934                "promised to pushes already arriving, and this push",
935            ] {
936                assert!(
937                    !line.contains(internal),
938                    "{internal:?} reaches the pusher: {line}"
939                );
940            }
941            assert!(
942                !line.chars().any(|c| c.is_ascii_digit()),
943                "a number reaches the pusher: {line}"
944            );
945        };
946        let disk = FakeDisk::new(10 * GIB, 2 * GIB);
947        let ledger = Arc::new(
948            SpaceLedger::with_probe(
949                "/data",
950                DiskFloor {
951                    percent: 0,
952                    bytes: GIB,
953                },
954                Arc::clone(&disk) as Arc<dyn SpaceProbe>,
955            )
956            .with_small_lane(0)
957            .with_fs_slack(FsSlack::from_measured(Some(1000))),
958        );
959        let hold = ledger.hold(PUSH_COPIES);
960        hold.admit(GIB).expect("some of it fits");
961        plain(&hold.admit(GIB).unwrap_err());
962        let second = ledger.hold(PUSH_COPIES);
963        let up_front = ledger.full_before_the_first_byte();
964        drop(second);
965        if let Some(line) = up_front {
966            plain(&line);
967        }
968        plain(&ledger.unmeasured_line(&std::io::Error::other("the mount went away")));
969    }
970
971    /// ★ **The probe sees what the filesystem holds for a growing file — XFS's doubling where
972    /// there is XFS — and leaves nothing behind** (round 5.2). Where the test's temp dir is XFS
973    /// (dwarf, oden), the worst is the doubling; elsewhere, whatever it is, it is measured.
974    #[test]
975    fn the_probe_sees_the_preallocation_of_a_growing_file_and_leaves_nothing_behind() {
976        let dir =
977            std::env::temp_dir().join(format!("gunnar-space-probe-test-{}", std::process::id()));
978        let _ = std::fs::remove_dir_all(&dir);
979        std::fs::create_dir_all(&dir).unwrap();
980        std::fs::write(
981            dir.join(format!("{PROBE_FILE_PREFIX}1-7")),
982            b"a crash left me",
983        )
984        .unwrap();
985        let measured = measure_fs_slack(&dir).expect("measured");
986        let left: Vec<_> = std::fs::read_dir(&dir)
987            .unwrap()
988            .flatten()
989            .map(|e| e.file_name())
990            .collect();
991        assert!(
992            left.is_empty(),
993            "the probe leaves nothing, and takes what a crash left: {left:?}"
994        );
995        let xfs = rustix::fs::statfs(&dir)
996            .map(|s| s.f_type as u64 == 0x5846_5342)
997            .unwrap_or(false);
998        eprintln!(
999            "MEASURED probe on {} ({}): a growing file held {} its bytes at worst",
1000            dir.display(),
1001            if xfs { "xfs" } else { "not xfs" },
1002            times(1000 + measured)
1003        );
1004        if xfs {
1005            assert!(
1006                measured >= 900,
1007                "XFS doubles a growing file's allocation; the probe saw only {} more",
1008                times(measured)
1009            );
1010        }
1011        let _ = std::fs::remove_dir_all(&dir);
1012    }
1013
1014    fn ledger(disk: &Arc<FakeDisk>, floor: u64, lane: u64) -> Arc<SpaceLedger> {
1015        Arc::new(
1016            SpaceLedger::with_probe(
1017                "/data",
1018                DiskFloor {
1019                    percent: 0,
1020                    bytes: floor,
1021                },
1022                Arc::clone(disk) as Arc<dyn SpaceProbe>,
1023            )
1024            .with_small_lane(lane),
1025        )
1026    }
1027
1028    /// One push the way the receive path drives it: ask a lease, spool what
1029    /// was granted (on the disk at once), and once the pack is whole, frame
1030    /// it and store it (two more copies), then drop the spool. `Err` is the
1031    /// refusal, after the push's spool was deleted as its ingest would.
1032    fn push(
1033        disk: &FakeDisk,
1034        ledger: &Arc<SpaceLedger>,
1035        bytes: u64,
1036        chunk: u64,
1037    ) -> Result<(), String> {
1038        let hold = ledger.hold(PUSH_COPIES);
1039        let mut spooled = 0;
1040        while spooled < bytes {
1041            match hold.admit(chunk.min(bytes - spooled)) {
1042                Ok(granted) => {
1043                    disk.write(granted);
1044                    hold.realize(granted);
1045                    spooled += granted;
1046                }
1047                Err(line) => {
1048                    disk.delete(spooled);
1049                    return Err(line);
1050                }
1051            }
1052        }
1053        disk.write(bytes); // the transaction log's frame
1054        hold.realize(bytes);
1055        disk.write(bytes); // the store
1056        hold.realize(bytes);
1057        disk.delete(spooled); // the spool, dropped
1058        drop(hold);
1059        Ok(())
1060    }
1061
1062    /// ★ cov-13: a push that would cross the floor is refused BY NAME, the line
1063    /// says what is free, what the floor is and what this push had sent, and
1064    /// the refused push gives back everything it promised.
1065    #[test]
1066    fn a_push_that_would_cross_the_floor_is_refused_by_name_and_gives_its_promise_back() {
1067        let disk = FakeDisk::new(10 * GIB, 3 * GIB);
1068        let ledger = ledger(&disk, 2 * GIB, 0);
1069        let hold = ledger.hold(PUSH_COPIES);
1070        assert_eq!(
1071            hold.admit(256 * MIB),
1072            Ok(256 * MIB),
1073            "896 MiB (3× + the filesystem's slack) of 1 GiB room fits"
1074        );
1075        assert_eq!(ledger.reserved(), hold.promise_for(256 * MIB));
1076        assert_eq!(
1077            hold.promise_for(256 * MIB),
1078            (256 * MIB * PUSH_RESERVE_PERMILLE).div_ceil(1000)
1079        );
1080        let partial = hold
1081            .admit(256 * MIB)
1082            .expect("what is left is granted in part");
1083        assert!(partial < 256 * MIB && partial > 0, "{partial}");
1084        let line = hold.admit(MIB).expect_err("no room left: refused");
1085        assert!(line.starts_with(FULL_NOW), "{line}");
1086        // The pusher's words are plain (wiring tracer B 2026-10-08); what was free, the floor
1087        // and what was sent are the journal's (`push.space.refused`).
1088        for part in ["Try again", "nothing of it was kept"] {
1089            assert!(line.contains(part), "the line must say {part:?}: {line}");
1090        }
1091        for internal in ["/data", "--disk-floor"] {
1092            assert!(
1093                !line.contains(internal),
1094                "the pusher is not told {internal:?}: {line}"
1095            );
1096        }
1097        assert!(
1098            !line.starts_with(crate::headroom::AT_FLOOR),
1099            "the transient refusal must not read as the floor gate's FINAL one (kits stop on that): {line}"
1100        );
1101        assert_eq!(ledger.reserved(), 0, "a refused push holds nothing");
1102        assert_eq!(
1103            hold.admit(1),
1104            Err(line.clone()),
1105            "refused is refused for good"
1106        );
1107        assert_eq!(ledger.refused(), 1);
1108    }
1109
1110    /// ★ cov-13's eight pushes on a fake disk, measured in sequence: the room
1111    /// is SHARED, so pushes that each fit alone cannot all be granted it, the
1112    /// first short of room is refused and its room goes back, and the others
1113    /// land — never all eight stalled at one line.
1114    #[test]
1115    fn eight_pushes_into_room_for_three_land_three_or_more_and_refuse_the_rest() {
1116        // 4 GiB above the floor; 8 × 384 MiB at 3.5× (3 copies + the filesystem's slack) is
1117        // 10.5 GiB of peak.
1118        let disk = FakeDisk::new(20 * GIB, 6 * GIB);
1119        let ledger = ledger(&disk, 2 * GIB, 0);
1120        let holds: Vec<_> = (0..8).map(|_| ledger.hold(PUSH_COPIES)).collect();
1121        let size = 384 * MIB;
1122        let mut spooled = [0u64; 8];
1123        let mut done = [false; 8];
1124        let mut refused = [false; 8];
1125        // Round-robin, 4 MiB at a time: eight pushes arriving at one speed.
1126        for _ in 0..100_000 {
1127            let mut moved = false;
1128            for i in 0..8 {
1129                if done[i] || refused[i] {
1130                    continue;
1131                }
1132                moved = true;
1133                if spooled[i] == size {
1134                    for _ in 0..2 {
1135                        disk.write(size);
1136                        holds[i].realize(size);
1137                    }
1138                    disk.delete(size);
1139                    holds[i].realize(u64::MAX);
1140                    done[i] = true;
1141                    continue;
1142                }
1143                match holds[i].admit(LEASE_BYTES.min(size - spooled[i])) {
1144                    Ok(g) => {
1145                        disk.write(g);
1146                        holds[i].realize(g);
1147                        spooled[i] += g;
1148                    }
1149                    Err(line) => {
1150                        assert!(line.starts_with(FULL_NOW), "{line}");
1151                        disk.delete(spooled[i]);
1152                        refused[i] = true;
1153                    }
1154                }
1155                assert!(
1156                    disk.free.load(Ordering::SeqCst) - ledger.reserved() as i64 >= (2 * GIB) as i64,
1157                    "promised + written crossed the floor"
1158                );
1159            }
1160            if !moved {
1161                break;
1162            }
1163        }
1164        let landed = done.iter().filter(|d| **d).count();
1165        assert!(
1166            landed >= 3,
1167            "room for {} at the peak, {} landed (refused {refused:?})",
1168            4 * GIB / holds[0].promise_for(size),
1169            landed
1170        );
1171        assert!(
1172            refused.iter().any(|r| *r),
1173            "8 × 1.3 GiB cannot fit in 4 GiB: something must be refused"
1174        );
1175        assert!(
1176            disk.lowest() >= (2 * GIB) as i64,
1177            "the floor was crossed: lowest free {}",
1178            disk.lowest()
1179        );
1180        drop(holds);
1181        assert_eq!((ledger.reserved(), ledger.holds()), (0, 0));
1182    }
1183
1184    /// ★ The floor is never crossed by REAL concurrency: sixteen threads, each
1185    /// pushing 160 MiB at its own pace into room for a handful, every write
1186    /// on the fake disk the moment it is granted.
1187    #[test]
1188    fn concurrent_pushes_never_take_the_disk_under_the_floor() {
1189        let disk = FakeDisk::new(20 * GIB, 4 * GIB);
1190        let ledger = ledger(&disk, 2 * GIB, 64 * MIB);
1191        let landed = Arc::new(AtomicU64::new(0));
1192        let refused = Arc::new(AtomicU64::new(0));
1193        std::thread::scope(|s| {
1194            for i in 0..16u64 {
1195                let (disk, ledger, landed, refused) = (&disk, &ledger, &landed, &refused);
1196                s.spawn(move || {
1197                    match push(disk, ledger, 160 * MIB, MIB * (1 + i % 3)) {
1198                        Ok(()) => landed.fetch_add(1, Ordering::SeqCst),
1199                        Err(_) => refused.fetch_add(1, Ordering::SeqCst),
1200                    };
1201                });
1202            }
1203        });
1204        let (landed, refused) = (
1205            landed.load(Ordering::SeqCst),
1206            refused.load(Ordering::SeqCst),
1207        );
1208        assert_eq!(landed + refused, 16);
1209        assert!(
1210            landed >= 1 && refused >= 1,
1211            "landed {landed}, refused {refused}"
1212        );
1213        assert!(
1214            disk.lowest() >= (2 * GIB) as i64,
1215            "the floor was crossed: lowest free {} B",
1216            disk.lowest()
1217        );
1218        assert_eq!(
1219            (ledger.reserved(), ledger.holds()),
1220            (0, 0),
1221            "every promise came back"
1222        );
1223    }
1224
1225    /// ★ Fair: a big push is cut at the small-push lane, and a small push
1226    /// beside it lands in the room the lane kept.
1227    #[test]
1228    fn a_small_push_lands_beside_a_big_one_cut_at_the_lane() {
1229        let disk = FakeDisk::new(10 * GIB, 3 * GIB);
1230        let ledger = ledger(&disk, 2 * GIB, 256 * MIB);
1231        let big = ledger.hold(PUSH_COPIES);
1232        let mut sent = 0;
1233        let mut cut = None;
1234        for _ in 0..10_000 {
1235            match big.admit(LEASE_BYTES) {
1236                Ok(g) => {
1237                    disk.write(g);
1238                    big.realize(g);
1239                    sent += g;
1240                }
1241                Err(line) => {
1242                    cut = Some(line);
1243                    break;
1244                }
1245            }
1246        }
1247        let line = cut.expect("a big push into 1 GiB of room is cut");
1248        assert!(
1249            line.starts_with(FULL_NOW) && line.contains("does not fit"),
1250            "{line}"
1251        );
1252        let room = disk.free.load(Ordering::SeqCst) as u64 - 2 * GIB - ledger.reserved();
1253        assert!(
1254            room >= 250 * MIB,
1255            "the big push left the lane: {} free above the floor after {} sent",
1256            human(room),
1257            human(sent)
1258        );
1259        let small = ledger.hold(PUSH_COPIES);
1260        let mut got = 0;
1261        for _ in 0..1_000 {
1262            if got == 8 * MIB {
1263                break;
1264            }
1265            got += small
1266                .admit(8 * MIB - got)
1267                .expect("a small push lands in the lane");
1268        }
1269        assert_eq!(got, 8 * MIB);
1270    }
1271
1272    /// A cancelled transfer — the client hung up, the ingest failed — gives
1273    /// back its whole promise when its hold drops.
1274    #[test]
1275    fn a_dropped_hold_gives_back_what_it_promised() {
1276        let disk = FakeDisk::new(10 * GIB, 5 * GIB);
1277        let ledger = ledger(&disk, 2 * GIB, 0);
1278        let hold = ledger.hold(PUSH_COPIES);
1279        hold.admit(100 * MIB).unwrap();
1280        hold.realize(40 * MIB);
1281        assert_eq!(
1282            (ledger.reserved(), ledger.holds()),
1283            (hold.promise_for(100 * MIB) - 40 * MIB, 1)
1284        );
1285        drop(hold);
1286        assert_eq!((ledger.reserved(), ledger.holds()), (0, 0));
1287    }
1288
1289    /// The up-front line counts the promises: a door that would let a push
1290    /// in when everything above the floor is already promised says so before
1291    /// the first byte.
1292    #[test]
1293    fn the_up_front_check_says_full_when_the_room_is_all_promised() {
1294        let disk = FakeDisk::new(10 * GIB, 3 * GIB);
1295        let ledger = ledger(&disk, 2 * GIB, 0);
1296        assert_eq!(ledger.full_before_the_first_byte(), None);
1297        let hold = ledger.hold(PUSH_COPIES);
1298        hold.admit(250 * MIB).unwrap();
1299        assert_eq!(
1300            ledger.full_before_the_first_byte(),
1301            None,
1302            "~149 MiB of room left"
1303        );
1304        hold.admit(LEASE_BYTES).unwrap();
1305        let rest = hold.admit(GIB).unwrap();
1306        assert!(rest > 0);
1307        let line = ledger
1308            .full_before_the_first_byte()
1309            .expect("all of it promised");
1310        assert!(
1311            line.starts_with(FULL_NOW) && line.contains("promised to pushes already arriving"),
1312            "{line}"
1313        );
1314        drop(hold);
1315        assert_eq!(ledger.full_before_the_first_byte(), None);
1316    }
1317
1318    /// The door asks the same ledger: under the floor it is the floor's own
1319    /// (final) line, above it with the room all promised it is [`FULL_NOW`]'s,
1320    /// and the pushes' promises given back open it again.
1321    #[test]
1322    fn the_disk_gate_refuses_up_front_when_the_room_is_all_promised() {
1323        use crate::plan::{PlanGate, PlanVerdict};
1324        let disk = FakeDisk::new(10 * GIB, 3 * GIB);
1325        let ledger = ledger(&disk, 2 * GIB, 0);
1326        let gate = crate::headroom::DiskHeadroomGate::over(Arc::clone(&ledger));
1327        assert_eq!(gate.pack_quota("team"), PlanVerdict::Allow);
1328        let hold = ledger.hold(PUSH_COPIES);
1329        // Bounded, so a ledger that never says no fails here rather than spins.
1330        for _ in 0..10_000 {
1331            if !hold.admit(LEASE_BYTES).is_ok_and(|g| g == LEASE_BYTES) {
1332                break;
1333            }
1334        }
1335        match gate.pack_quota("team") {
1336            PlanVerdict::Refuse(line) => assert!(line.starts_with(FULL_NOW), "{line}"),
1337            other => panic!("everything above the floor is promised: {other:?}"),
1338        }
1339        assert_eq!(
1340            gate.headroom().unwrap().free,
1341            3 * GIB,
1342            "Admin.Capacity reads the disk, not the promises"
1343        );
1344        drop(hold);
1345        assert_eq!(
1346            gate.pack_quota("team"),
1347            PlanVerdict::Allow,
1348            "the promise came back"
1349        );
1350        disk.write(GIB + 1);
1351        match gate.pack_quota("team") {
1352            PlanVerdict::Refuse(line) => {
1353                assert!(line.starts_with(crate::headroom::AT_FLOOR), "{line}")
1354            }
1355            other => panic!("under the floor: {other:?}"),
1356        }
1357    }
1358
1359    /// A disk that cannot be measured refuses — mid-stream too.
1360    #[test]
1361    fn an_unmeasurable_disk_refuses_the_next_chunk() {
1362        let disk = FakeDisk::new(10 * GIB, 5 * GIB);
1363        let ledger = ledger(&disk, 2 * GIB, 0);
1364        let hold = ledger.hold(PUSH_COPIES);
1365        hold.admit(MIB).unwrap();
1366        disk.fail.store(true, Ordering::SeqCst);
1367        let line = hold.admit(MIB).unwrap_err();
1368        assert!(line.contains("could not check its storage"), "{line}");
1369        assert_eq!(ledger.reserved(), 0);
1370    }
1371
1372    /// An LFS object declares its size: all of it, or a refusal.
1373    #[test]
1374    fn a_declared_size_is_admitted_whole_or_refused() {
1375        let disk = FakeDisk::new(10 * GIB, 3 * GIB);
1376        let ledger = ledger(&disk, 2 * GIB, 0);
1377        let fits = ledger.hold(PUSH_COPIES);
1378        fits.admit_all(200 * MIB)
1379            .expect("700 MiB (3× + the filesystem's slack) of 1 GiB");
1380        let no = ledger.hold(PUSH_COPIES);
1381        let line = no
1382            .admit_all(200 * MIB)
1383            .expect_err("700 MiB more does not fit in 324 MiB");
1384        assert!(line.starts_with(FULL_NOW), "{line}");
1385        assert_eq!(
1386            ledger.reserved(),
1387            fits.promise_for(200 * MIB),
1388            "the refused one holds nothing, the other its whole promise"
1389        );
1390    }
1391
1392    #[test]
1393    fn enospc_and_edquot_are_out_of_space_and_a_plain_cancel_is_not() {
1394        assert!(is_out_of_space(&std::io::Error::from_raw_os_error(28)));
1395        assert!(is_out_of_space(&std::io::Error::from_raw_os_error(122)));
1396        assert!(is_out_of_space(&std::io::Error::from(
1397            std::io::ErrorKind::StorageFull
1398        )));
1399        assert!(!is_out_of_space(&std::io::Error::from_raw_os_error(125)));
1400    }
1401}