crates/gunnar-server/tests/audit_read_io.rs
//! **What the control plane's audit trail costs main's disk, per decision, counted at the
//! system call** (P3 of the load test, 2026-10-06).
//!
//! The UpCloud mockup's load test found every authorization decision — reads included — written
//! to the replicated trail (`grpc_gate::ChainLedger` → `AuditChain` → a `txlog` `Append` frame,
//! fdatasync'd, then the line into `audit-chain.jsonl`, `sync_all`'d): ~160–360 KiB of disk per
//! log entry on main, and every one of those frames is the twin's to store too. A console page
//! load, monetize's `Usage.Read` per tenant every 300 s and every other read became two flushes.
//!
//! Two tests:
//!
//! * **reads** (in process) — [`READS`] plain reads (allowed at level `read`) by ONE principal on
//! ONE scope: "who read what" stays in the trail (GUN-044, the enterprise audit row cov-03), but
//! as at most [`MAX_READ_FRAMES`] entries — the first read of the window at once, and one entry
//! with the count and the first and last time of the repeats (`grpc_gate::READ_WINDOW`) — so at
//! most that many frames and fsyncs (each sequential frame is at most one `fdatasync`).
//! * **security** (in a CHILD under `strace -f -y`: the kernel's own count of every `write`- and
//! `fsync`-family call under the primary's data root, fenced by markers) — [`SECURITY`]
//! decisions that are one entry each (refusals at any level, allowed writes and admin calls, an
//! override), made by [`THREADS`] request threads at once: every one lands, in one chain that
//! verifies, each line framed — and they share their fsyncs (group commit), at most
//! [`MAX_SECURITY_FSYNCS_PER_DECISION`] each. The count is of what the group commit SHARES:
//! each leader is held until all [`THREADS`] writers have staged a frame
//! (`gunnar_core::txlog::TEST_GROUP_ENV`), so it does not depend on how `strace` and the
//! scheduler interleaved them (round 5.2: unheld, it read 0.68–0.76 on dwarf and oden — luck,
//! not a count).
//!
//! MEASURED before (loki, xfs, debug): 300 plain reads = 300 frames, 600 fsyncs, 250,880 B (836 B and
//! two flushes per read: the frame + fdatasync, the line + `sync_all`); 240 security decisions on 8
//! threads = 480 fsyncs, 2.0 per decision (the chain held its lock across both flushes).
use std::sync::Arc;
use gunnar_grpc::auth::AuditRecord;
use gunnar_server::stores::ServerStores;
const READS: usize = 300;
/// The first read of the window, and the one entry that counts the rest.
const MAX_READ_FRAMES: u64 = 4;
const SECURITY: usize = 240;
const THREADS: usize = 8;
/// With every writer there for each fsync: one fsync per [`THREADS`] decisions (0.125), the
/// stragglers at the end allowed for — at most twice that.
const MAX_SECURITY_FSYNCS_PER_DECISION: f64 = 0.25;
const INNER: &str = "the_security_decisions_inner";
const ENV_ROOT: &str = "AUDIT_READ_IO_ROOT";
fn mark(what: &str) {
let _ = std::fs::metadata(format!("/MARK-audit-read-io-{what}"));
}
fn rec<'a>(method: &'a str, level: &'a str, decision: &'a str, reason: &'a str) -> AuditRecord<'a> {
AuditRecord {
method,
principal: "alice",
fingerprint: "SHA256:alice",
scope: "namespace:alice",
level,
auth_mode: "required",
decision,
reason,
}
}
/// **Repeated plain reads: a few entries that still say who read what, and when.**
#[test]
fn repeated_plain_reads_are_a_few_entries_that_still_say_who_read_what_and_when() {
let dir = tempfile::tempdir().unwrap();
let stores = Arc::new(ServerStores::open(dir.path()).unwrap());
let chain = Arc::new(
gunnar_core::AuditChain::open_framed(stores.audit_path())
.expect("the primary's chain is framed"),
);
let ledger = gunnar_server::control_audit_ledger(Arc::clone(&chain));
let (h0, t0) = (stores.txlog().head_seq(), gunnar_core::now_ms());
for _ in 0..READS {
ledger.record(rec("Usage.Read", "read", "allow", ""));
}
// The ledger goes (a shutdown): the open window's count is written, not lost with it.
drop(ledger);
let t1 = gunnar_core::now_ms();
let frames = stores.txlog().head_seq() - h0;
let entries = chain.entries().unwrap();
assert_eq!(
chain.verify().unwrap() as u64,
frames,
"every entry is a frame"
);
let mut reads = 0u64;
for e in &entries {
assert_eq!(
(
e.principal.as_str(),
e.scope.as_str(),
e.method.as_str(),
e.decision.as_str()
),
("alice", "namespace:alice", "Usage.Read", "allow"),
"who read what: {e:?}"
);
assert!(e.at_unix_ms >= t0 && e.at_unix_ms <= t1, "when: {e:?}");
reads += match e.reason.strip_prefix(gunnar_server::grpc_gate::AGGREGATED) {
Some(rest) => {
let n: u64 = rest
.split_whitespace()
.next()
.and_then(|n| n.parse().ok())
.unwrap_or_else(|| panic!("a count: {}", e.reason));
assert!(
rest.contains("first ") && rest.contains("last "),
"the window's first and last read: {}",
e.reason
);
n
}
None => 1,
};
}
let line = format!(
"{READS} plain reads by one principal on one scope: {frames} frames, {} entries {:?}",
entries.len(),
entries.iter().map(|e| e.reason.clone()).collect::<Vec<_>>()
);
eprintln!("{line}");
assert!(
frames <= MAX_READ_FRAMES,
"repeated reads are aggregated: {line}"
);
assert_eq!(
reads, READS as u64,
"every read is counted in the trail: {line}"
);
}
/// The child of the security test: a primary's own framed chain behind the box's own ledger, as
/// `serve` wires it.
#[test]
#[ignore = "run under strace by concurrent_security_decisions_land_one_each_and_share_their_fsyncs"]
fn the_security_decisions_inner() {
let root = std::path::PathBuf::from(std::env::var(ENV_ROOT).expect("run by the outer test"));
let stores = Arc::new(ServerStores::open(&root).unwrap());
let chain = Arc::new(
gunnar_core::AuditChain::open_framed(stores.audit_path())
.expect("the primary's chain is framed"),
);
let ledger = gunnar_server::control_audit_ledger(Arc::clone(&chain));
mark("security");
let h1 = stores.txlog().head_seq();
std::thread::scope(|s| {
for t in 0..THREADS {
let ledger = Arc::clone(&ledger);
s.spawn(move || {
for i in 0..SECURITY / THREADS {
match (t + i) % 4 {
0 => ledger.record(rec("Repos.Get", "read", "deny", "no grant")),
1 => ledger.record(rec("Repos.Create", "write", "allow", "")),
2 => ledger.record(rec("Access.Grant", "admin", "allow", "")),
_ => ledger.record(rec(
"Mirror.LogRetain",
"admin",
"allow",
"OVERRIDE: past twin-slow",
)),
}
}
});
}
});
mark("done");
let h2 = stores.txlog().head_seq();
let entries = chain.verify().expect("the chain verifies");
println!("\nFRAMES security {} ENTRIES {entries}", h2 - h1);
}
/// `(writes, fsyncs, bytes)` the strace log saw in phase `phase` on files under `under`.
fn count(trace: &str, under: &str, phase: &str, next: &str) -> (u64, u64, u64, Vec<String>) {
let (mut on, mut writes, mut fsyncs, mut bytes) = (false, 0u64, 0u64, 0u64);
let mut by_file: std::collections::BTreeMap<String, (u64, u64, u64)> = Default::default();
for line in trace.lines() {
if line.contains(&format!("/MARK-audit-read-io-{phase}")) {
on = true;
continue;
}
if line.contains(&format!("/MARK-audit-read-io-{next}")) {
break;
}
if !on {
continue;
}
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 n: u64 = line
.rsplit_once(" = ")
.and_then(|(_, r)| r.trim().parse().ok())
.unwrap_or(0);
let e = by_file.entry(file).or_default();
if is_write {
writes += 1;
bytes += n;
e.0 += 1;
e.2 += n;
} else {
fsyncs += 1;
e.1 += 1;
}
}
(
writes,
fsyncs,
bytes,
by_file
.into_iter()
.map(|(f, (w, s, b))| format!("{f}: {w} writes, {s} fsyncs, {b} B"))
.collect(),
)
}
#[test]
fn concurrent_security_decisions_land_one_each_and_share_their_fsyncs() {
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");
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 root = dir.path().join("primary");
std::fs::create_dir_all(&root).unwrap();
let trace = dir.path().join("trace");
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(std::env::current_exe().unwrap())
.args(["--exact", INNER, "--ignored", "--nocapture", "--test-threads=1"])
.env(ENV_ROOT, &root)
// Each leader waits for the whole group (all THREADS writers staged) before its fsync.
.env(gunnar_core::txlog::TEST_GROUP_ENV, THREADS.to_string())
.output()
.expect("strace runs");
let stdout = String::from_utf8_lossy(&out.stdout);
assert!(
out.status.success(),
"the decisions under strace failed:\n{stdout}\n{}",
String::from_utf8_lossy(&out.stderr)
);
let nums: Vec<u64> = stdout
.lines()
.find_map(|l| l.split_once("FRAMES ").map(|(_, v)| v.to_owned()))
.expect("the child says what it framed")
.split_whitespace()
.filter_map(|w| w.parse().ok())
.collect();
let (security_frames, entries) = (nums[0], nums[1]);
let text = std::fs::read_to_string(&trace).unwrap();
let under = root.canonicalize().unwrap().to_string_lossy().into_owned();
let (sw, sf, sb, s_by) = count(&text, &under, "security", "done");
let line = format!(
"{SECURITY} security decisions on {THREADS} threads: {security_frames} frames, {sw} writes, {sf} fsyncs ({:.3} per decision, bound {MAX_SECURITY_FSYNCS_PER_DECISION}), {sb} B ({s_by:?}); the chain holds {entries} entries",
sf as f64 / SECURITY as f64
);
eprintln!("{line}");
assert_eq!(
security_frames, SECURITY as u64,
"every security decision is framed, one each: {line}"
);
assert_eq!(
entries, SECURITY as u64,
"the chain holds every security decision: {line}"
);
assert!(
sf as f64 / SECURITY as f64 <= MAX_SECURITY_FSYNCS_PER_DECISION,
"concurrent decisions share their fsyncs: {line}"
);
}