crates/gunnar-server/tests/twin_apply_io.rs
//! **What one applied entry costs the twin's disk, counted at the system call** (perf
//! twin-apply-fsync, 2026-10-06).
//!
//! The load test on the UpCloud mockup (2026-10-06 20:25 UTC) measured the twin — 1 vCPU, one
//! HDD-tier volume — doing ~10 write operations and ~3 flushes PER transaction-log entry while it
//! followed a storm of small pushes and sign-ups: ~2,800 write IOPS, 0.03–0.2 cores used, and a
//! lag that grew to ~93,000 entries. It was bound by its own fsyncs, not by its CPU.
//!
//! This file counts the same thing in one process, where it can be held to a bound: a primary
//! takes a storm of sign-ups (principal, key, namespace, repository — each a run of
//! `ControlBlob` + `ControlDoc` frames) and small pushes (`Pack` + `Refs`), then a twin follows
//! it through the same `Follower` the appliance runs. The follow runs in a CHILD of this test
//! under `strace -f -y`, so every `write`-family and `fsync`-family call is the kernel's own
//! count, attributed by path to the twin's data root — no counter in the code under test can be
//! forgotten or lie. The phase is fenced by two marker `statx` calls.
//!
//! MEASURED before the fix (oden, xfs, debug build): 1,351 entries applied with 12,998 writes
//! (9.6 per entry: the frame's header, meta, piece and trailer as four writes, one index write,
//! ~5.7 redb page writes per control frame) and 2,503 fsyncs (1.85 per entry: one per frame in
//! the log, one per redb commit).
//!
//! The bound (group commit): an entry is written into the log with the others of its batch, the
//! batch is made durable with ONE `fdatasync`, and the control frames it carries are applied in
//! ONE durable redb commit — so writes per entry ≤ [`MAX_WRITES_PER_ENTRY`] amortized and fsyncs
//! per entry ≤ [`MAX_FSYNCS_PER_ENTRY`].
//!
//! Needs `strace` (it is on every build host); without it the test FAILS by name rather than
//! passing about nothing.
use std::sync::Arc;
use gunnar_core::{HashKind, NewRepo, Principal, PrincipalKind, RepoRef};
use gunnar_server::stores::ServerStores;
use gunnar_server::twin::{self, Follower, LocalSource, Progress};
use gunnar_store::conformance::pack::PackBuilder;
/// Sign-ups in the storm.
const ACCOUNTS: usize = 50;
/// Small pushes per account.
const PUSHES: usize = 2;
/// The bound on write calls per applied entry, amortized over the storm.
const MAX_WRITES_PER_ENTRY: f64 = 2.0;
/// The bound on fsync-family calls per applied entry: one per batch for the log and one for the
/// control store, batches of tens of entries at least.
const MAX_FSYNCS_PER_ENTRY: f64 = 0.1;
const INNER: &str = "the_storm_followed_by_a_twin_inner";
const ENV_ROOT: &str = "TWIN_APPLY_IO_ROOT";
fn mark(what: &str) {
let _ = std::fs::metadata(format!("/MARK-twin-apply-io-{what}"));
}
fn key_line() -> String {
let key = gunnar_core::ssh_key::PrivateKey::from_openssh(gunnar_core::mint_ed25519().unwrap())
.unwrap();
key.public_key().to_openssh().unwrap()
}
fn push(stores: &ServerStores, repo: &RepoRef, content: &[u8]) {
let entry = stores.catalog().get(repo).unwrap().unwrap();
let store = stores.repo_store(&entry).expect("store");
let hash = store.hash_kind();
let id = gix_object::compute_hash(hash, gix_object::Kind::Blob, content).expect("hash");
let mut pb = PackBuilder::new(hash);
pb.object(gix_object::Kind::Blob, content.to_vec());
store
.serve()
.put_pack(&pb.build().expect("pack").bytes)
.expect("pack");
let oid = id.as_bytes().to_vec();
let old = store
.serve()
.refs()
.unwrap()
.into_iter()
.find(|r| r.name == "refs/heads/main")
.and_then(|r| r.oid);
store
.serve()
.update_ref("refs/heads/main", old.as_deref(), Some(&oid))
.expect("ref");
}
/// The child: the storm on a primary, then a twin follows it; the follow is fenced by markers.
/// Run only by [`the_twin_applies_a_storm_with_few_writes_and_fsyncs_per_entry`].
#[test]
#[ignore = "run under strace by the_twin_applies_a_storm_with_few_writes_and_fsyncs_per_entry"]
fn the_storm_followed_by_a_twin_inner() {
let root = std::path::PathBuf::from(std::env::var(ENV_ROOT).expect("run by the outer test"));
let primary = Arc::new(ServerStores::open(root.join("primary")).unwrap());
let mut repos = Vec::new();
for i in 0..ACCOUNTS {
let id = format!("u{i:05}");
primary
.auth()
.add_principal(Principal::new(&id, &id, PrincipalKind::User).unwrap())
.unwrap();
primary.auth().register_key(&id, &key_line(), None).unwrap();
primary.catalog().create_namespace(&id).unwrap();
let r = RepoRef::new(&id, "r").unwrap();
primary
.create_repo(NewRepo::new(r.clone()).hash_kind(HashKind::Sha256), false)
.unwrap();
repos.push(r);
}
for p in 0..PUSHES {
for (i, r) in repos.iter().enumerate() {
push(&primary, r, format!("push {p} of {i}\n").as_bytes());
}
}
let head = primary.txlog().head_seq();
let tw = Arc::new(ServerStores::open(root.join("twin")).unwrap());
twin::become_twin(&tw, "primary.test:50051").unwrap();
let from = tw.txlog().head_seq();
let f = Follower::new(
Arc::clone(&tw),
Box::new(LocalSource {
stores: Arc::clone(&primary),
retained_from: 0,
}),
);
mark("follow");
loop {
match f.step().unwrap() {
Progress::CaughtUp { .. } => break,
Progress::Applied { .. } => {}
other => panic!("the twin did not follow: {other:?}"),
}
}
mark("done");
assert_eq!(
tw.txlog().head_seq(),
head,
"the twin is at the primary's head"
);
println!("\nENTRIES {}", head - from);
}
/// `(writes, fsyncs)` the strace log at `trace` saw between the markers on files under `under`.
fn count(trace: &str, under: &str) -> (u64, u64, Vec<String>) {
let (mut on, mut writes, mut fsyncs) = (false, 0u64, 0u64);
let mut by_file: std::collections::BTreeMap<String, (u64, u64)> = Default::default();
for line in trace.lines() {
if line.contains("/MARK-twin-apply-io-follow") {
on = true;
continue;
}
if line.contains("/MARK-twin-apply-io-done") {
break;
}
if !on {
continue;
}
// `<pid> <call>(<fd><<path>>, …` — strace -y spells the fd's path in angle brackets.
let rest = line.split_once(' ').map(|(_, r)| r).unwrap_or(line);
let Some((call, args)) = rest.split_once('(') else {
continue;
};
let is_write = matches!(
call,
"write" | "writev" | "pwrite64" | "pwritev" | "pwritev2"
);
let is_sync = matches!(call, "fsync" | "fdatasync" | "sync_file_range" | "syncfs");
if !is_write && !is_sync {
continue;
}
let Some(path) = args
.split_once('<')
.and_then(|(_, p)| p.split_once('>'))
.map(|(p, _)| p)
else {
continue;
};
if !path.starts_with(under) {
continue;
}
let file = path[under.len()..].trim_start_matches('/').to_owned();
let file = if file.starts_with("txlog/") {
if file.ends_with(".idx") {
"txlog/*.idx".into()
} else {
"txlog/*.log".into()
}
} else {
file
};
let e = by_file.entry(file).or_default();
if is_write {
writes += 1;
e.0 += 1;
} else {
fsyncs += 1;
e.1 += 1;
}
}
(
writes,
fsyncs,
by_file
.into_iter()
.map(|(f, (w, s))| format!("{f}: {w} writes, {s} fsyncs"))
.collect(),
)
}
#[test]
fn the_twin_applies_a_storm_with_few_writes_and_fsyncs_per_entry() {
let strace = ["/usr/bin/strace", "/bin/strace", "/usr/local/bin/strace"]
.into_iter()
.find(|p| std::path::Path::new(p).exists())
.expect("this test counts system calls with strace, and strace is not installed here — install it; a count nobody took is not a pass");
// A real filesystem, not tmpfs: the twin's volume is one.
let base = option_env!("CARGO_TARGET_TMPDIR")
.map(std::path::PathBuf::from)
.unwrap_or_else(std::env::temp_dir);
let dir = tempfile::tempdir_in(&base).unwrap();
let trace = dir.path().join("trace");
let exe = std::env::current_exe().unwrap();
let out = std::process::Command::new(strace)
.args(["-f", "-y", "-qq", "-o"])
.arg(&trace)
.args(["-e", "trace=write,writev,pwrite64,pwritev,pwritev2,fsync,fdatasync,sync_file_range,syncfs,statx,newfstatat"])
.arg(&exe)
.args(["--exact", INNER, "--ignored", "--nocapture", "--test-threads=1"])
.env(ENV_ROOT, dir.path())
.output()
.expect("strace runs");
let stdout = String::from_utf8_lossy(&out.stdout);
assert!(
out.status.success(),
"the storm under strace failed:\n{stdout}\n{}",
String::from_utf8_lossy(&out.stderr)
);
let entries: u64 = stdout
.lines()
.find_map(|l| l.split_once("ENTRIES ").map(|(_, v)| v))
.and_then(|v| v.trim().parse().ok())
.expect("the child says how many entries it applied");
let text = std::fs::read_to_string(&trace).unwrap();
let twin_root = dir
.path()
.join("twin")
.canonicalize()
.unwrap()
.to_string_lossy()
.into_owned();
let (writes, fsyncs, by_file) = count(&text, &twin_root);
let (wpe, fpe) = (
writes as f64 / entries as f64,
fsyncs as f64 / entries as f64,
);
let line = format!(
"the twin applied {entries} entries with {writes} writes ({wpe:.2} per entry, bound {MAX_WRITES_PER_ENTRY}) and {fsyncs} fsyncs ({fpe:.3} per entry, bound {MAX_FSYNCS_PER_ENTRY}); by file: {by_file:#?}"
);
eprintln!("{line}");
assert!(entries >= 1000, "the storm is the size it says: {line}");
assert!(
wpe <= MAX_WRITES_PER_ENTRY && fpe <= MAX_FSYNCS_PER_ENTRY,
"{line}"
);
}