use std::path::{Path, PathBuf};
use std::sync::{Arc, Mutex, MutexGuard};
use crate::headroom::{human, DiskFloor};
const MIB: u64 = 1024 * 1024;
pub const PUSH_COPIES: u64 = 3;
pub const PUSH_FS_SLACK_PERMILLE: u64 = 167;
pub const PUSH_RESERVE_PERMILLE: u64 = PUSH_COPIES * (1000 + PUSH_FS_SLACK_PERMILLE);
pub const PROBE_BYTES: u64 = 33 * MIB;
pub const PROBE_FROM: u64 = SMALL_PUSH_BYTES;
pub const SLACK_MARGIN_PERMILLE: u64 = 50;
pub const UNMEASURED_SLACK_PERMILLE: u64 = 1000 + SLACK_MARGIN_PERMILLE;
pub const PROBE_NEEDS_FREE: u64 = 4 * PROBE_BYTES;
pub const PROBE_FILE_PREFIX: &str = ".gunnar-space-probe-";
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub struct FsSlack {
pub measured_permille: Option<u64>,
pub slack_permille: u64,
}
impl Default for FsSlack {
fn default() -> Self {
FsSlack { measured_permille: None, slack_permille: PUSH_FS_SLACK_PERMILLE }
}
}
impl FsSlack {
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 }
}
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())),
}
}
pub fn reserve_permille(&self, copies: u64) -> u64 {
copies.max(1).saturating_mul(1000 + self.slack_permille)
}
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,
),
}
}
}
fn times(permille: u64) -> String {
format!("{}.{:02}×", permille / 1000, (permille % 1000) / 10)
}
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)
}
pub const SMALL_PUSH_BYTES: u64 = 16 * MIB;
pub const SMALL_PUSH_LANE: u64 = 256 * MIB;
pub const LEASE_BYTES: u64 = 4 * MIB;
pub const MIN_ROOM: u64 = PUSH_RESERVE_PERMILLE * 64 * 1024 / 1000;
pub const FULL_NOW: &str = "the data set is full right now";
pub trait SpaceProbe: Send + Sync + std::fmt::Debug {
fn measure(&self) -> std::io::Result<(u64, u64)>;
}
#[derive(Debug, Clone)]
pub struct Statvfs(pub PathBuf);
impl SpaceProbe for Statvfs {
fn measure(&self) -> std::io::Result<(u64, u64)> {
statvfs_bytes(&self.0)
}
}
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),
))
}
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub struct Standing {
pub free: u64,
pub total: u64,
pub floor: u64,
pub reserved: u64,
}
impl Standing {
pub fn room(&self) -> u64 {
self.free
.saturating_sub(self.floor)
.saturating_sub(self.reserved)
}
}
#[derive(Debug, Default)]
struct LedgerState {
reserved: u64,
holds: usize,
refused: u64,
}
#[derive(Debug)]
pub struct SpaceLedger {
root: PathBuf,
floor: DiskFloor,
probe: Arc<dyn SpaceProbe>,
lane: u64,
slack: FsSlack,
state: Mutex<LedgerState>,
}
impl SpaceLedger {
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)
}
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()),
}
}
pub fn with_fs_slack(mut self, slack: FsSlack) -> Self {
self.slack = slack;
self
}
pub fn fs_slack(&self) -> FsSlack {
self.slack
}
pub fn with_small_lane(mut self, bytes: u64) -> Self {
self.lane = bytes;
self
}
pub fn root(&self) -> &Path {
&self.root
}
pub fn floor(&self) -> DiskFloor {
self.floor
}
pub fn small_lane(&self) -> u64 {
self.lane
}
pub fn probe(&self) -> &Arc<dyn SpaceProbe> {
&self.probe
}
fn lock(&self) -> MutexGuard<'_, LedgerState> {
self.state.lock().unwrap_or_else(|poisoned| poisoned.into_inner())
}
pub fn reserved(&self) -> u64 {
self.lock().reserved
}
pub fn holds(&self) -> usize {
self.lock().holds
}
pub fn refused(&self) -> u64 {
self.lock().refused
}
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,
})
}
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()),
}
}
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"
))
}
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 {
held: u64,
admitted: u64,
refusal: Option<String>,
}
#[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())
}
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)
}
pub fn admit_all(&self, bytes: u64) -> Result<(), String> {
let mut left = bytes;
while left > 0 {
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);
}
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);
}
pub fn held(&self) -> u64 {
self.lock().held
}
pub fn admitted(&self) -> u64 {
self.lock().admitted
}
pub fn refusal(&self) -> Option<String> {
self.lock().refusal.clone()
}
pub fn copies(&self) -> u64 {
self.copies
}
pub fn per_mille(&self) -> u64 {
self.ledger.slack.reserve_permille(self.copies)
}
pub fn promise_for(&self, wire: u64) -> u64 {
wire.saturating_mul(self.per_mille()).div_ceil(1000)
}
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
}
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);
}
}
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;
#[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))
}
}
#[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}");
}
#[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")));
}
#[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),
)
}
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); hold.realize(bytes);
disk.write(bytes); hold.realize(bytes);
disk.delete(spooled); drop(hold);
Ok(())
}
#[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}");
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);
}
#[test]
fn eight_pushes_into_room_for_three_land_three_or_more_and_refuse_the_rest() {
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];
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));
}
#[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");
}
#[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);
}
#[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));
}
#[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);
}
#[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);
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:?}"),
}
}
#[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);
}
#[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)));
}
}