crates/gunnar-cli/tests/push_space_admission.rs
#![cfg(feature = "server")]
//! **Concurrent pushes near the floor are refused BY NAME — never failed by
//! the disk, and the floor is never crossed.** (cov-13, 2026-10-06)
//!
//! The UpCloud mockup's grow run pushed eight 512 MiB packs at once into gunnar
//! main's 19.9 GiB set with its 2 GiB floor. Every one was admitted at the door
//! — the set was above its floor — and together they ran /data out of space
//! while still receiving: the store failed them as `uring: op … failed:
//! Operation canceled`, the client read "the repository is unavailable; try
//! again", and the floor the gate exists to keep was crossed.
//!
//! This is that run, small: stock `git push` over SSH into a live server, eight
//! at once, each pack 96 MiB (over the 64 MiB spool line, the path cov-13
//! filled) into a data set of 1.25 GiB with a 64 MiB floor — room for about
//! two of them at their peak. It asserts, in order:
//!
//! 1. one push alone occupies at most 3 × its pack at the peak and about 2 ×
//! after it lands (spool + log + store, then the spool dropped) — MEASURED
//! and printed, the factor the space ledger reserves;
//! 2. every one of the eight either landed or was refused with
//! [`gunnar_ssh::FULL_NOW`]'s line — no ENOSPC, no io error, no "try again"
//! that does not say why;
//! 3. some landed and some were refused;
//! 4. the data set's free bytes never went under the floor;
//! 5. no spool file is left anywhere, and a refused push left no `Pack` frame
//! in the transaction log — nothing of it was kept;
//! 6. a small push still lands afterwards.
//!
//! # Two disks
//!
//! * `GUNNAR_TEST_SMALL_FS=/mnt/x` — a real filesystem with a size cap (a
//! 1.25 GiB tmpfs: `sudo mount -t tmpfs -o size=1280m,uid=$(id -u) tmpfs
//! /mnt/x`). The disk really fills and ENOSPC is real; this is the RED the
//! fix was written against.
//! * otherwise a VIRTUAL disk of the same size over a temp dir
//! (`common::VirtualDisk`, handed to the server as its space probe): the
//! floor and the ledger read it, the bytes land on the real disk. No
//! privileges, so it runs everywhere; its RED is the floor crossed.
mod common;
use std::path::{Path, PathBuf};
use std::process::Command;
use std::sync::atomic::{AtomicBool, AtomicU64, Ordering};
use std::sync::Arc;
use std::time::Duration;
use common::{tools_present, Harness, HarnessOptions, VirtualDisk};
use gunnar_core::txlog::Meta;
use gunnar_core::{Grant, HashKind, Level, NewRepo, RepoRef};
const NS: &str = "team";
const MIB: u64 = 1 << 20;
/// cov-13's eight.
const PUSHES: usize = 8;
/// Over `SPOOL_ABOVE` (64 MiB): the spooled path, the one cov-13 filled.
const BLOB_BYTES: usize = 96 << 20;
const FLOOR: u64 = 64 * MIB;
/// The virtual disk's size, and the size the small filesystem is checked to have.
const CAPACITY: u64 = 1280 * MIB;
enum Disk {
/// A filesystem of its own with a size cap — `GUNNAR_TEST_SMALL_FS`.
Small(PathBuf),
/// [`VirtualDisk`] of [`CAPACITY`] over a temp dir.
Virtual(PathBuf),
}
impl Disk {
fn root(&self) -> &Path {
match self {
Disk::Small(p) | Disk::Virtual(p) => p,
}
}
fn free(&self) -> u64 {
match self {
Disk::Small(p) => gunnar_ssh::space::statvfs_bytes(p).expect("statvfs").0,
Disk::Virtual(p) => CAPACITY.saturating_sub(VirtualDisk::used(p)),
}
}
fn name(&self) -> &'static str {
match self {
Disk::Small(_) => "small filesystem",
Disk::Virtual(_) => "virtual disk",
}
}
}
/// Samples the data set's free bytes until stopped and keeps the lowest.
struct Lowest {
stop: Arc<AtomicBool>,
lowest: Arc<AtomicU64>,
thread: Option<std::thread::JoinHandle<()>>,
}
impl Lowest {
fn watch(disk: &Disk) -> Lowest {
let stop = Arc::new(AtomicBool::new(false));
let lowest = Arc::new(AtomicU64::new(disk.free()));
let root = disk.root().to_path_buf();
let small = matches!(disk, Disk::Small(_));
let (s, l) = (Arc::clone(&stop), Arc::clone(&lowest));
let thread = std::thread::spawn(move || {
let disk = if small {
Disk::Small(root)
} else {
Disk::Virtual(root)
};
while !s.load(Ordering::SeqCst) {
l.fetch_min(disk.free(), Ordering::SeqCst);
std::thread::sleep(Duration::from_millis(2));
}
});
Lowest {
stop,
lowest,
thread: Some(thread),
}
}
fn finish(mut self) -> u64 {
self.stop.store(true, Ordering::SeqCst);
if let Some(t) = self.thread.take() {
t.join().expect("the sampler");
}
self.lowest.load(Ordering::SeqCst)
}
}
/// `git push` from `work` to `url`, as the harness runs git, callable from a
/// thread of its own.
fn push(ssh: &str, home: &Path, work: &Path, url: &str) -> (bool, String) {
let out = Command::new("git")
.current_dir(work)
.env("GIT_SSH_COMMAND", ssh)
.env("GIT_TERMINAL_PROMPT", "0")
.env("LC_ALL", "C")
.env("GIT_CONFIG_NOSYSTEM", "1")
.env("HOME", home)
.args(["push", url, "HEAD:refs/heads/main"])
.output()
.expect("git is on PATH");
(
out.status.success(),
String::from_utf8_lossy(&out.stderr).into_owned(),
)
}
/// Incompressible bytes, so the pack is the size of the blob.
fn noise(len: usize, mut state: u64) -> Vec<u8> {
let mut out = vec![0u8; len];
for chunk in out.chunks_mut(8) {
state ^= state << 13;
state ^= state >> 7;
state ^= state << 17;
let bytes = state.to_le_bytes();
chunk.copy_from_slice(&bytes[..chunk.len()]);
}
out
}
/// Every `incoming-*` spool file under `root`.
fn spool_files(root: &Path) -> Vec<PathBuf> {
let mut found = Vec::new();
let mut stack = vec![root.to_path_buf()];
while let Some(dir) = stack.pop() {
let Ok(entries) = std::fs::read_dir(&dir) else {
continue;
};
for entry in entries.flatten() {
let path = entry.path();
if path.is_dir() {
stack.push(path);
} else if entry.file_name().to_string_lossy().starts_with("incoming-") {
found.push(path);
}
}
}
found
}
fn add_repo(harness: &Harness, name: &str) {
let path = RepoRef::new(NS, name).expect("legal");
let row = harness
.stores
.catalog()
.create(NewRepo::new(path.clone()).hash_kind(HashKind::Sha256))
.expect("create the repository");
harness.stores.repo_store(&row).expect("materialise it");
harness
.stores
.auth()
.grant(Grant::repo("tester", path, Level::Write))
.expect("grant write");
}
fn gib(bytes: u64) -> String {
gunnar_ssh::headroom::human(bytes)
}
#[test]
fn eight_concurrent_pushes_near_the_floor_are_refused_by_name_and_never_cross_it() {
if !tools_present() {
eprintln!("skipping: no git or ssh on PATH");
return;
}
// ── the disk ─────────────────────────────────────────────────────────────
let _virtual_dir;
let disk = match std::env::var_os("GUNNAR_TEST_SMALL_FS") {
Some(mount) => {
let unique = std::time::SystemTime::now()
.duration_since(std::time::UNIX_EPOCH)
.unwrap()
.as_nanos();
let root =
PathBuf::from(mount).join(format!("push-space-{}-{unique}", std::process::id()));
std::fs::create_dir_all(&root).expect("a directory on the small filesystem");
let (free, total) = gunnar_ssh::space::statvfs_bytes(&root).expect("statvfs");
assert!(
total <= CAPACITY + 64 * MIB && free > 1024 * MIB,
"GUNNAR_TEST_SMALL_FS must be a filesystem of about {} with nearly all of it free \
(it is {} with {} free): the test needs eight pushes NOT to fit, and two to",
gib(CAPACITY),
gib(total),
gib(free)
);
Disk::Small(root)
}
None => {
let dir = tempfile::tempdir().expect("tempdir");
let root = dir.path().join("data");
std::fs::create_dir_all(&root).expect("data root");
_virtual_dir = dir;
Disk::Virtual(root)
}
};
let data_root: &'static Path = Box::leak(disk.root().to_path_buf().into_boxed_path());
let harness = Harness::start_with_options(
NS,
"p0",
HashKind::Sha256,
HarnessOptions {
disk_floor: Some(gunnar_ssh::DiskFloor {
percent: 0,
bytes: FLOOR,
}),
data_root: Some(data_root),
virtual_capacity: matches!(disk, Disk::Virtual(_)).then_some(CAPACITY),
..HarnessOptions::default()
},
);
for i in 1..=PUSHES {
add_repo(&harness, &format!("p{i}"));
}
// ── one commit holding a 96 MiB incompressible blob, packed once ─────────
let work = harness.dir.path().join("work");
std::fs::create_dir_all(&work).unwrap();
harness
.git(
&work,
&[
"init",
"--quiet",
"--object-format=sha256",
"--initial-branch=main",
".",
],
)
.expect_ok("git init");
std::fs::write(
work.join("big.bin"),
noise(BLOB_BYTES, 0x9E37_79B9_7F4A_7C15),
)
.unwrap();
harness.git(&work, &["add", "big.bin"]).expect_ok("git add");
harness
.git(&work, &["commit", "--quiet", "-m", "96 MiB"])
.expect_ok("git commit");
// Packed here, so eight `pack-objects` reuse it instead of deflating 96 MiB each.
harness
.git(&work, &["repack", "-adq"])
.expect_ok("git repack");
let ssh = harness.ssh_command();
let home = harness.dir.path().to_path_buf();
// ── 1. what ONE push occupies: the factor the ledger reserves ────────────
let before = disk.free();
let watch = Lowest::watch(&disk);
let (ok, stderr) = push(&ssh, &home, &work, &harness.url(NS, "p0"));
let lowest = watch.finish();
assert!(
ok,
"one 96 MiB push alone fits ({} free): {stderr}",
gib(before)
);
let after = disk.free();
let pack = BLOB_BYTES as f64;
let peak = before.saturating_sub(lowest) as f64 / pack;
let landed = before.saturating_sub(after) as f64 / pack;
eprintln!(
"MEASURED ({}): one push of a {} blob: peak {peak:.2}× its bytes on the data set, {landed:.2}× after it landed",
disk.name(),
gib(BLOB_BYTES as u64)
);
// The ledger's promise per wire byte — the copies AND the filesystem's slack — must cover the
// measured peak. The slack is MEASURED on the data set at start (round 5.2: XFS doubles a
// growing file — 3.24× on dwarf's 256 KiB stripe, 3.58× on oden's 1 MiB stripe — over the
// fixed 3.5× r5.1 promised; tmpfs 3.00×). The server measured this root with the same probe.
let slack = gunnar_ssh::space::FsSlack::of(data_root).0;
let reserved = slack.reserve_permille(gunnar_ssh::PUSH_COPIES) as f64 / 1000.0;
eprintln!("MEASURED ({}): the filesystem's slack {:?} ‰ — the ledger reserves {reserved:.2}× per wire byte", disk.name(), slack.measured_permille);
assert!(peak <= reserved, "a push occupied {peak:.2}× its bytes at the peak; the ledger reserves {reserved:.2}× ({}× + {} ‰ filesystem slack)", gunnar_ssh::PUSH_COPIES, slack.slack_permille);
let landed_bound = 2.0 * (1000 + slack.slack_permille) as f64 / 1000.0;
assert!(landed <= landed_bound, "a landed push holds {landed:.2}× its bytes (store + log); expected about 2× (+ the filesystem's slack: ≤ {landed_bound:.2}×)");
// ── 2–4. eight at once ───────────────────────────────────────────────────
let room = disk.free().saturating_sub(FLOOR);
assert!(
(PUSHES as u64) * 3 * BLOB_BYTES as u64 > 2 * room,
"the burst must NOT fit: {} of room above the floor",
gib(room)
);
let watch = Lowest::watch(&disk);
let outcomes: Vec<(usize, bool, String)> = std::thread::scope(|s| {
let handles: Vec<_> = (1..=PUSHES)
.map(|i| {
let (ssh, home, work, url) =
(&ssh, &home, &work, harness.url(NS, &format!("p{i}")));
s.spawn(move || {
let (ok, stderr) = push(ssh, home, work, &url);
(i, ok, stderr)
})
})
.collect();
handles
.into_iter()
.map(|h| h.join().expect("a push thread"))
.collect()
});
let lowest = watch.finish();
for (i, ok, stderr) in &outcomes {
assert!(
*ok || stderr.contains(gunnar_ssh::FULL_NOW),
"push p{i} failed WITHOUT the named refusal — the disk failed it, not the admission:\n{stderr}"
);
}
let refused: Vec<usize> = outcomes
.iter()
.filter(|(_, ok, _)| !ok)
.map(|(i, _, _)| *i)
.collect();
let landed: Vec<usize> = outcomes
.iter()
.filter(|(_, ok, _)| *ok)
.map(|(i, _, _)| *i)
.collect();
eprintln!(
"{}: {} landed {landed:?}, {} refused {refused:?}; lowest free {} (floor {})",
disk.name(),
landed.len(),
refused.len(),
gib(lowest),
gib(FLOOR)
);
if let Some((_, _, stderr)) = outcomes.iter().find(|(_, ok, _)| !ok) {
eprintln!("what a refused push was told:\n{stderr}");
}
assert!(
lowest >= FLOOR,
"the data set went UNDER its floor: {} free at the lowest, floor {} ({} landed, {} refused)",
gib(lowest),
gib(FLOOR),
landed.len(),
refused.len()
);
assert!(
!refused.is_empty(),
"eight 96 MiB pushes into {} of room: some must be refused",
gib(room)
);
assert!(!landed.is_empty(), "room for at least one: none landed");
// ── 5. nothing of a refused push was kept ────────────────────────────────
let left = spool_files(disk.root());
assert!(left.is_empty(), "spool files left behind: {left:?}");
let log = harness.stores.txlog();
let entries = log.range(log.first_seq(), log.head_seq()).expect("the log");
for i in &refused {
let key = format!("{NS}/p{i}");
assert!(
!entries
.iter()
.any(|e| matches!(e.meta(), Ok(Meta::Pack { ref repo, .. }) if *repo == key)),
"refused push p{i} left a Pack frame in the transaction log"
);
}
// ── 6. a small push still lands ──────────────────────────────────────────
std::fs::write(work.join("small.txt"), "a small change after the burst\n").unwrap();
harness
.git(&work, &["add", "small.txt"])
.expect_ok("git add");
harness
.git(&work, &["commit", "--quiet", "-m", "small"])
.expect_ok("git commit");
let (ok, stderr) = push(&ssh, &home, &work, &harness.url(NS, "p0"));
assert!(ok, "a small push after the burst must land: {stderr}");
}