1use std::path::{Path, PathBuf};
83use std::sync::{Arc, Mutex, MutexGuard};
84
85use crate::headroom::{human, DiskFloor};
86
87const MIB: u64 = 1024 * 1024;
88
89pub const PUSH_COPIES: u64 = 3;
92
93pub const PUSH_FS_SLACK_PERMILLE: u64 = 167;
100
101pub const PUSH_RESERVE_PERMILLE: u64 = PUSH_COPIES * (1000 + PUSH_FS_SLACK_PERMILLE);
105
106pub const PROBE_BYTES: u64 = 33 * MIB;
119
120pub const PROBE_FROM: u64 = SMALL_PUSH_BYTES;
123
124pub const SLACK_MARGIN_PERMILLE: u64 = 50;
126
127pub const UNMEASURED_SLACK_PERMILLE: u64 = 1000 + SLACK_MARGIN_PERMILLE;
131
132pub const PROBE_NEEDS_FREE: u64 = 4 * PROBE_BYTES;
135
136pub const PROBE_FILE_PREFIX: &str = ".gunnar-space-probe-";
139
140#[derive(Debug, Clone, Copy, PartialEq, Eq)]
142pub struct FsSlack {
143 pub measured_permille: Option<u64>,
146 pub slack_permille: u64,
150}
151
152impl Default for FsSlack {
153 fn default() -> Self {
155 FsSlack { measured_permille: None, slack_permille: PUSH_FS_SLACK_PERMILLE }
156 }
157}
158
159impl FsSlack {
160 pub fn from_measured(measured_permille: Option<u64>) -> FsSlack {
162 let slack_permille = match measured_permille {
163 Some(m) => m.saturating_add(SLACK_MARGIN_PERMILLE).max(PUSH_FS_SLACK_PERMILLE),
164 None => UNMEASURED_SLACK_PERMILLE,
165 };
166 FsSlack { measured_permille, slack_permille }
167 }
168
169 pub fn of(root: &Path) -> (FsSlack, Option<String>) {
171 match measure_fs_slack(root) {
172 Ok(m) => (FsSlack::from_measured(Some(m)), None),
173 Err(e) => (FsSlack::from_measured(None), Some(e.to_string())),
174 }
175 }
176
177 pub fn reserve_permille(&self, copies: u64) -> u64 {
179 copies.max(1).saturating_mul(1000 + self.slack_permille)
180 }
181
182 pub fn words(&self, root: &Path, unmeasured_because: Option<&str>) -> String {
185 let reserve = times(self.reserve_permille(PUSH_COPIES));
186 match self.measured_permille {
187 Some(m) => format!(
188 "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}",
189 root = root.display(),
190 held = times(1000 + m),
191 probe = human(PROBE_BYTES),
192 margin = SLACK_MARGIN_PERMILLE,
193 floor = times(PUSH_RESERVE_PERMILLE),
194 ),
195 None => format!(
196 "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",
197 root = root.display(),
198 why = unmeasured_because.unwrap_or("no reason given"),
199 margin = SLACK_MARGIN_PERMILLE,
200 ),
201 }
202 }
203}
204
205fn times(permille: u64) -> String {
207 format!("{}.{:02}×", permille / 1000, (permille % 1000) / 10)
208}
209
210pub fn measure_fs_slack(dir: &Path) -> std::io::Result<u64> {
216 use std::io::Write as _;
217 use std::os::unix::fs::MetadataExt as _;
218 if let Ok(entries) = std::fs::read_dir(dir) {
219 for e in entries.flatten() {
220 if e.file_name().to_string_lossy().starts_with(PROBE_FILE_PREFIX) {
221 let _ = std::fs::remove_file(e.path());
222 }
223 }
224 }
225 let (free, _) = statvfs_bytes(dir)?;
226 if free < PROBE_NEEDS_FREE {
227 return Err(std::io::Error::other(format!("{} free, and the probe needs {}", human(free), human(PROBE_NEEDS_FREE))));
228 }
229 static SEQ: std::sync::atomic::AtomicU64 = std::sync::atomic::AtomicU64::new(0);
230 let path = dir.join(format!("{PROBE_FILE_PREFIX}{}-{}", std::process::id(), SEQ.fetch_add(1, std::sync::atomic::Ordering::Relaxed)));
231 struct Gone(PathBuf);
232 impl Drop for Gone {
233 fn drop(&mut self) {
234 let _ = std::fs::remove_file(&self.0);
235 }
236 }
237 let _gone = Gone(path.clone());
238 let mut file = std::fs::OpenOptions::new().write(true).create_new(true).open(&path)?;
239 let chunk: Vec<u8> = (0..MIB as usize).map(|i| (i.wrapping_mul(2_654_435_761) >> 13) as u8).collect();
240 let over = |file: &std::fs::File, written: u64| -> std::io::Result<u64> {
241 let held = file.metadata()?.blocks().saturating_mul(512);
242 Ok(held.saturating_sub(written).saturating_mul(1000) / written.max(1))
243 };
244 let (mut written, mut worst) = (0u64, 0u64);
245 while written < PROBE_BYTES {
246 file.write_all(&chunk)?;
247 written += MIB;
248 if written >= PROBE_FROM {
249 worst = worst.max(over(&file, written)?);
250 }
251 }
252 file.sync_data()?;
253 worst = worst.max(over(&file, written)?);
254 Ok(worst)
255}
256
257pub const SMALL_PUSH_BYTES: u64 = 16 * MIB;
260
261pub const SMALL_PUSH_LANE: u64 = 256 * MIB;
265
266pub const LEASE_BYTES: u64 = 4 * MIB;
268
269pub const MIN_ROOM: u64 = PUSH_RESERVE_PERMILLE * 64 * 1024 / 1000;
272
273pub const FULL_NOW: &str = "the data set is full right now";
279
280pub trait SpaceProbe: Send + Sync + std::fmt::Debug {
283 fn measure(&self) -> std::io::Result<(u64, u64)>;
285}
286
287#[derive(Debug, Clone)]
289pub struct Statvfs(pub PathBuf);
290
291impl SpaceProbe for Statvfs {
292 fn measure(&self) -> std::io::Result<(u64, u64)> {
293 statvfs_bytes(&self.0)
294 }
295}
296
297pub fn statvfs_bytes(root: &Path) -> std::io::Result<(u64, u64)> {
299 let vfs = rustix::fs::statvfs(root).map_err(std::io::Error::from)?;
300 let frsize = if vfs.f_frsize == 0 {
301 vfs.f_bsize
302 } else {
303 vfs.f_frsize
304 };
305 Ok((
306 vfs.f_bavail.saturating_mul(frsize),
307 vfs.f_blocks.saturating_mul(frsize),
308 ))
309}
310
311#[derive(Debug, Clone, Copy, PartialEq, Eq)]
313pub struct Standing {
314 pub free: u64,
316 pub total: u64,
318 pub floor: u64,
320 pub reserved: u64,
322}
323
324impl Standing {
325 pub fn room(&self) -> u64 {
327 self.free
328 .saturating_sub(self.floor)
329 .saturating_sub(self.reserved)
330 }
331}
332
333#[derive(Debug, Default)]
334struct LedgerState {
335 reserved: u64,
337 holds: usize,
339 refused: u64,
341}
342
343#[derive(Debug)]
347pub struct SpaceLedger {
348 root: PathBuf,
349 floor: DiskFloor,
350 probe: Arc<dyn SpaceProbe>,
351 lane: u64,
352 slack: FsSlack,
354 state: Mutex<LedgerState>,
355}
356
357impl SpaceLedger {
358 pub fn new(root: impl Into<PathBuf>, floor: DiskFloor) -> Self {
360 let root = root.into();
361 let probe = Arc::new(Statvfs(root.clone()));
362 Self::with_probe(root, floor, probe)
363 }
364
365 pub fn with_probe(root: impl Into<PathBuf>, floor: DiskFloor, probe: Arc<dyn SpaceProbe>) -> Self {
367 SpaceLedger {
368 root: root.into(),
369 floor,
370 probe,
371 lane: SMALL_PUSH_LANE,
372 slack: FsSlack::default(),
373 state: Mutex::new(LedgerState::default()),
374 }
375 }
376
377 pub fn with_fs_slack(mut self, slack: FsSlack) -> Self {
380 self.slack = slack;
381 self
382 }
383
384 pub fn fs_slack(&self) -> FsSlack {
386 self.slack
387 }
388
389 pub fn with_small_lane(mut self, bytes: u64) -> Self {
392 self.lane = bytes;
393 self
394 }
395
396 pub fn root(&self) -> &Path {
398 &self.root
399 }
400
401 pub fn floor(&self) -> DiskFloor {
403 self.floor
404 }
405
406 pub fn small_lane(&self) -> u64 {
408 self.lane
409 }
410
411 pub fn probe(&self) -> &Arc<dyn SpaceProbe> {
413 &self.probe
414 }
415
416 fn lock(&self) -> MutexGuard<'_, LedgerState> {
417 self.state.lock().unwrap_or_else(|poisoned| poisoned.into_inner())
418 }
419
420 pub fn reserved(&self) -> u64 {
422 self.lock().reserved
423 }
424
425 pub fn holds(&self) -> usize {
427 self.lock().holds
428 }
429
430 pub fn refused(&self) -> u64 {
432 self.lock().refused
433 }
434
435 pub fn standing(&self) -> std::io::Result<Standing> {
437 let reserved = self.reserved();
438 self.measure_with(reserved)
439 }
440
441 fn measure_with(&self, reserved: u64) -> std::io::Result<Standing> {
442 let (free, total) = self.probe.measure()?;
443 Ok(Standing {
444 free,
445 total,
446 floor: self.floor.bytes_for(total),
447 reserved,
448 })
449 }
450
451 pub fn hold(self: &Arc<Self>, copies: u64) -> SpaceHold {
455 self.lock().holds += 1;
456 SpaceHold {
457 ledger: Arc::clone(self),
458 copies: copies.max(1),
459 state: Mutex::new(HoldState::default()),
460 }
461 }
462
463 pub fn full_before_the_first_byte(&self) -> Option<String> {
470 let standing = self.standing().ok()?;
471 if standing.free <= standing.floor || standing.reserved == 0 || standing.room() >= MIN_ROOM {
472 return None;
473 }
474 tracing::warn!(
475 target: "gunnar::space",
476 event = "push.space.full_up_front",
477 root = %self.root.display(),
478 free = %human(standing.free),
479 floor = %human(standing.floor),
480 disk_floor = %self.floor,
481 promised = %human(standing.reserved),
482 "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"
483 );
484 Some(format!(
485 "{FULL_NOW}: everything free is promised to pushes already arriving, so nothing was \
486 received. Try again in a while; the storage is grown when it fills"
487 ))
488 }
489
490 fn full_line(&self, standing: &Standing, own_held: u64, sent: u64, copies: u64, lane: u64) -> String {
493 let others = standing.reserved.saturating_sub(own_held);
494 let prealloc = match self.slack.measured_permille {
495 Some(m) => format!("measured at start to hold {} a growing file's bytes", times(1000 + m)),
496 None => "not measured, so a doubling is assumed".to_owned(),
497 };
498 tracing::warn!(
499 target: "gunnar::space",
500 event = "push.space.refused",
501 root = %self.root.display(),
502 free = %human(standing.free),
503 floor = %human(standing.floor),
504 disk_floor = %self.floor,
505 small_lane = %human(lane),
506 promised_to_others = %human(others),
507 sent = %human(sent),
508 copies,
509 reserve = %times(self.slack.reserve_permille(copies)),
510 filesystem = %prealloc,
511 "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"
512 );
513 format!(
514 "{FULL_NOW}: this push does not fit beside the pushes already arriving, so nothing of it \
515 was kept. Try again in a while; the storage is grown when it fills"
516 )
517 }
518
519 fn unmeasured_line(&self, err: &std::io::Error) -> String {
520 tracing::error!(
521 target: "gunnar::space",
522 event = "push.space.unmeasured",
523 root = %self.root.display(),
524 %err,
525 "the data set could not be measured: a push was refused rather than guessing there is room"
526 );
527 format!(
528 "{FULL_NOW}: Gunnar could not check its storage, so this push was refused rather than \
529 risk it. Nothing of it was kept; try again in a while"
530 )
531 }
532}
533
534#[derive(Debug, Default)]
535struct HoldState {
536 held: u64,
538 admitted: u64,
540 refusal: Option<String>,
542}
543
544#[derive(Debug)]
547pub struct SpaceHold {
548 ledger: Arc<SpaceLedger>,
549 copies: u64,
550 state: Mutex<HoldState>,
551}
552
553impl SpaceHold {
554 fn lock(&self) -> MutexGuard<'_, HoldState> {
555 self.state.lock().unwrap_or_else(|poisoned| poisoned.into_inner())
556 }
557
558 pub fn admit(&self, want: u64) -> Result<u64, String> {
563 let mut hold = self.lock();
564 if let Some(line) = &hold.refusal {
565 return Err(line.clone());
566 }
567 if want == 0 {
568 return Ok(0);
569 }
570 let ledger = &self.ledger;
571 let mut state = ledger.lock();
572 let standing = match ledger.measure_with(state.reserved) {
573 Ok(standing) => standing,
574 Err(err) => {
575 let line = ledger.unmeasured_line(&err);
576 Self::refuse(&mut hold, &mut state, line.clone());
577 return Err(line);
578 }
579 };
580 let big = hold.admitted.saturating_add(want) > SMALL_PUSH_BYTES;
581 let lane = if big { ledger.lane } else { 0 };
582 let granted = want.min(standing.room().saturating_sub(lane).saturating_mul(1000) / self.per_mille());
583 if granted == 0 {
584 let line = ledger.full_line(&standing, hold.held, hold.admitted, self.copies, lane);
585 Self::refuse(&mut hold, &mut state, line.clone());
586 return Err(line);
587 }
588 let promise = self.promise_for(granted);
589 state.reserved = state.reserved.saturating_add(promise);
590 hold.held = hold.held.saturating_add(promise);
591 hold.admitted = hold.admitted.saturating_add(granted);
592 Ok(granted)
593 }
594
595 pub fn admit_all(&self, bytes: u64) -> Result<(), String> {
598 let mut left = bytes;
599 while left > 0 {
600 let granted = self.admit(left)?;
603 if granted < left {
604 let mut hold = self.lock();
605 let mut state = self.ledger.lock();
606 let standing = self
607 .ledger
608 .measure_with(state.reserved)
609 .unwrap_or(Standing { free: 0, total: 0, floor: 0, reserved: state.reserved });
610 let lane = if bytes > SMALL_PUSH_BYTES { self.ledger.lane } else { 0 };
611 let line = self.ledger.full_line(&standing, hold.held, bytes, self.copies, lane);
612 Self::refuse(&mut hold, &mut state, line.clone());
613 return Err(line);
614 }
615 left -= granted;
616 }
617 Ok(())
618 }
619
620 fn refuse(hold: &mut HoldState, state: &mut LedgerState, line: String) {
621 state.reserved = state.reserved.saturating_sub(hold.held);
622 state.refused += 1;
623 hold.held = 0;
624 hold.refusal = Some(line);
625 }
626
627 pub fn realize(&self, bytes: u64) {
632 let mut hold = self.lock();
633 let given = bytes.min(hold.held);
634 if given == 0 {
635 return;
636 }
637 hold.held -= given;
638 let mut state = self.ledger.lock();
639 state.reserved = state.reserved.saturating_sub(given);
640 }
641
642 pub fn held(&self) -> u64 {
644 self.lock().held
645 }
646
647 pub fn admitted(&self) -> u64 {
649 self.lock().admitted
650 }
651
652 pub fn refusal(&self) -> Option<String> {
654 self.lock().refusal.clone()
655 }
656
657 pub fn copies(&self) -> u64 {
659 self.copies
660 }
661
662 pub fn per_mille(&self) -> u64 {
665 self.ledger.slack.reserve_permille(self.copies)
666 }
667
668 pub fn promise_for(&self, wire: u64) -> u64 {
671 wire.saturating_mul(self.per_mille()).div_ceil(1000)
672 }
673
674 pub fn full_now(&self) -> String {
679 let mut hold = self.lock();
680 if let Some(line) = &hold.refusal {
681 return line.clone();
682 }
683 let mut state = self.ledger.lock();
684 let line = match self.ledger.measure_with(state.reserved) {
685 Ok(standing) => self.ledger.full_line(&standing, hold.held, hold.admitted, self.copies, 0),
686 Err(err) => self.ledger.unmeasured_line(&err),
687 };
688 Self::refuse(&mut hold, &mut state, line.clone());
689 line
690 }
691
692 pub fn at_the_floor(&self, freed: u64) -> bool {
698 match self.ledger.probe.measure() {
699 Ok((free, total)) => free.saturating_sub(freed) <= self.ledger.floor.bytes_for(total).saturating_add(MIN_ROOM),
700 Err(_) => true,
701 }
702 }
703}
704
705impl Drop for SpaceHold {
706 fn drop(&mut self) {
707 let held = std::mem::take(&mut self.lock().held);
708 let mut state = self.ledger.lock();
709 state.reserved = state.reserved.saturating_sub(held);
710 state.holds = state.holds.saturating_sub(1);
711 }
712}
713
714pub fn is_out_of_space(err: &std::io::Error) -> bool {
720 use std::io::ErrorKind;
721 matches!(err.kind(), ErrorKind::StorageFull | ErrorKind::QuotaExceeded)
722 || matches!(err.raw_os_error(), Some(28) | Some(122))
723}
724
725#[cfg(test)]
726mod tests {
727 use super::*;
728 use std::sync::atomic::{AtomicI64, AtomicU64, Ordering};
729
730 const GIB: u64 = 1024 * MIB;
731
732 #[derive(Debug)]
736 struct FakeDisk {
737 free: AtomicI64,
738 total: u64,
739 lowest: AtomicI64,
740 fail: std::sync::atomic::AtomicBool,
741 }
742
743 impl FakeDisk {
744 fn new(total: u64, free: u64) -> Arc<FakeDisk> {
745 Arc::new(FakeDisk {
746 free: AtomicI64::new(free as i64),
747 total,
748 lowest: AtomicI64::new(free as i64),
749 fail: Default::default(),
750 })
751 }
752 fn write(&self, bytes: u64) {
753 let now = self.free.fetch_sub(bytes as i64, Ordering::SeqCst) - bytes as i64;
754 self.lowest.fetch_min(now, Ordering::SeqCst);
755 }
756 fn delete(&self, bytes: u64) {
757 self.free.fetch_add(bytes as i64, Ordering::SeqCst);
758 }
759 fn lowest(&self) -> i64 {
760 self.lowest.load(Ordering::SeqCst)
761 }
762 }
763
764 impl SpaceProbe for FakeDisk {
765 fn measure(&self) -> std::io::Result<(u64, u64)> {
766 if self.fail.load(Ordering::SeqCst) {
767 return Err(std::io::Error::other("the mount went away"));
768 }
769 Ok((self.free.load(Ordering::SeqCst).max(0) as u64, self.total))
770 }
771 }
772
773 #[test]
779 fn the_slack_is_the_measured_preallocation_with_a_margin_and_never_under_a_sixth() {
780 assert_eq!(FsSlack::from_measured(Some(0)).slack_permille, PUSH_FS_SLACK_PERMILLE);
781 assert_eq!(FsSlack::from_measured(Some(0)).reserve_permille(PUSH_COPIES), PUSH_RESERVE_PERMILLE);
782 assert_eq!(FsSlack::from_measured(Some(1000)).slack_permille, 1050);
783 assert_eq!(FsSlack::from_measured(Some(1000)).reserve_permille(PUSH_COPIES), 6150);
784 assert_eq!(FsSlack::from_measured(None).slack_permille, UNMEASURED_SLACK_PERMILLE);
785 assert_eq!(FsSlack::default().reserve_permille(PUSH_COPIES), PUSH_RESERVE_PERMILLE, "a ledger nobody measured promises the floor");
786 let said = FsSlack::from_measured(Some(1000)).words(Path::new("/data"), None);
787 assert!(said.contains("held 2.00× a growing file's bytes") && said.contains("every push reserves 6.15× its bytes"), "{said}");
788 assert!(!said.contains(" "), "one line, single-spaced: {said}");
789 let said = FsSlack::from_measured(None).words(Path::new("/data"), Some("12 MiB free"));
790 assert!(said.contains("could not be measured (12 MiB free)") && said.contains("6.15×"), "{said}");
791
792 let disk = FakeDisk::new(10 * GIB, 2 * GIB);
793 let ledger = Arc::new(
794 SpaceLedger::with_probe("/data", DiskFloor { percent: 0, bytes: GIB }, Arc::clone(&disk) as Arc<dyn SpaceProbe>)
795 .with_small_lane(0)
796 .with_fs_slack(FsSlack::from_measured(Some(1000))),
797 );
798 let hold = ledger.hold(PUSH_COPIES);
799 assert_eq!(hold.promise_for(100 * MIB), 615 * MIB, "6.15× promised per wire byte");
800 let granted = hold.admit(GIB).expect("some of it fits");
801 assert_eq!(granted, (GIB * 1000) / 6150, "a measured slack grants less near the floor");
802 let line = hold.admit(GIB).unwrap_err();
803 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}");
804 }
805
806 #[test]
812 fn the_pusher_is_told_full_in_plain_words_and_the_numbers_go_to_the_journal() {
813 let plain = |line: &str| {
814 assert!(line.starts_with(FULL_NOW), "{line}");
815 assert!(line.contains("try again") || line.contains("Try again"), "{line}");
816 for internal in ["/data", "--disk-floor", "transaction log", "spool", "preallocation", "gunnar gc", "repo delete", "×", "MiB", "GiB", "promised to pushes already arriving, and this push"] {
817 assert!(!line.contains(internal), "{internal:?} reaches the pusher: {line}");
818 }
819 assert!(!line.chars().any(|c| c.is_ascii_digit()), "a number reaches the pusher: {line}");
820 };
821 let disk = FakeDisk::new(10 * GIB, 2 * GIB);
822 let ledger = Arc::new(
823 SpaceLedger::with_probe("/data", DiskFloor { percent: 0, bytes: GIB }, Arc::clone(&disk) as Arc<dyn SpaceProbe>)
824 .with_small_lane(0)
825 .with_fs_slack(FsSlack::from_measured(Some(1000))),
826 );
827 let hold = ledger.hold(PUSH_COPIES);
828 hold.admit(GIB).expect("some of it fits");
829 plain(&hold.admit(GIB).unwrap_err());
830 let second = ledger.hold(PUSH_COPIES);
831 let up_front = ledger.full_before_the_first_byte();
832 drop(second);
833 if let Some(line) = up_front {
834 plain(&line);
835 }
836 plain(&ledger.unmeasured_line(&std::io::Error::other("the mount went away")));
837 }
838
839 #[test]
843 fn the_probe_sees_the_preallocation_of_a_growing_file_and_leaves_nothing_behind() {
844 let dir = std::env::temp_dir().join(format!("gunnar-space-probe-test-{}", std::process::id()));
845 let _ = std::fs::remove_dir_all(&dir);
846 std::fs::create_dir_all(&dir).unwrap();
847 std::fs::write(dir.join(format!("{PROBE_FILE_PREFIX}1-7")), b"a crash left me").unwrap();
848 let measured = measure_fs_slack(&dir).expect("measured");
849 let left: Vec<_> = std::fs::read_dir(&dir).unwrap().flatten().map(|e| e.file_name()).collect();
850 assert!(left.is_empty(), "the probe leaves nothing, and takes what a crash left: {left:?}");
851 let xfs = rustix::fs::statfs(&dir).map(|s| s.f_type as u64 == 0x5846_5342).unwrap_or(false);
852 eprintln!("MEASURED probe on {} ({}): a growing file held {} its bytes at worst", dir.display(), if xfs { "xfs" } else { "not xfs" }, times(1000 + measured));
853 if xfs {
854 assert!(measured >= 900, "XFS doubles a growing file's allocation; the probe saw only {} more", times(measured));
855 }
856 let _ = std::fs::remove_dir_all(&dir);
857 }
858
859 fn ledger(disk: &Arc<FakeDisk>, floor: u64, lane: u64) -> Arc<SpaceLedger> {
860 Arc::new(
861 SpaceLedger::with_probe(
862 "/data",
863 DiskFloor { percent: 0, bytes: floor },
864 Arc::clone(disk) as Arc<dyn SpaceProbe>,
865 )
866 .with_small_lane(lane),
867 )
868 }
869
870 fn push(disk: &FakeDisk, ledger: &Arc<SpaceLedger>, bytes: u64, chunk: u64) -> Result<(), String> {
875 let hold = ledger.hold(PUSH_COPIES);
876 let mut spooled = 0;
877 while spooled < bytes {
878 match hold.admit(chunk.min(bytes - spooled)) {
879 Ok(granted) => {
880 disk.write(granted);
881 hold.realize(granted);
882 spooled += granted;
883 }
884 Err(line) => {
885 disk.delete(spooled);
886 return Err(line);
887 }
888 }
889 }
890 disk.write(bytes); hold.realize(bytes);
892 disk.write(bytes); hold.realize(bytes);
894 disk.delete(spooled); drop(hold);
896 Ok(())
897 }
898
899 #[test]
903 fn a_push_that_would_cross_the_floor_is_refused_by_name_and_gives_its_promise_back() {
904 let disk = FakeDisk::new(10 * GIB, 3 * GIB);
905 let ledger = ledger(&disk, 2 * GIB, 0);
906 let hold = ledger.hold(PUSH_COPIES);
907 assert_eq!(hold.admit(256 * MIB), Ok(256 * MIB), "896 MiB (3× + the filesystem's slack) of 1 GiB room fits");
908 assert_eq!(ledger.reserved(), hold.promise_for(256 * MIB));
909 assert_eq!(hold.promise_for(256 * MIB), (256 * MIB * PUSH_RESERVE_PERMILLE).div_ceil(1000));
910 let partial = hold.admit(256 * MIB).expect("what is left is granted in part");
911 assert!(partial < 256 * MIB && partial > 0, "{partial}");
912 let line = hold.admit(MIB).expect_err("no room left: refused");
913 assert!(line.starts_with(FULL_NOW), "{line}");
914 for part in ["Try again", "nothing of it was kept"] {
917 assert!(line.contains(part), "the line must say {part:?}: {line}");
918 }
919 for internal in ["/data", "--disk-floor"] {
920 assert!(!line.contains(internal), "the pusher is not told {internal:?}: {line}");
921 }
922 assert!(
923 !line.starts_with(crate::headroom::AT_FLOOR),
924 "the transient refusal must not read as the floor gate's FINAL one (kits stop on that): {line}"
925 );
926 assert_eq!(ledger.reserved(), 0, "a refused push holds nothing");
927 assert_eq!(hold.admit(1), Err(line.clone()), "refused is refused for good");
928 assert_eq!(ledger.refused(), 1);
929 }
930
931 #[test]
936 fn eight_pushes_into_room_for_three_land_three_or_more_and_refuse_the_rest() {
937 let disk = FakeDisk::new(20 * GIB, 6 * GIB);
940 let ledger = ledger(&disk, 2 * GIB, 0);
941 let holds: Vec<_> = (0..8).map(|_| ledger.hold(PUSH_COPIES)).collect();
942 let size = 384 * MIB;
943 let mut spooled = [0u64; 8];
944 let mut done = [false; 8];
945 let mut refused = [false; 8];
946 for _ in 0..100_000 {
948 let mut moved = false;
949 for i in 0..8 {
950 if done[i] || refused[i] {
951 continue;
952 }
953 moved = true;
954 if spooled[i] == size {
955 for _ in 0..2 {
956 disk.write(size);
957 holds[i].realize(size);
958 }
959 disk.delete(size);
960 holds[i].realize(u64::MAX);
961 done[i] = true;
962 continue;
963 }
964 match holds[i].admit(LEASE_BYTES.min(size - spooled[i])) {
965 Ok(g) => {
966 disk.write(g);
967 holds[i].realize(g);
968 spooled[i] += g;
969 }
970 Err(line) => {
971 assert!(line.starts_with(FULL_NOW), "{line}");
972 disk.delete(spooled[i]);
973 refused[i] = true;
974 }
975 }
976 assert!(
977 disk.free.load(Ordering::SeqCst) - ledger.reserved() as i64 >= (2 * GIB) as i64,
978 "promised + written crossed the floor"
979 );
980 }
981 if !moved {
982 break;
983 }
984 }
985 let landed = done.iter().filter(|d| **d).count();
986 assert!(landed >= 3, "room for {} at the peak, {} landed (refused {refused:?})", 4 * GIB / holds[0].promise_for(size), landed);
987 assert!(refused.iter().any(|r| *r), "8 × 1.3 GiB cannot fit in 4 GiB: something must be refused");
988 assert!(disk.lowest() >= (2 * GIB) as i64, "the floor was crossed: lowest free {}", disk.lowest());
989 drop(holds);
990 assert_eq!((ledger.reserved(), ledger.holds()), (0, 0));
991 }
992
993 #[test]
997 fn concurrent_pushes_never_take_the_disk_under_the_floor() {
998 let disk = FakeDisk::new(20 * GIB, 4 * GIB);
999 let ledger = ledger(&disk, 2 * GIB, 64 * MIB);
1000 let landed = Arc::new(AtomicU64::new(0));
1001 let refused = Arc::new(AtomicU64::new(0));
1002 std::thread::scope(|s| {
1003 for i in 0..16u64 {
1004 let (disk, ledger, landed, refused) = (&disk, &ledger, &landed, &refused);
1005 s.spawn(move || {
1006 match push(disk, ledger, 160 * MIB, MIB * (1 + i % 3)) {
1007 Ok(()) => landed.fetch_add(1, Ordering::SeqCst),
1008 Err(_) => refused.fetch_add(1, Ordering::SeqCst),
1009 };
1010 });
1011 }
1012 });
1013 let (landed, refused) = (landed.load(Ordering::SeqCst), refused.load(Ordering::SeqCst));
1014 assert_eq!(landed + refused, 16);
1015 assert!(landed >= 1 && refused >= 1, "landed {landed}, refused {refused}");
1016 assert!(disk.lowest() >= (2 * GIB) as i64, "the floor was crossed: lowest free {} B", disk.lowest());
1017 assert_eq!((ledger.reserved(), ledger.holds()), (0, 0), "every promise came back");
1018 }
1019
1020 #[test]
1023 fn a_small_push_lands_beside_a_big_one_cut_at_the_lane() {
1024 let disk = FakeDisk::new(10 * GIB, 3 * GIB);
1025 let ledger = ledger(&disk, 2 * GIB, 256 * MIB);
1026 let big = ledger.hold(PUSH_COPIES);
1027 let mut sent = 0;
1028 let mut cut = None;
1029 for _ in 0..10_000 {
1030 match big.admit(LEASE_BYTES) {
1031 Ok(g) => {
1032 disk.write(g);
1033 big.realize(g);
1034 sent += g;
1035 }
1036 Err(line) => {
1037 cut = Some(line);
1038 break;
1039 }
1040 }
1041 }
1042 let line = cut.expect("a big push into 1 GiB of room is cut");
1043 assert!(line.starts_with(FULL_NOW) && line.contains("does not fit"), "{line}");
1044 let room = disk.free.load(Ordering::SeqCst) as u64 - 2 * GIB - ledger.reserved();
1045 assert!(room >= 250 * MIB, "the big push left the lane: {} free above the floor after {} sent", human(room), human(sent));
1046 let small = ledger.hold(PUSH_COPIES);
1047 let mut got = 0;
1048 for _ in 0..1_000 {
1049 if got == 8 * MIB {
1050 break;
1051 }
1052 got += small.admit(8 * MIB - got).expect("a small push lands in the lane");
1053 }
1054 assert_eq!(got, 8 * MIB);
1055 }
1056
1057 #[test]
1060 fn a_dropped_hold_gives_back_what_it_promised() {
1061 let disk = FakeDisk::new(10 * GIB, 5 * GIB);
1062 let ledger = ledger(&disk, 2 * GIB, 0);
1063 let hold = ledger.hold(PUSH_COPIES);
1064 hold.admit(100 * MIB).unwrap();
1065 hold.realize(40 * MIB);
1066 assert_eq!((ledger.reserved(), ledger.holds()), (hold.promise_for(100 * MIB) - 40 * MIB, 1));
1067 drop(hold);
1068 assert_eq!((ledger.reserved(), ledger.holds()), (0, 0));
1069 }
1070
1071 #[test]
1075 fn the_up_front_check_says_full_when_the_room_is_all_promised() {
1076 let disk = FakeDisk::new(10 * GIB, 3 * GIB);
1077 let ledger = ledger(&disk, 2 * GIB, 0);
1078 assert_eq!(ledger.full_before_the_first_byte(), None);
1079 let hold = ledger.hold(PUSH_COPIES);
1080 hold.admit(250 * MIB).unwrap();
1081 assert_eq!(ledger.full_before_the_first_byte(), None, "~149 MiB of room left");
1082 hold.admit(LEASE_BYTES).unwrap();
1083 let rest = hold.admit(GIB).unwrap();
1084 assert!(rest > 0);
1085 let line = ledger.full_before_the_first_byte().expect("all of it promised");
1086 assert!(line.starts_with(FULL_NOW) && line.contains("promised to pushes already arriving"), "{line}");
1087 drop(hold);
1088 assert_eq!(ledger.full_before_the_first_byte(), None);
1089 }
1090
1091 #[test]
1095 fn the_disk_gate_refuses_up_front_when_the_room_is_all_promised() {
1096 use crate::plan::{PlanGate, PlanVerdict};
1097 let disk = FakeDisk::new(10 * GIB, 3 * GIB);
1098 let ledger = ledger(&disk, 2 * GIB, 0);
1099 let gate = crate::headroom::DiskHeadroomGate::over(Arc::clone(&ledger));
1100 assert_eq!(gate.pack_quota("team"), PlanVerdict::Allow);
1101 let hold = ledger.hold(PUSH_COPIES);
1102 for _ in 0..10_000 {
1104 if !hold.admit(LEASE_BYTES).is_ok_and(|g| g == LEASE_BYTES) {
1105 break;
1106 }
1107 }
1108 match gate.pack_quota("team") {
1109 PlanVerdict::Refuse(line) => assert!(line.starts_with(FULL_NOW), "{line}"),
1110 other => panic!("everything above the floor is promised: {other:?}"),
1111 }
1112 assert_eq!(gate.headroom().unwrap().free, 3 * GIB, "Admin.Capacity reads the disk, not the promises");
1113 drop(hold);
1114 assert_eq!(gate.pack_quota("team"), PlanVerdict::Allow, "the promise came back");
1115 disk.write(GIB + 1);
1116 match gate.pack_quota("team") {
1117 PlanVerdict::Refuse(line) => assert!(line.starts_with(crate::headroom::AT_FLOOR), "{line}"),
1118 other => panic!("under the floor: {other:?}"),
1119 }
1120 }
1121
1122 #[test]
1124 fn an_unmeasurable_disk_refuses_the_next_chunk() {
1125 let disk = FakeDisk::new(10 * GIB, 5 * GIB);
1126 let ledger = ledger(&disk, 2 * GIB, 0);
1127 let hold = ledger.hold(PUSH_COPIES);
1128 hold.admit(MIB).unwrap();
1129 disk.fail.store(true, Ordering::SeqCst);
1130 let line = hold.admit(MIB).unwrap_err();
1131 assert!(line.contains("could not check its storage"), "{line}");
1132 assert_eq!(ledger.reserved(), 0);
1133 }
1134
1135 #[test]
1137 fn a_declared_size_is_admitted_whole_or_refused() {
1138 let disk = FakeDisk::new(10 * GIB, 3 * GIB);
1139 let ledger = ledger(&disk, 2 * GIB, 0);
1140 let fits = ledger.hold(PUSH_COPIES);
1141 fits.admit_all(200 * MIB).expect("700 MiB (3× + the filesystem's slack) of 1 GiB");
1142 let no = ledger.hold(PUSH_COPIES);
1143 let line = no.admit_all(200 * MIB).expect_err("700 MiB more does not fit in 324 MiB");
1144 assert!(line.starts_with(FULL_NOW), "{line}");
1145 assert_eq!(ledger.reserved(), fits.promise_for(200 * MIB), "the refused one holds nothing, the other its whole promise");
1146 }
1147
1148 #[test]
1149 fn enospc_and_edquot_are_out_of_space_and_a_plain_cancel_is_not() {
1150 assert!(is_out_of_space(&std::io::Error::from_raw_os_error(28)));
1151 assert!(is_out_of_space(&std::io::Error::from_raw_os_error(122)));
1152 assert!(is_out_of_space(&std::io::Error::from(std::io::ErrorKind::StorageFull)));
1153 assert!(!is_out_of_space(&std::io::Error::from_raw_os_error(125)));
1154 }
1155}