crates/gunnar-server/src/disktest.rs
//! **The start-up disk test**: what /data can actually do, said once on the
//! screen.
//!
//! ```text
//! 17:40:43 disktest /data seq w 812 MB/s · r 1.9 GB/s · 4k sync 3100 ops/s (1.6 s)
//! ```
//!
//! Three numbers, because they are the three a git server lives on: a
//! sequential write (a pack arriving), a sequential read (a clone leaving),
//! and 4 KiB overwrites each followed by `fsync` — the loose-object / ref
//! update pattern, and the number that separates a FAT /data from an xfs one
//! (an open item; the rig reads this row).
//!
//! The write and the 4 KiB legs are plain `std::fs`. The READ leg measures the
//! device, not the page cache: the file it just wrote is still cached, so the
//! read reopens it with `O_DIRECT` (one 4 MiB buffer, 4 KiB-aligned,
//! allocated once for the whole run) and streams it back off the disk. Where
//! `O_DIRECT` is refused (tmpfs, some FUSE) the fallback is
//! `posix_fadvise(DONTNEED)` to drop the cached pages, then a buffered read —
//! and a result that is still implausible (over 5 GB/s) is said to be
//! `cached` rather than reported as disk speed. On the appliance `/data` is
//! the mounted xfs over the whole md raid0 set, so the numbers are the
//! stripe's, not one member disk's. 256 MiB under `<data>/.korp-disktest`,
//! deleted afterwards. Runs on its own thread the moment `/data` is known to
//! be mounted and never delays serving: the row arrives when it arrives.
//!
//! Never a panic, never a reason PID 1 is unhappy: too little free space is a
//! `skipped` row, any I/O error is one WARN row, and both leave the file gone.
use std::fs::{File, OpenOptions};
use std::io::{Read, Seek, SeekFrom, Write};
use std::os::unix::fs::OpenOptionsExt;
use std::path::{Path, PathBuf};
use std::time::{Duration, Instant};
use rustix::fs::{fadvise, Advice, OFlags};
/// The env seam: `GUNNAR_DISKTEST=1` runs it on a developer box regardless
/// of the PID-1 gate.
pub const ENV: &str = "GUNNAR_DISKTEST";
/// The file's directory under the data root; removed when the test is done.
pub const DIR_NAME: &str = ".korp-disktest";
/// What the test writes.
pub const FILE_BYTES: u64 = 256 << 20;
/// Skip below this much free space: the test must never be the thing that
/// fills the disk.
pub const MIN_FREE_BYTES: u64 = 1 << 30;
/// How long the 4 KiB fsync loop runs.
pub const SYNC_WINDOW: Duration = Duration::from_millis(500);
/// A read faster than this did not touch the disk.
pub const CACHED_BPS: f64 = 5e9;
/// The write chunk and the read buffer: one buffer of this size, allocated
/// once per run, aligned for `O_DIRECT`.
const CHUNK: usize = 4 << 20;
const BLOCK: usize = 4096;
/// `O_DIRECT` wants the buffer, the offset and the length aligned to the
/// logical block; 4 KiB satisfies every disk the appliance runs on.
const DIRECT_ALIGN: usize = 4096;
/// How the read leg got its bytes.
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub enum ReadHow {
/// `O_DIRECT`: the page cache was bypassed, the number is the device's.
Direct,
/// `O_DIRECT` refused; the cached pages were dropped (`fadvise
/// DONTNEED`) and the file read through the cache.
Dropped,
}
impl ReadHow {
pub fn as_str(self) -> &'static str {
match self {
ReadHow::Direct => "direct",
ReadHow::Dropped => "dropped",
}
}
}
/// What the test measured.
#[derive(Debug, Clone, PartialEq)]
pub struct Report {
pub seq_write_bps: f64,
pub seq_read_bps: f64,
/// Whether the read bypassed the page cache or only dropped it first.
pub read_how: ReadHow,
pub sync_ops_per_s: f64,
pub elapsed: Duration,
/// **What the rates are made of** (2026-09-28): the bytes written and
/// read back, each leg's own clock, and the fsync'd 4 KiB operations the
/// sync rate counts — so a reader can check a rate against the run that
/// produced it instead of taking a derived number on trust.
pub file_bytes: u64,
pub write_elapsed: Duration,
pub read_elapsed: Duration,
pub sync_ops: u64,
pub sync_elapsed: Duration,
}
impl Report {
/// `seq w 812 MB/s · r 1.9 GB/s · 4k sync 3100 ops/s (1.6 s)`; a read
/// over [`CACHED_BPS`] reads `r cached`.
pub fn row(&self) -> String {
let read = if self.seq_read_bps > CACHED_BPS {
"cached".to_owned()
} else {
fmt_rate(self.seq_read_bps)
};
format!(
"seq w {} \u{b7} r {} \u{b7} 4k sync {:.0} ops/s ({:.1} s)",
fmt_rate(self.seq_write_bps),
read,
self.sync_ops_per_s,
self.elapsed.as_secs_f64()
)
}
}
/// `812 MB/s` / `1.9 GB/s` (decimal units, as disks are sold).
pub fn fmt_rate(bps: f64) -> String {
if bps >= 1e9 {
format!("{:.1} GB/s", bps / 1e9)
} else if bps >= 1e6 {
format!("{:.0} MB/s", bps / 1e6)
} else {
format!("{:.0} kB/s", bps / 1e3)
}
}
/// How one run ended.
#[derive(Debug, Clone, PartialEq)]
pub enum Outcome {
Done(Report),
/// Too little free space: `(free, needed)` bytes.
Skipped {
free: u64,
need: u64,
},
Failed(String),
}
impl Outcome {
/// The row's text; the kind is `disktest` either way, `Failed` warns.
pub fn row(&self) -> (String, bool) {
match self {
Outcome::Done(r) => (r.row(), false),
Outcome::Skipped { free, need } => (
format!(
"skipped: {} free (< {})",
crate::dashboard::fmt_bytes(*free),
crate::dashboard::fmt_bytes(*need)
),
false,
),
Outcome::Failed(why) => (format!("failed: {why}"), true),
}
}
}
/// The skip rule, pure: `free` bytes available against [`MIN_FREE_BYTES`]
/// (or `need`).
pub fn should_skip(free: u64, need: u64) -> bool {
free < need
}
/// Run the test under `data_root` with the production sizes.
pub fn run(data_root: &Path) -> Outcome {
run_sized(data_root, FILE_BYTES, MIN_FREE_BYTES, SYNC_WINDOW)
}
/// The test at any size — what the unit test calls with a few MiB.
pub fn run_sized(
data_root: &Path,
file_bytes: u64,
min_free: u64,
sync_window: Duration,
) -> Outcome {
let free = match crate::dashboard::probe::disk(data_root) {
Some((_, avail)) => avail,
None => {
return Outcome::Failed(format!(
"statfs {}: not a mounted filesystem",
data_root.display()
))
}
};
if should_skip(free, min_free) {
return Outcome::Skipped {
free,
need: min_free,
};
}
let dir = data_root.join(DIR_NAME);
let path = dir.join("probe");
let result = measure(&dir, &path, file_bytes, sync_window);
// Leave nothing behind whichever way it went.
let _ = std::fs::remove_file(&path);
let _ = std::fs::remove_dir(&dir);
match result {
Ok(report) => Outcome::Done(report),
Err(why) => Outcome::Failed(why),
}
}
fn measure(
dir: &Path,
path: &Path,
file_bytes: u64,
sync_window: Duration,
) -> Result<Report, String> {
let started = Instant::now();
std::fs::create_dir_all(dir).map_err(|e| format!("mkdir {}: {e}", dir.display()))?;
// Not zeros: a filesystem or a virtual disk that compresses or
// deduplicates would report a write it never made. ONE buffer for the
// whole run — the write source, the O_DIRECT read target and the 4 KiB
// block all live in it (allocate once, reuse).
let mut buf = AlignedBuf::new(CHUNK, DIRECT_ALIGN);
buf.fill(&pattern(CHUNK));
// ── sequential write + fsync ───────────────────────────────────────────
let t = Instant::now();
let mut f = File::create(path).map_err(|e| format!("create: {e}"))?;
let mut left = file_bytes;
while left > 0 {
let n = (left as usize).min(CHUNK);
f.write_all(&buf.as_slice()[..n])
.map_err(|e| format!("write: {e}"))?;
left -= n as u64;
}
f.sync_all().map_err(|e| format!("fsync: {e}"))?;
drop(f);
let write_elapsed = t.elapsed();
let seq_write_bps = file_bytes as f64 / write_elapsed.as_secs_f64().max(1e-9);
// ── sequential read, off the device ───────────────────────────────────
let (got, secs, read_how) = read_back(path, &mut buf, true)?;
if got != file_bytes {
return Err(format!("read back {got} of {file_bytes} bytes"));
}
let seq_read_bps = got as f64 / secs.max(1e-9);
// ── 4 KiB random overwrites, fsync each ────────────────────────────────
// The read leg may have filled the buffer with what it read — the same
// pattern, but restore the block so the leg does not depend on that.
let block_src = pattern(BLOCK);
buf.as_mut_slice()[..BLOCK].copy_from_slice(&block_src);
let mut f = OpenOptions::new()
.write(true)
.open(path)
.map_err(|e| format!("reopen: {e}"))?;
let blocks = (file_bytes / BLOCK as u64).max(1);
let block = &buf.as_slice()[..BLOCK];
let mut rng = started.elapsed().as_nanos() as u64 | 1;
let mut ops = 0u64;
let t = Instant::now();
while t.elapsed() < sync_window {
// xorshift64: enough randomness to defeat any write-combining.
rng ^= rng << 13;
rng ^= rng >> 7;
rng ^= rng << 17;
let at = (rng % blocks) * BLOCK as u64;
f.seek(SeekFrom::Start(at))
.map_err(|e| format!("seek: {e}"))?;
f.write_all(block).map_err(|e| format!("write 4k: {e}"))?;
f.sync_data().map_err(|e| format!("fdatasync: {e}"))?;
ops += 1;
}
let sync_elapsed = t.elapsed();
let sync_ops_per_s = ops as f64 / sync_elapsed.as_secs_f64().max(1e-9);
drop(f);
Ok(Report {
seq_write_bps,
seq_read_bps,
read_how,
sync_ops_per_s,
elapsed: started.elapsed(),
file_bytes,
write_elapsed,
read_elapsed: Duration::from_secs_f64(secs.max(0.0)),
sync_ops: ops,
sync_elapsed,
})
}
/// Read `path` back sequentially into `buf`, past the page cache: `O_DIRECT`
/// when `try_direct` and the filesystem takes it, else drop the cached pages
/// (`fadvise DONTNEED`, after the caller's fsync they are clean) and read
/// buffered. Returns `(bytes, seconds, how)`; the clock starts after the
/// open so the number is the transfer, not the open.
pub fn read_back(
path: &Path,
buf: &mut AlignedBuf,
try_direct: bool,
) -> Result<(u64, f64, ReadHow), String> {
let direct = if try_direct {
OpenOptions::new()
.read(true)
.custom_flags(OFlags::DIRECT.bits() as i32)
.open(path)
.ok()
} else {
None
};
let (mut f, how) = match direct {
Some(f) => (f, ReadHow::Direct),
None => {
let f = File::open(path).map_err(|e| format!("open: {e}"))?;
// Best effort: a filesystem that ignores the advice is one whose
// "read" is memory anyway, and the > 5 GB/s rule catches it.
let _ = fadvise(&f, 0, None, Advice::DontNeed);
(f, ReadHow::Dropped)
}
};
let t = Instant::now();
let mut got = 0u64;
loop {
let n = f
.read(buf.as_mut_slice())
.map_err(|e| format!("read ({}): {e}", how.as_str()))?;
if n == 0 {
break;
}
got += n as u64;
}
Ok((got, t.elapsed().as_secs_f64(), how))
}
/// A heap buffer whose start is aligned to `align` (a power of two): what
/// `O_DIRECT` needs and `Vec<u8>` cannot promise. Allocated once per run.
pub struct AlignedBuf {
raw: Vec<u8>,
at: usize,
len: usize,
}
impl AlignedBuf {
pub fn new(len: usize, align: usize) -> Self {
debug_assert!(align.is_power_of_two());
let raw = vec![0u8; len + align];
let at = raw.as_ptr().align_offset(align);
Self { raw, at, len }
}
pub fn fill(&mut self, src: &[u8]) {
let n = src.len().min(self.len);
self.as_mut_slice()[..n].copy_from_slice(&src[..n]);
}
pub fn as_slice(&self) -> &[u8] {
&self.raw[self.at..self.at + self.len]
}
pub fn as_mut_slice(&mut self) -> &mut [u8] {
&mut self.raw[self.at..self.at + self.len]
}
}
fn pattern(n: usize) -> Vec<u8> {
let mut v = Vec::with_capacity(n);
let mut x = 0x9E37_79B9_7F4A_7C15u64;
while v.len() < n {
x ^= x << 13;
x ^= x >> 7;
x ^= x << 17;
v.extend_from_slice(&x.to_le_bytes());
}
v.truncate(n);
v
}
/// **The latch.** Until 2026-09-21 the outcome went to the screen and nowhere
/// else, so nothing but a human reading a console could learn what this box's
/// disk does. The twin election compares two boxes' disks over gRPC, so the
/// last outcome is kept here and read by [`crate::capacity::MachineFacts`].
///
/// `None` means the test has not finished — the gate was closed, or the
/// thread is still running. It is never a zero and never an empty `Report`:
/// `Admin.Capacity` renders it `DISKTEST_NOT_RUN`, which is a different fact
/// from a disk that measured badly.
static LAST: std::sync::OnceLock<std::sync::Mutex<Option<Outcome>>> = std::sync::OnceLock::new();
fn latch() -> &'static std::sync::Mutex<Option<Outcome>> {
LAST.get_or_init(|| std::sync::Mutex::new(None))
}
/// Record the outcome of a run. Called once per run, beside [`announce`].
pub fn record(outcome: &Outcome) {
if let Ok(mut slot) = latch().lock() {
*slot = Some(outcome.clone());
}
}
/// The last outcome this process measured, or `None` if it has not finished
/// one. A lookup, never a run: asking this question must never start a disk
/// test, because the answer a caller wants is what the disk did at start-up.
pub fn last() -> Option<Outcome> {
latch().lock().ok().and_then(|slot| slot.clone())
}
/// Say the outcome as the ONE tracing line the log tap turns into the
/// `disktest` row (`dashboard::describe`).
pub fn announce(data_root: &Path, outcome: &Outcome) {
let (text, warn) = outcome.row();
let read = match outcome {
Outcome::Done(r) => r.read_how.as_str(),
_ => "-",
};
if warn {
tracing::warn!(event = "disktest", path = %data_root.display(), read, "{text}");
} else {
tracing::info!(event = "disktest", path = %data_root.display(), read, "{text}");
}
}
/// Start the test on its own thread when this is the appliance's PID 1 (the
/// dashboard's gate) or [`ENV`] is set; say nothing otherwise. Returns
/// whether it started. Serving never waits for it.
pub fn spawn(data_root: PathBuf) -> bool {
let explicit = std::env::var(ENV)
.ok()
.filter(|v| !v.is_empty() && v != "0");
let gated = appliance_pid1::net::is_init() && appliance_pid1::net::booted_by_korp_start();
let started = std::time::Instant::now();
if explicit.is_none() && !gated {
// No disk test, so nothing competes with the store for the disk and
// the server is ready from its first bound socket. No window is opened
// and none has to be closed.
return false;
}
// **The boot window opens here.** See `crate::readiness`.
crate::readiness::begin("the start-up disk test");
// A28 M18, stays a thread: one bounded measurement, its panic caught, its result and its start failure answered.
let spawned = std::thread::Builder::new()
.name("gunnar-disktest".into())
.spawn(move || {
let outcome = std::panic::catch_unwind(|| run(&data_root))
.unwrap_or_else(|_| Outcome::Failed("the disk test panicked".to_owned()));
record(&outcome);
announce(&data_root, &outcome);
// **The boot window closes here**, and only here. For the whole of
// this thread's life another thread of this process has been
// saturating the same disk a repository's first post-crash open
// has to read (`4k sync 4517 ops/s (0.7 s)` on the rig), which is
// the measured 0.6-0.7 s of `the repository is unavailable`.
crate::readiness::mark_ready("the start-up disk test", started);
})
.is_ok();
if !spawned {
// A thread that could not be spawned must not leave the server refusing
// every transfer for ever. The failure is already the caller's `false`.
crate::readiness::mark_ready("a disk test that could not be spawned", started);
}
spawned
}
#[cfg(test)]
mod tests {
use super::*;
#[test]
fn the_row_reads_as_one_line_and_a_page_cache_read_is_called_cached() {
let r = Report {
seq_write_bps: 812e6,
seq_read_bps: 1.9e9,
read_how: ReadHow::Direct,
sync_ops_per_s: 3100.4,
elapsed: Duration::from_millis(1600),
file_bytes: 256 << 20,
write_elapsed: Duration::from_millis(330),
read_elapsed: Duration::from_millis(141),
sync_ops: 1550,
sync_elapsed: Duration::from_millis(500),
};
assert_eq!(
r.row(),
"seq w 812 MB/s \u{b7} r 1.9 GB/s \u{b7} 4k sync 3100 ops/s (1.6 s)"
);
let cached = Report {
seq_read_bps: 7.2e9,
..r.clone()
};
assert_eq!(
cached.row(),
"seq w 812 MB/s \u{b7} r cached \u{b7} 4k sync 3100 ops/s (1.6 s)"
);
assert_eq!(fmt_rate(512e3), "512 kB/s");
assert!(!Outcome::Done(r).row().1);
assert_eq!(
Outcome::Skipped {
free: 800 << 20,
need: 1 << 30
}
.row(),
("skipped: 800M free (< 1.0G)".to_owned(), false)
);
assert_eq!(
Outcome::Failed("write: No space left on device".into()).row(),
("failed: write: No space left on device".to_owned(), true)
);
}
#[test]
fn the_skip_rule_is_free_space_below_the_need() {
assert!(should_skip((1 << 30) - 1, 1 << 30));
assert!(!should_skip(1 << 30, 1 << 30));
assert!(!should_skip(u64::MAX, MIN_FREE_BYTES));
// On a filesystem with next to nothing free the run says skipped and
// touches no file: asking for more than any disk has.
let dir = tempfile::tempdir().unwrap();
let out = run_sized(dir.path(), 4 << 20, u64::MAX, Duration::from_millis(1));
assert!(
matches!(out, Outcome::Skipped { need: u64::MAX, .. }),
"{out:?}"
);
assert!(!dir.path().join(DIR_NAME).exists());
}
#[test]
fn a_small_run_measures_three_numbers_and_leaves_nothing_behind() {
let dir = tempfile::tempdir().unwrap();
let out = run_sized(dir.path(), 4 << 20, 0, Duration::from_millis(50));
let Outcome::Done(r) = out else {
panic!("{out:?}")
};
assert!(
r.seq_write_bps > 0.0 && r.seq_read_bps > 0.0 && r.sync_ops_per_s > 0.0,
"{r:?}"
);
assert!(r.elapsed >= Duration::from_millis(50));
assert!(
!dir.path().join(DIR_NAME).exists(),
"the probe directory is removed"
);
let text = r.row();
assert!(
text.starts_with("seq w ") && text.contains(" 4k sync "),
"{text}"
);
}
/// A fixture file the read leg is pointed at: 4 MiB + a tail that is
/// not a whole block, so the O_DIRECT short read at EOF is exercised.
fn fixture(dir: &Path) -> (PathBuf, u64) {
let path = dir.join("fixture");
let body = pattern(4 * (1 << 20) + 1234);
std::fs::write(&path, &body).unwrap();
File::open(&path).unwrap().sync_all().unwrap();
(path, body.len() as u64)
}
#[test]
fn the_read_leg_reads_the_fixture_back_off_the_device_with_o_direct() {
// A scratch directory on the real disk; `/tmp` is often tmpfs, which
// refuses O_DIRECT — pick a real filesystem when one is at hand.
let base = std::env::var_os("GUNNAR_DISKTEST_FIXTURE_DIR").map(PathBuf::from);
let dir = match base {
Some(b) => tempfile::tempdir_in(b).unwrap(),
None => tempfile::tempdir().unwrap(),
};
let (path, len) = fixture(dir.path());
let mut buf = AlignedBuf::new(CHUNK, DIRECT_ALIGN);
assert_eq!(
buf.as_slice().as_ptr() as usize % DIRECT_ALIGN,
0,
"the buffer is 4 KiB-aligned"
);
assert_eq!(buf.as_slice().len(), CHUNK);
let (got, secs, how) = read_back(&path, &mut buf, true).unwrap();
assert_eq!(got, len, "every byte, including the partial last block");
assert!(secs > 0.0);
// The bytes that landed in the buffer are the file's tail (the last,
// short read) — the leg really read, it did not just count.
let tail = &pattern(len as usize)[(len as usize) - 1234..];
assert_eq!(&buf.as_slice()[..1234], tail);
let fs_direct = OpenOptions::new()
.read(true)
.custom_flags(OFlags::DIRECT.bits() as i32)
.open(&path)
.is_ok();
assert_eq!(
how,
if fs_direct {
ReadHow::Direct
} else {
ReadHow::Dropped
},
"O_DIRECT when the filesystem takes it, the fallback when not"
);
}
#[test]
fn the_read_leg_falls_back_to_dropping_the_cache_and_reading_buffered() {
let dir = tempfile::tempdir().unwrap();
let (path, len) = fixture(dir.path());
let mut buf = AlignedBuf::new(CHUNK, DIRECT_ALIGN);
let (got, _, how) = read_back(&path, &mut buf, false).unwrap();
assert_eq!(got, len);
assert_eq!(how, ReadHow::Dropped);
assert!(
read_back(Path::new("/nonexistent/korp/fixture"), &mut buf, false).is_err(),
"a missing file is an error, not a panic"
);
}
#[test]
fn a_full_run_names_how_it_read() {
let dir = tempfile::tempdir().unwrap();
let out = run_sized(dir.path(), 4 << 20, 0, Duration::from_millis(20));
let Outcome::Done(r) = out else {
panic!("{out:?}")
};
assert!(matches!(r.read_how, ReadHow::Direct | ReadHow::Dropped));
assert!(r.seq_read_bps > 0.0);
}
#[test]
fn a_missing_root_is_a_failure_row_not_a_panic() {
let out = run_sized(
Path::new("/nonexistent/korp"),
4 << 20,
0,
Duration::from_millis(1),
);
assert!(matches!(out, Outcome::Failed(_)), "{out:?}");
}
}
/// **The repeatability of this test, measured, not assumed.**
///
/// The twin election (`private-gunnar-ops`, `cargo xtask elect twin`) refuses
/// when two boxes' disks differ by less than the noise, so somebody has to
/// know what the noise IS. This is how it was measured: N production-sized
/// runs back to back on one idle filesystem, printing the spread of each leg
/// as a percentage of its own median. `--ignored` because it takes minutes
/// and writes gigabytes; it is a MEASUREMENT, re-runnable on any box.
///
/// ```text
/// cargo test -p gunnar-server --lib disktest::noise -- --ignored --nocapture
/// ```
#[cfg(test)]
mod noise {
use super::*;
#[test]
#[ignore = "a measurement: minutes, and gigabytes written"]
fn the_spread_of_ten_runs_on_one_idle_disk() {
let runs: usize = std::env::var("GUNNAR_DISKTEST_RUNS")
.ok()
.and_then(|v| v.parse().ok())
.unwrap_or(10);
let dir = match std::env::var_os("GUNNAR_DISKTEST_FIXTURE_DIR") {
Some(b) => tempfile::tempdir_in(PathBuf::from(b)).unwrap(),
None => tempfile::tempdir().unwrap(),
};
let mut w = Vec::with_capacity(runs);
let mut r = Vec::with_capacity(runs);
let mut s = Vec::with_capacity(runs);
for i in 0..runs {
match run_sized(dir.path(), FILE_BYTES, 0, SYNC_WINDOW) {
Outcome::Done(rep) => {
println!("run {i:2} {}", rep.row());
w.push(rep.seq_write_bps);
r.push(rep.seq_read_bps);
s.push(rep.sync_ops_per_s);
}
other => panic!("run {i}: {other:?}"),
}
}
for (name, mut v) in [("seq write", w), ("seq read", r), ("4k sync", s)] {
v.sort_by(|a, b| a.partial_cmp(b).unwrap());
let median = v[v.len() / 2];
let (lo, hi) = (v[0], v[v.len() - 1]);
println!(
"{name:<10} median {median:>14.0} min {lo:>14.0} max {hi:>14.0} \
spread {:>6.1}% of median",
(hi - lo) / median * 100.0
);
}
}
}