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 {
156 measured_permille: None,
157 slack_permille: PUSH_FS_SLACK_PERMILLE,
158 }
159 }
160}
161
162impl FsSlack {
163 pub fn from_measured(measured_permille: Option<u64>) -> FsSlack {
165 let slack_permille = match measured_permille {
166 Some(m) => m
167 .saturating_add(SLACK_MARGIN_PERMILLE)
168 .max(PUSH_FS_SLACK_PERMILLE),
169 None => UNMEASURED_SLACK_PERMILLE,
170 };
171 FsSlack {
172 measured_permille,
173 slack_permille,
174 }
175 }
176
177 pub fn of(root: &Path) -> (FsSlack, Option<String>) {
179 match measure_fs_slack(root) {
180 Ok(m) => (FsSlack::from_measured(Some(m)), None),
181 Err(e) => (FsSlack::from_measured(None), Some(e.to_string())),
182 }
183 }
184
185 pub fn reserve_permille(&self, copies: u64) -> u64 {
187 copies.max(1).saturating_mul(1000 + self.slack_permille)
188 }
189
190 pub fn words(&self, root: &Path, unmeasured_because: Option<&str>) -> String {
193 let reserve = times(self.reserve_permille(PUSH_COPIES));
194 match self.measured_permille {
195 Some(m) => format!(
196 "push space: the filesystem of {root} held {held} a growing file's bytes at worst while it grew (measured at start with a {probe} write); every push reserves {reserve} its bytes until it lands — its {PUSH_COPIES} copies (spool, transaction log, store), each with room for that preallocation and a {margin} ‰ margin, never under {floor}",
197 root = root.display(),
198 held = times(1000 + m),
199 probe = human(PROBE_BYTES),
200 margin = SLACK_MARGIN_PERMILLE,
201 floor = times(PUSH_RESERVE_PERMILLE),
202 ),
203 None => format!(
204 "push space: the filesystem of {root} could not be measured ({why}); every push reserves {reserve} its bytes until it lands — its {PUSH_COPIES} copies, each with room for a filesystem that doubles a growing file (XFS) and a {margin} ‰ margin",
205 root = root.display(),
206 why = unmeasured_because.unwrap_or("no reason given"),
207 margin = SLACK_MARGIN_PERMILLE,
208 ),
209 }
210 }
211}
212
213fn times(permille: u64) -> String {
215 format!("{}.{:02}×", permille / 1000, (permille % 1000) / 10)
216}
217
218pub fn measure_fs_slack(dir: &Path) -> std::io::Result<u64> {
224 use std::io::Write as _;
225 use std::os::unix::fs::MetadataExt as _;
226 if let Ok(entries) = std::fs::read_dir(dir) {
227 for e in entries.flatten() {
228 if e.file_name()
229 .to_string_lossy()
230 .starts_with(PROBE_FILE_PREFIX)
231 {
232 let _ = std::fs::remove_file(e.path());
233 }
234 }
235 }
236 let (free, _) = statvfs_bytes(dir)?;
237 if free < PROBE_NEEDS_FREE {
238 return Err(std::io::Error::other(format!(
239 "{} free, and the probe needs {}",
240 human(free),
241 human(PROBE_NEEDS_FREE)
242 )));
243 }
244 static SEQ: std::sync::atomic::AtomicU64 = std::sync::atomic::AtomicU64::new(0);
245 let path = dir.join(format!(
246 "{PROBE_FILE_PREFIX}{}-{}",
247 std::process::id(),
248 SEQ.fetch_add(1, std::sync::atomic::Ordering::Relaxed)
249 ));
250 struct Gone(PathBuf);
251 impl Drop for Gone {
252 fn drop(&mut self) {
253 let _ = std::fs::remove_file(&self.0);
254 }
255 }
256 let _gone = Gone(path.clone());
257 let mut file = std::fs::OpenOptions::new()
258 .write(true)
259 .create_new(true)
260 .open(&path)?;
261 let chunk: Vec<u8> = (0..MIB as usize)
262 .map(|i| (i.wrapping_mul(2_654_435_761) >> 13) as u8)
263 .collect();
264 let over = |file: &std::fs::File, written: u64| -> std::io::Result<u64> {
265 let held = file.metadata()?.blocks().saturating_mul(512);
266 Ok(held.saturating_sub(written).saturating_mul(1000) / written.max(1))
267 };
268 let (mut written, mut worst) = (0u64, 0u64);
269 while written < PROBE_BYTES {
270 file.write_all(&chunk)?;
271 written += MIB;
272 if written >= PROBE_FROM {
273 worst = worst.max(over(&file, written)?);
274 }
275 }
276 file.sync_data()?;
277 worst = worst.max(over(&file, written)?);
278 Ok(worst)
279}
280
281pub const SMALL_PUSH_BYTES: u64 = 16 * MIB;
284
285pub const SMALL_PUSH_LANE: u64 = 256 * MIB;
289
290pub const LEASE_BYTES: u64 = 4 * MIB;
292
293pub const MIN_ROOM: u64 = PUSH_RESERVE_PERMILLE * 64 * 1024 / 1000;
296
297pub const FULL_NOW: &str = "the data set is full right now";
303
304pub trait SpaceProbe: Send + Sync + std::fmt::Debug {
307 fn measure(&self) -> std::io::Result<(u64, u64)>;
309}
310
311#[derive(Debug, Clone)]
313pub struct Statvfs(pub PathBuf);
314
315impl SpaceProbe for Statvfs {
316 fn measure(&self) -> std::io::Result<(u64, u64)> {
317 statvfs_bytes(&self.0)
318 }
319}
320
321pub fn statvfs_bytes(root: &Path) -> std::io::Result<(u64, u64)> {
323 let vfs = rustix::fs::statvfs(root).map_err(std::io::Error::from)?;
324 let frsize = if vfs.f_frsize == 0 {
325 vfs.f_bsize
326 } else {
327 vfs.f_frsize
328 };
329 Ok((
330 vfs.f_bavail.saturating_mul(frsize),
331 vfs.f_blocks.saturating_mul(frsize),
332 ))
333}
334
335#[derive(Debug, Clone, Copy, PartialEq, Eq)]
337pub struct Standing {
338 pub free: u64,
340 pub total: u64,
342 pub floor: u64,
344 pub reserved: u64,
346}
347
348impl Standing {
349 pub fn room(&self) -> u64 {
351 self.free
352 .saturating_sub(self.floor)
353 .saturating_sub(self.reserved)
354 }
355}
356
357#[derive(Debug, Default)]
358struct LedgerState {
359 reserved: u64,
361 holds: usize,
363 refused: u64,
365}
366
367#[derive(Debug)]
371pub struct SpaceLedger {
372 root: PathBuf,
373 floor: DiskFloor,
374 probe: Arc<dyn SpaceProbe>,
375 lane: u64,
376 slack: FsSlack,
378 state: Mutex<LedgerState>,
379}
380
381impl SpaceLedger {
382 pub fn new(root: impl Into<PathBuf>, floor: DiskFloor) -> Self {
384 let root = root.into();
385 let probe = Arc::new(Statvfs(root.clone()));
386 Self::with_probe(root, floor, probe)
387 }
388
389 pub fn with_probe(
391 root: impl Into<PathBuf>,
392 floor: DiskFloor,
393 probe: Arc<dyn SpaceProbe>,
394 ) -> Self {
395 SpaceLedger {
396 root: root.into(),
397 floor,
398 probe,
399 lane: SMALL_PUSH_LANE,
400 slack: FsSlack::default(),
401 state: Mutex::new(LedgerState::default()),
402 }
403 }
404
405 pub fn with_fs_slack(mut self, slack: FsSlack) -> Self {
408 self.slack = slack;
409 self
410 }
411
412 pub fn fs_slack(&self) -> FsSlack {
414 self.slack
415 }
416
417 pub fn with_small_lane(mut self, bytes: u64) -> Self {
420 self.lane = bytes;
421 self
422 }
423
424 pub fn root(&self) -> &Path {
426 &self.root
427 }
428
429 pub fn floor(&self) -> DiskFloor {
431 self.floor
432 }
433
434 pub fn small_lane(&self) -> u64 {
436 self.lane
437 }
438
439 pub fn probe(&self) -> &Arc<dyn SpaceProbe> {
441 &self.probe
442 }
443
444 fn lock(&self) -> MutexGuard<'_, LedgerState> {
445 self.state
446 .lock()
447 .unwrap_or_else(|poisoned| poisoned.into_inner())
448 }
449
450 pub fn reserved(&self) -> u64 {
452 self.lock().reserved
453 }
454
455 pub fn holds(&self) -> usize {
457 self.lock().holds
458 }
459
460 pub fn refused(&self) -> u64 {
462 self.lock().refused
463 }
464
465 pub fn standing(&self) -> std::io::Result<Standing> {
467 let reserved = self.reserved();
468 self.measure_with(reserved)
469 }
470
471 fn measure_with(&self, reserved: u64) -> std::io::Result<Standing> {
472 let (free, total) = self.probe.measure()?;
473 Ok(Standing {
474 free,
475 total,
476 floor: self.floor.bytes_for(total),
477 reserved,
478 })
479 }
480
481 pub fn hold(self: &Arc<Self>, copies: u64) -> SpaceHold {
485 self.lock().holds += 1;
486 SpaceHold {
487 ledger: Arc::clone(self),
488 copies: copies.max(1),
489 state: Mutex::new(HoldState::default()),
490 }
491 }
492
493 pub fn full_before_the_first_byte(&self) -> Option<String> {
500 let standing = self.standing().ok()?;
501 if standing.free <= standing.floor || standing.reserved == 0 || standing.room() >= MIN_ROOM
502 {
503 return None;
504 }
505 tracing::warn!(
506 target: "gunnar::space",
507 event = "push.space.full_up_front",
508 root = %self.root.display(),
509 free = %human(standing.free),
510 floor = %human(standing.floor),
511 disk_floor = %self.floor,
512 promised = %human(standing.reserved),
513 "a push was refused before its first byte: everything above the disk floor is promised to pushes already arriving; reclaim space (`gunnar control repo delete --purge`, `gunnar gc`) or grow the storage"
514 );
515 Some(format!(
516 "{FULL_NOW}: everything free is promised to pushes already arriving, so nothing was \
517 received. Try again in a while; the storage is grown when it fills"
518 ))
519 }
520
521 fn full_line(
524 &self,
525 standing: &Standing,
526 own_held: u64,
527 sent: u64,
528 copies: u64,
529 lane: u64,
530 ) -> String {
531 let others = standing.reserved.saturating_sub(own_held);
532 let prealloc = match self.slack.measured_permille {
533 Some(m) => format!(
534 "measured at start to hold {} a growing file's bytes",
535 times(1000 + m)
536 ),
537 None => "not measured, so a doubling is assumed".to_owned(),
538 };
539 tracing::warn!(
540 target: "gunnar::space",
541 event = "push.space.refused",
542 root = %self.root.display(),
543 free = %human(standing.free),
544 floor = %human(standing.floor),
545 disk_floor = %self.floor,
546 small_lane = %human(lane),
547 promised_to_others = %human(others),
548 sent = %human(sent),
549 copies,
550 reserve = %times(self.slack.reserve_permille(copies)),
551 filesystem = %prealloc,
552 "a push did not fit beside the others while it landed: nothing of it was kept; reclaim space (`gunnar control repo delete --purge`, `gunnar gc`) or grow the storage"
553 );
554 format!(
555 "{FULL_NOW}: this push does not fit beside the pushes already arriving, so nothing of it \
556 was kept. Try again in a while; the storage is grown when it fills"
557 )
558 }
559
560 fn unmeasured_line(&self, err: &std::io::Error) -> String {
561 tracing::error!(
562 target: "gunnar::space",
563 event = "push.space.unmeasured",
564 root = %self.root.display(),
565 %err,
566 "the data set could not be measured: a push was refused rather than guessing there is room"
567 );
568 format!(
569 "{FULL_NOW}: Gunnar could not check its storage, so this push was refused rather than \
570 risk it. Nothing of it was kept; try again in a while"
571 )
572 }
573}
574
575#[derive(Debug, Default)]
576struct HoldState {
577 held: u64,
579 admitted: u64,
581 refusal: Option<String>,
583}
584
585#[derive(Debug)]
588pub struct SpaceHold {
589 ledger: Arc<SpaceLedger>,
590 copies: u64,
591 state: Mutex<HoldState>,
592}
593
594impl SpaceHold {
595 fn lock(&self) -> MutexGuard<'_, HoldState> {
596 self.state
597 .lock()
598 .unwrap_or_else(|poisoned| poisoned.into_inner())
599 }
600
601 pub fn admit(&self, want: u64) -> Result<u64, String> {
606 let mut hold = self.lock();
607 if let Some(line) = &hold.refusal {
608 return Err(line.clone());
609 }
610 if want == 0 {
611 return Ok(0);
612 }
613 let ledger = &self.ledger;
614 let mut state = ledger.lock();
615 let standing = match ledger.measure_with(state.reserved) {
616 Ok(standing) => standing,
617 Err(err) => {
618 let line = ledger.unmeasured_line(&err);
619 Self::refuse(&mut hold, &mut state, line.clone());
620 return Err(line);
621 }
622 };
623 let big = hold.admitted.saturating_add(want) > SMALL_PUSH_BYTES;
624 let lane = if big { ledger.lane } else { 0 };
625 let granted =
626 want.min(standing.room().saturating_sub(lane).saturating_mul(1000) / self.per_mille());
627 if granted == 0 {
628 let line = ledger.full_line(&standing, hold.held, hold.admitted, self.copies, lane);
629 Self::refuse(&mut hold, &mut state, line.clone());
630 return Err(line);
631 }
632 let promise = self.promise_for(granted);
633 state.reserved = state.reserved.saturating_add(promise);
634 hold.held = hold.held.saturating_add(promise);
635 hold.admitted = hold.admitted.saturating_add(granted);
636 Ok(granted)
637 }
638
639 pub fn admit_all(&self, bytes: u64) -> Result<(), String> {
642 let mut left = bytes;
643 while left > 0 {
644 let granted = self.admit(left)?;
647 if granted < left {
648 let mut hold = self.lock();
649 let mut state = self.ledger.lock();
650 let standing = self
651 .ledger
652 .measure_with(state.reserved)
653 .unwrap_or(Standing {
654 free: 0,
655 total: 0,
656 floor: 0,
657 reserved: state.reserved,
658 });
659 let lane = if bytes > SMALL_PUSH_BYTES {
660 self.ledger.lane
661 } else {
662 0
663 };
664 let line = self
665 .ledger
666 .full_line(&standing, hold.held, bytes, self.copies, lane);
667 Self::refuse(&mut hold, &mut state, line.clone());
668 return Err(line);
669 }
670 left -= granted;
671 }
672 Ok(())
673 }
674
675 fn refuse(hold: &mut HoldState, state: &mut LedgerState, line: String) {
676 state.reserved = state.reserved.saturating_sub(hold.held);
677 state.refused += 1;
678 hold.held = 0;
679 hold.refusal = Some(line);
680 }
681
682 pub fn realize(&self, bytes: u64) {
687 let mut hold = self.lock();
688 let given = bytes.min(hold.held);
689 if given == 0 {
690 return;
691 }
692 hold.held -= given;
693 let mut state = self.ledger.lock();
694 state.reserved = state.reserved.saturating_sub(given);
695 }
696
697 pub fn held(&self) -> u64 {
699 self.lock().held
700 }
701
702 pub fn admitted(&self) -> u64 {
704 self.lock().admitted
705 }
706
707 pub fn refusal(&self) -> Option<String> {
709 self.lock().refusal.clone()
710 }
711
712 pub fn copies(&self) -> u64 {
714 self.copies
715 }
716
717 pub fn per_mille(&self) -> u64 {
720 self.ledger.slack.reserve_permille(self.copies)
721 }
722
723 pub fn promise_for(&self, wire: u64) -> u64 {
726 wire.saturating_mul(self.per_mille()).div_ceil(1000)
727 }
728
729 pub fn full_now(&self) -> String {
734 let mut hold = self.lock();
735 if let Some(line) = &hold.refusal {
736 return line.clone();
737 }
738 let mut state = self.ledger.lock();
739 let line = match self.ledger.measure_with(state.reserved) {
740 Ok(standing) => {
741 self.ledger
742 .full_line(&standing, hold.held, hold.admitted, self.copies, 0)
743 }
744 Err(err) => self.ledger.unmeasured_line(&err),
745 };
746 Self::refuse(&mut hold, &mut state, line.clone());
747 line
748 }
749
750 pub fn at_the_floor(&self, freed: u64) -> bool {
756 match self.ledger.probe.measure() {
757 Ok((free, total)) => {
758 free.saturating_sub(freed)
759 <= self.ledger.floor.bytes_for(total).saturating_add(MIN_ROOM)
760 }
761 Err(_) => true,
762 }
763 }
764}
765
766impl Drop for SpaceHold {
767 fn drop(&mut self) {
768 let held = std::mem::take(&mut self.lock().held);
769 let mut state = self.ledger.lock();
770 state.reserved = state.reserved.saturating_sub(held);
771 state.holds = state.holds.saturating_sub(1);
772 }
773}
774
775pub fn is_out_of_space(err: &std::io::Error) -> bool {
781 use std::io::ErrorKind;
782 matches!(
783 err.kind(),
784 ErrorKind::StorageFull | ErrorKind::QuotaExceeded
785 ) || matches!(err.raw_os_error(), Some(28) | Some(122))
786}
787
788#[cfg(test)]
789mod tests {
790 use super::*;
791 use std::sync::atomic::{AtomicI64, AtomicU64, Ordering};
792
793 const GIB: u64 = 1024 * MIB;
794
795 #[derive(Debug)]
799 struct FakeDisk {
800 free: AtomicI64,
801 total: u64,
802 lowest: AtomicI64,
803 fail: std::sync::atomic::AtomicBool,
804 }
805
806 impl FakeDisk {
807 fn new(total: u64, free: u64) -> Arc<FakeDisk> {
808 Arc::new(FakeDisk {
809 free: AtomicI64::new(free as i64),
810 total,
811 lowest: AtomicI64::new(free as i64),
812 fail: Default::default(),
813 })
814 }
815 fn write(&self, bytes: u64) {
816 let now = self.free.fetch_sub(bytes as i64, Ordering::SeqCst) - bytes as i64;
817 self.lowest.fetch_min(now, Ordering::SeqCst);
818 }
819 fn delete(&self, bytes: u64) {
820 self.free.fetch_add(bytes as i64, Ordering::SeqCst);
821 }
822 fn lowest(&self) -> i64 {
823 self.lowest.load(Ordering::SeqCst)
824 }
825 }
826
827 impl SpaceProbe for FakeDisk {
828 fn measure(&self) -> std::io::Result<(u64, u64)> {
829 if self.fail.load(Ordering::SeqCst) {
830 return Err(std::io::Error::other("the mount went away"));
831 }
832 Ok((self.free.load(Ordering::SeqCst).max(0) as u64, self.total))
833 }
834 }
835
836 #[test]
842 fn the_slack_is_the_measured_preallocation_with_a_margin_and_never_under_a_sixth() {
843 assert_eq!(
844 FsSlack::from_measured(Some(0)).slack_permille,
845 PUSH_FS_SLACK_PERMILLE
846 );
847 assert_eq!(
848 FsSlack::from_measured(Some(0)).reserve_permille(PUSH_COPIES),
849 PUSH_RESERVE_PERMILLE
850 );
851 assert_eq!(FsSlack::from_measured(Some(1000)).slack_permille, 1050);
852 assert_eq!(
853 FsSlack::from_measured(Some(1000)).reserve_permille(PUSH_COPIES),
854 6150
855 );
856 assert_eq!(
857 FsSlack::from_measured(None).slack_permille,
858 UNMEASURED_SLACK_PERMILLE
859 );
860 assert_eq!(
861 FsSlack::default().reserve_permille(PUSH_COPIES),
862 PUSH_RESERVE_PERMILLE,
863 "a ledger nobody measured promises the floor"
864 );
865 let said = FsSlack::from_measured(Some(1000)).words(Path::new("/data"), None);
866 assert!(
867 said.contains("held 2.00× a growing file's bytes")
868 && said.contains("every push reserves 6.15× its bytes"),
869 "{said}"
870 );
871 assert!(!said.contains(" "), "one line, single-spaced: {said}");
872 let said = FsSlack::from_measured(None).words(Path::new("/data"), Some("12 MiB free"));
873 assert!(
874 said.contains("could not be measured (12 MiB free)") && said.contains("6.15×"),
875 "{said}"
876 );
877
878 let disk = FakeDisk::new(10 * GIB, 2 * GIB);
879 let ledger = Arc::new(
880 SpaceLedger::with_probe(
881 "/data",
882 DiskFloor {
883 percent: 0,
884 bytes: GIB,
885 },
886 Arc::clone(&disk) as Arc<dyn SpaceProbe>,
887 )
888 .with_small_lane(0)
889 .with_fs_slack(FsSlack::from_measured(Some(1000))),
890 );
891 let hold = ledger.hold(PUSH_COPIES);
892 assert_eq!(
893 hold.promise_for(100 * MIB),
894 615 * MIB,
895 "6.15× promised per wire byte"
896 );
897 let granted = hold.admit(GIB).expect("some of it fits");
898 assert_eq!(
899 granted,
900 (GIB * 1000) / 6150,
901 "a measured slack grants less near the floor"
902 );
903 let line = hold.admit(GIB).unwrap_err();
904 assert!(
905 line.starts_with(FULL_NOW) && line.contains("does not fit"),
906 "the pusher's line is plain; the 2.00x slack is the journal's: {line}"
907 );
908 }
909
910 #[test]
916 fn the_pusher_is_told_full_in_plain_words_and_the_numbers_go_to_the_journal() {
917 let plain = |line: &str| {
918 assert!(line.starts_with(FULL_NOW), "{line}");
919 assert!(
920 line.contains("try again") || line.contains("Try again"),
921 "{line}"
922 );
923 for internal in [
924 "/data",
925 "--disk-floor",
926 "transaction log",
927 "spool",
928 "preallocation",
929 "gunnar gc",
930 "repo delete",
931 "×",
932 "MiB",
933 "GiB",
934 "promised to pushes already arriving, and this push",
935 ] {
936 assert!(
937 !line.contains(internal),
938 "{internal:?} reaches the pusher: {line}"
939 );
940 }
941 assert!(
942 !line.chars().any(|c| c.is_ascii_digit()),
943 "a number reaches the pusher: {line}"
944 );
945 };
946 let disk = FakeDisk::new(10 * GIB, 2 * GIB);
947 let ledger = Arc::new(
948 SpaceLedger::with_probe(
949 "/data",
950 DiskFloor {
951 percent: 0,
952 bytes: GIB,
953 },
954 Arc::clone(&disk) as Arc<dyn SpaceProbe>,
955 )
956 .with_small_lane(0)
957 .with_fs_slack(FsSlack::from_measured(Some(1000))),
958 );
959 let hold = ledger.hold(PUSH_COPIES);
960 hold.admit(GIB).expect("some of it fits");
961 plain(&hold.admit(GIB).unwrap_err());
962 let second = ledger.hold(PUSH_COPIES);
963 let up_front = ledger.full_before_the_first_byte();
964 drop(second);
965 if let Some(line) = up_front {
966 plain(&line);
967 }
968 plain(&ledger.unmeasured_line(&std::io::Error::other("the mount went away")));
969 }
970
971 #[test]
975 fn the_probe_sees_the_preallocation_of_a_growing_file_and_leaves_nothing_behind() {
976 let dir =
977 std::env::temp_dir().join(format!("gunnar-space-probe-test-{}", std::process::id()));
978 let _ = std::fs::remove_dir_all(&dir);
979 std::fs::create_dir_all(&dir).unwrap();
980 std::fs::write(
981 dir.join(format!("{PROBE_FILE_PREFIX}1-7")),
982 b"a crash left me",
983 )
984 .unwrap();
985 let measured = measure_fs_slack(&dir).expect("measured");
986 let left: Vec<_> = std::fs::read_dir(&dir)
987 .unwrap()
988 .flatten()
989 .map(|e| e.file_name())
990 .collect();
991 assert!(
992 left.is_empty(),
993 "the probe leaves nothing, and takes what a crash left: {left:?}"
994 );
995 let xfs = rustix::fs::statfs(&dir)
996 .map(|s| s.f_type as u64 == 0x5846_5342)
997 .unwrap_or(false);
998 eprintln!(
999 "MEASURED probe on {} ({}): a growing file held {} its bytes at worst",
1000 dir.display(),
1001 if xfs { "xfs" } else { "not xfs" },
1002 times(1000 + measured)
1003 );
1004 if xfs {
1005 assert!(
1006 measured >= 900,
1007 "XFS doubles a growing file's allocation; the probe saw only {} more",
1008 times(measured)
1009 );
1010 }
1011 let _ = std::fs::remove_dir_all(&dir);
1012 }
1013
1014 fn ledger(disk: &Arc<FakeDisk>, floor: u64, lane: u64) -> Arc<SpaceLedger> {
1015 Arc::new(
1016 SpaceLedger::with_probe(
1017 "/data",
1018 DiskFloor {
1019 percent: 0,
1020 bytes: floor,
1021 },
1022 Arc::clone(disk) as Arc<dyn SpaceProbe>,
1023 )
1024 .with_small_lane(lane),
1025 )
1026 }
1027
1028 fn push(
1033 disk: &FakeDisk,
1034 ledger: &Arc<SpaceLedger>,
1035 bytes: u64,
1036 chunk: u64,
1037 ) -> Result<(), String> {
1038 let hold = ledger.hold(PUSH_COPIES);
1039 let mut spooled = 0;
1040 while spooled < bytes {
1041 match hold.admit(chunk.min(bytes - spooled)) {
1042 Ok(granted) => {
1043 disk.write(granted);
1044 hold.realize(granted);
1045 spooled += granted;
1046 }
1047 Err(line) => {
1048 disk.delete(spooled);
1049 return Err(line);
1050 }
1051 }
1052 }
1053 disk.write(bytes); hold.realize(bytes);
1055 disk.write(bytes); hold.realize(bytes);
1057 disk.delete(spooled); drop(hold);
1059 Ok(())
1060 }
1061
1062 #[test]
1066 fn a_push_that_would_cross_the_floor_is_refused_by_name_and_gives_its_promise_back() {
1067 let disk = FakeDisk::new(10 * GIB, 3 * GIB);
1068 let ledger = ledger(&disk, 2 * GIB, 0);
1069 let hold = ledger.hold(PUSH_COPIES);
1070 assert_eq!(
1071 hold.admit(256 * MIB),
1072 Ok(256 * MIB),
1073 "896 MiB (3× + the filesystem's slack) of 1 GiB room fits"
1074 );
1075 assert_eq!(ledger.reserved(), hold.promise_for(256 * MIB));
1076 assert_eq!(
1077 hold.promise_for(256 * MIB),
1078 (256 * MIB * PUSH_RESERVE_PERMILLE).div_ceil(1000)
1079 );
1080 let partial = hold
1081 .admit(256 * MIB)
1082 .expect("what is left is granted in part");
1083 assert!(partial < 256 * MIB && partial > 0, "{partial}");
1084 let line = hold.admit(MIB).expect_err("no room left: refused");
1085 assert!(line.starts_with(FULL_NOW), "{line}");
1086 for part in ["Try again", "nothing of it was kept"] {
1089 assert!(line.contains(part), "the line must say {part:?}: {line}");
1090 }
1091 for internal in ["/data", "--disk-floor"] {
1092 assert!(
1093 !line.contains(internal),
1094 "the pusher is not told {internal:?}: {line}"
1095 );
1096 }
1097 assert!(
1098 !line.starts_with(crate::headroom::AT_FLOOR),
1099 "the transient refusal must not read as the floor gate's FINAL one (kits stop on that): {line}"
1100 );
1101 assert_eq!(ledger.reserved(), 0, "a refused push holds nothing");
1102 assert_eq!(
1103 hold.admit(1),
1104 Err(line.clone()),
1105 "refused is refused for good"
1106 );
1107 assert_eq!(ledger.refused(), 1);
1108 }
1109
1110 #[test]
1115 fn eight_pushes_into_room_for_three_land_three_or_more_and_refuse_the_rest() {
1116 let disk = FakeDisk::new(20 * GIB, 6 * GIB);
1119 let ledger = ledger(&disk, 2 * GIB, 0);
1120 let holds: Vec<_> = (0..8).map(|_| ledger.hold(PUSH_COPIES)).collect();
1121 let size = 384 * MIB;
1122 let mut spooled = [0u64; 8];
1123 let mut done = [false; 8];
1124 let mut refused = [false; 8];
1125 for _ in 0..100_000 {
1127 let mut moved = false;
1128 for i in 0..8 {
1129 if done[i] || refused[i] {
1130 continue;
1131 }
1132 moved = true;
1133 if spooled[i] == size {
1134 for _ in 0..2 {
1135 disk.write(size);
1136 holds[i].realize(size);
1137 }
1138 disk.delete(size);
1139 holds[i].realize(u64::MAX);
1140 done[i] = true;
1141 continue;
1142 }
1143 match holds[i].admit(LEASE_BYTES.min(size - spooled[i])) {
1144 Ok(g) => {
1145 disk.write(g);
1146 holds[i].realize(g);
1147 spooled[i] += g;
1148 }
1149 Err(line) => {
1150 assert!(line.starts_with(FULL_NOW), "{line}");
1151 disk.delete(spooled[i]);
1152 refused[i] = true;
1153 }
1154 }
1155 assert!(
1156 disk.free.load(Ordering::SeqCst) - ledger.reserved() as i64 >= (2 * GIB) as i64,
1157 "promised + written crossed the floor"
1158 );
1159 }
1160 if !moved {
1161 break;
1162 }
1163 }
1164 let landed = done.iter().filter(|d| **d).count();
1165 assert!(
1166 landed >= 3,
1167 "room for {} at the peak, {} landed (refused {refused:?})",
1168 4 * GIB / holds[0].promise_for(size),
1169 landed
1170 );
1171 assert!(
1172 refused.iter().any(|r| *r),
1173 "8 × 1.3 GiB cannot fit in 4 GiB: something must be refused"
1174 );
1175 assert!(
1176 disk.lowest() >= (2 * GIB) as i64,
1177 "the floor was crossed: lowest free {}",
1178 disk.lowest()
1179 );
1180 drop(holds);
1181 assert_eq!((ledger.reserved(), ledger.holds()), (0, 0));
1182 }
1183
1184 #[test]
1188 fn concurrent_pushes_never_take_the_disk_under_the_floor() {
1189 let disk = FakeDisk::new(20 * GIB, 4 * GIB);
1190 let ledger = ledger(&disk, 2 * GIB, 64 * MIB);
1191 let landed = Arc::new(AtomicU64::new(0));
1192 let refused = Arc::new(AtomicU64::new(0));
1193 std::thread::scope(|s| {
1194 for i in 0..16u64 {
1195 let (disk, ledger, landed, refused) = (&disk, &ledger, &landed, &refused);
1196 s.spawn(move || {
1197 match push(disk, ledger, 160 * MIB, MIB * (1 + i % 3)) {
1198 Ok(()) => landed.fetch_add(1, Ordering::SeqCst),
1199 Err(_) => refused.fetch_add(1, Ordering::SeqCst),
1200 };
1201 });
1202 }
1203 });
1204 let (landed, refused) = (
1205 landed.load(Ordering::SeqCst),
1206 refused.load(Ordering::SeqCst),
1207 );
1208 assert_eq!(landed + refused, 16);
1209 assert!(
1210 landed >= 1 && refused >= 1,
1211 "landed {landed}, refused {refused}"
1212 );
1213 assert!(
1214 disk.lowest() >= (2 * GIB) as i64,
1215 "the floor was crossed: lowest free {} B",
1216 disk.lowest()
1217 );
1218 assert_eq!(
1219 (ledger.reserved(), ledger.holds()),
1220 (0, 0),
1221 "every promise came back"
1222 );
1223 }
1224
1225 #[test]
1228 fn a_small_push_lands_beside_a_big_one_cut_at_the_lane() {
1229 let disk = FakeDisk::new(10 * GIB, 3 * GIB);
1230 let ledger = ledger(&disk, 2 * GIB, 256 * MIB);
1231 let big = ledger.hold(PUSH_COPIES);
1232 let mut sent = 0;
1233 let mut cut = None;
1234 for _ in 0..10_000 {
1235 match big.admit(LEASE_BYTES) {
1236 Ok(g) => {
1237 disk.write(g);
1238 big.realize(g);
1239 sent += g;
1240 }
1241 Err(line) => {
1242 cut = Some(line);
1243 break;
1244 }
1245 }
1246 }
1247 let line = cut.expect("a big push into 1 GiB of room is cut");
1248 assert!(
1249 line.starts_with(FULL_NOW) && line.contains("does not fit"),
1250 "{line}"
1251 );
1252 let room = disk.free.load(Ordering::SeqCst) as u64 - 2 * GIB - ledger.reserved();
1253 assert!(
1254 room >= 250 * MIB,
1255 "the big push left the lane: {} free above the floor after {} sent",
1256 human(room),
1257 human(sent)
1258 );
1259 let small = ledger.hold(PUSH_COPIES);
1260 let mut got = 0;
1261 for _ in 0..1_000 {
1262 if got == 8 * MIB {
1263 break;
1264 }
1265 got += small
1266 .admit(8 * MIB - got)
1267 .expect("a small push lands in the lane");
1268 }
1269 assert_eq!(got, 8 * MIB);
1270 }
1271
1272 #[test]
1275 fn a_dropped_hold_gives_back_what_it_promised() {
1276 let disk = FakeDisk::new(10 * GIB, 5 * GIB);
1277 let ledger = ledger(&disk, 2 * GIB, 0);
1278 let hold = ledger.hold(PUSH_COPIES);
1279 hold.admit(100 * MIB).unwrap();
1280 hold.realize(40 * MIB);
1281 assert_eq!(
1282 (ledger.reserved(), ledger.holds()),
1283 (hold.promise_for(100 * MIB) - 40 * MIB, 1)
1284 );
1285 drop(hold);
1286 assert_eq!((ledger.reserved(), ledger.holds()), (0, 0));
1287 }
1288
1289 #[test]
1293 fn the_up_front_check_says_full_when_the_room_is_all_promised() {
1294 let disk = FakeDisk::new(10 * GIB, 3 * GIB);
1295 let ledger = ledger(&disk, 2 * GIB, 0);
1296 assert_eq!(ledger.full_before_the_first_byte(), None);
1297 let hold = ledger.hold(PUSH_COPIES);
1298 hold.admit(250 * MIB).unwrap();
1299 assert_eq!(
1300 ledger.full_before_the_first_byte(),
1301 None,
1302 "~149 MiB of room left"
1303 );
1304 hold.admit(LEASE_BYTES).unwrap();
1305 let rest = hold.admit(GIB).unwrap();
1306 assert!(rest > 0);
1307 let line = ledger
1308 .full_before_the_first_byte()
1309 .expect("all of it promised");
1310 assert!(
1311 line.starts_with(FULL_NOW) && line.contains("promised to pushes already arriving"),
1312 "{line}"
1313 );
1314 drop(hold);
1315 assert_eq!(ledger.full_before_the_first_byte(), None);
1316 }
1317
1318 #[test]
1322 fn the_disk_gate_refuses_up_front_when_the_room_is_all_promised() {
1323 use crate::plan::{PlanGate, PlanVerdict};
1324 let disk = FakeDisk::new(10 * GIB, 3 * GIB);
1325 let ledger = ledger(&disk, 2 * GIB, 0);
1326 let gate = crate::headroom::DiskHeadroomGate::over(Arc::clone(&ledger));
1327 assert_eq!(gate.pack_quota("team"), PlanVerdict::Allow);
1328 let hold = ledger.hold(PUSH_COPIES);
1329 for _ in 0..10_000 {
1331 if !hold.admit(LEASE_BYTES).is_ok_and(|g| g == LEASE_BYTES) {
1332 break;
1333 }
1334 }
1335 match gate.pack_quota("team") {
1336 PlanVerdict::Refuse(line) => assert!(line.starts_with(FULL_NOW), "{line}"),
1337 other => panic!("everything above the floor is promised: {other:?}"),
1338 }
1339 assert_eq!(
1340 gate.headroom().unwrap().free,
1341 3 * GIB,
1342 "Admin.Capacity reads the disk, not the promises"
1343 );
1344 drop(hold);
1345 assert_eq!(
1346 gate.pack_quota("team"),
1347 PlanVerdict::Allow,
1348 "the promise came back"
1349 );
1350 disk.write(GIB + 1);
1351 match gate.pack_quota("team") {
1352 PlanVerdict::Refuse(line) => {
1353 assert!(line.starts_with(crate::headroom::AT_FLOOR), "{line}")
1354 }
1355 other => panic!("under the floor: {other:?}"),
1356 }
1357 }
1358
1359 #[test]
1361 fn an_unmeasurable_disk_refuses_the_next_chunk() {
1362 let disk = FakeDisk::new(10 * GIB, 5 * GIB);
1363 let ledger = ledger(&disk, 2 * GIB, 0);
1364 let hold = ledger.hold(PUSH_COPIES);
1365 hold.admit(MIB).unwrap();
1366 disk.fail.store(true, Ordering::SeqCst);
1367 let line = hold.admit(MIB).unwrap_err();
1368 assert!(line.contains("could not check its storage"), "{line}");
1369 assert_eq!(ledger.reserved(), 0);
1370 }
1371
1372 #[test]
1374 fn a_declared_size_is_admitted_whole_or_refused() {
1375 let disk = FakeDisk::new(10 * GIB, 3 * GIB);
1376 let ledger = ledger(&disk, 2 * GIB, 0);
1377 let fits = ledger.hold(PUSH_COPIES);
1378 fits.admit_all(200 * MIB)
1379 .expect("700 MiB (3× + the filesystem's slack) of 1 GiB");
1380 let no = ledger.hold(PUSH_COPIES);
1381 let line = no
1382 .admit_all(200 * MIB)
1383 .expect_err("700 MiB more does not fit in 324 MiB");
1384 assert!(line.starts_with(FULL_NOW), "{line}");
1385 assert_eq!(
1386 ledger.reserved(),
1387 fits.promise_for(200 * MIB),
1388 "the refused one holds nothing, the other its whole promise"
1389 );
1390 }
1391
1392 #[test]
1393 fn enospc_and_edquot_are_out_of_space_and_a_plain_cancel_is_not() {
1394 assert!(is_out_of_space(&std::io::Error::from_raw_os_error(28)));
1395 assert!(is_out_of_space(&std::io::Error::from_raw_os_error(122)));
1396 assert!(is_out_of_space(&std::io::Error::from(
1397 std::io::ErrorKind::StorageFull
1398 )));
1399 assert!(!is_out_of_space(&std::io::Error::from_raw_os_error(125)));
1400 }
1401}