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)));
}
}