bench-scenarios/src/audit.rs
//! Audit-log append throughput scenario.
//!
//! Drives holger's REAL [`server_lib::audit::FileAuditLog::record`] — the
//! durable audit write that fires on the hot path of **every** fetch / put /
//! delete / list — in a tight serial loop, exactly how the live server drives it
//! (one shared `Arc<dyn AuditLog>` across all request handlers, `record` taking
//! `&self` and internally `Mutex`-synchronised). It reports the append
//! throughput: records/sec (`ops_sec`) + on-disk MB/s (`mb_per_sec`).
//!
//! Each `record` encodes a one-row Arrow [`RecordBatch`](arrow), writes it to the
//! IPC stream, and flushes the underlying `BufWriter → File` to the OS — the full
//! per-request audit cost. Every event carries distinct variable-length fields
//! (identity / repo / artifact / source-ip / detail) so no accidental constant
//! folding hides work.
//!
//! This is a single-writer measurement (the server serialises audit writes behind
//! one mutex), NOT a fan-out — no parallelism primitive is used here.
//!
//! Twin correctness coverage lives in `server/lib/src/audit/*` unit tests (the
//! round-trip decode); this module is the *numbers*. After the burst the segment
//! is finalized (drop writes the IPC end-of-stream marker) and re-read from disk;
//! the row count is asserted so a broken writer fails the bench instead of
//! silently reporting a hollow number.
use std::time::Instant;
use anyhow::{Context, Result};
use server_lib::audit::{AuditAction, AuditEvent, AuditLog, FileAuditLog};
use crate::{Scale, Throughput};
/// How many audit records to append for a given [`Scale`]. Audit rows are tiny
/// and the append is fast, so the count is clamped high enough to be HEAVY
/// (hundreds of thousands of durable flushes) yet bounded so a `huge` run does
/// not spend minutes here — the append rate saturates well before the ceiling.
fn event_count(scale: &Scale) -> usize {
scale.entries.clamp(10_000, 200_000)
}
/// The action cycle — the four operations the server actually audits, so the
/// `action` column carries realistic cardinality (not one repeated value).
const ACTIONS: [AuditAction; 4] = [
AuditAction::Download,
AuditAction::Upload,
AuditAction::List,
AuditAction::Delete,
];
/// Build a realistic, **distinct** audit event for record `i` — the security 5
/// W's with per-row-varying identity, repo, artifact, source address, status and
/// byte count, so each `record` encodes genuinely different bytes.
fn event(i: usize) -> AuditEvent {
let action = ACTIONS[i % ACTIONS.len()];
AuditEvent::new(
format!("user-{}@holger", i % 512),
action,
format!("repo-{}", i % 32),
format!("crate-{i}/pkg@1.{}.0", i % 97),
format!("10.0.{}.{}:443", (i / 256) % 256, i % 256),
if i % 17 == 0 { 403 } else { 200 },
(i as u64 % 4096) * 1024,
)
.with_detail("bench")
}
/// Append [`event_count`] events through a real [`FileAuditLog`], returning
/// the throughput (records/sec = `ops_sec`, on-disk bytes/sec = `mb_per_sec`).
/// The segment is finalized and re-read; the round-tripped row count is asserted
/// so a broken append path fails the bench rather than reporting a hollow number.
pub fn audit_append_throughput(scale: &Scale) -> Result<Throughput> {
let n = event_count(scale);
// The segment lives under a scratch dir the OS reclaims on drop.
let dir = tempfile::tempdir().context("audit bench: tempdir")?;
// Pre-build the events so field synthesis is not in the timed region.
let events: Vec<AuditEvent> = (0..n).map(event).collect();
let log = FileAuditLog::new(dir.path()).context("audit bench: open the audit file")?;
let segment = dir.path().join(server_lib::audit::AUDIT_FILE);
let start = Instant::now();
for ev in events {
log.record(ev)
.map_err(|e| anyhow::anyhow!("audit bench: record failed: {e}"))?;
}
let elapsed = start.elapsed();
// Read every row back from disk.
drop(log);
let read_back = FileAuditLog::read_dir(dir.path()).context("audit bench: read segment back")?;
anyhow::ensure!(
read_back.len() == n,
"audit bench: expected {n} records back, read {} — the append path is broken",
read_back.len()
);
// Audit records carry no payload of their own, so the honest throughput
// denominator is the real IPC bytes the writer put on disk.
let bytes = std::fs::metadata(&segment).map(|m| m.len()).unwrap_or(0);
let s = elapsed.as_secs_f64().max(f64::MIN_POSITIVE);
Ok(Throughput {
ops: n as u64,
elapsed,
bytes,
ops_sec: n as f64 / s,
mb_per_sec: bytes as f64 / s / 1_048_576.0,
})
}
/// **The same appends from every core at once** (stinky 30): [`event_count`]
/// events recorded through ONE shared [`FileAuditLog`] by a gatling fan-out —
/// what the live server does when every door's request thread audits at once.
/// The row count is read back and the hash chain verified, so a combining writer
/// that dropped, duplicated or mis-chained a row fails the bench.
pub fn audit_append_concurrent_throughput(scale: &Scale) -> Result<Throughput> {
let n = event_count(scale);
let dir = tempfile::tempdir().context("audit bench: tempdir")?;
let events: Vec<AuditEvent> = (0..n).map(event).collect();
let log = FileAuditLog::new(dir.path()).context("audit bench: open the audit file")?;
let segment = dir.path().join(server_lib::audit::AUDIT_FILE);
let start = Instant::now();
let failed = gatling::gatling_forkjoin::gatling_for_each(n, 0, |i| {
log.record(events[i].clone()).is_err()
});
let elapsed = start.elapsed();
anyhow::ensure!(
!failed.iter().any(|f| *f),
"audit bench: a concurrent record failed"
);
drop(log);
let read_back = FileAuditLog::read_dir(dir.path()).context("audit bench: read segment back")?;
anyhow::ensure!(
read_back.len() == n,
"audit bench: expected {n} records back, read {}",
read_back.len()
);
let report = server_lib::audit::chain::verify_file(&segment, &[])
.map_err(|e| anyhow::anyhow!("audit bench: the chain cannot be read: {e}"))?;
anyhow::ensure!(
report.is_intact(),
"audit bench: the chain does not verify: {}",
report.summary()
);
let bytes = std::fs::metadata(&segment).map(|m| m.len()).unwrap_or(0);
let s = elapsed.as_secs_f64().max(f64::MIN_POSITIVE);
Ok(Throughput {
ops: n as u64,
elapsed,
bytes,
ops_sec: n as f64 / s,
mb_per_sec: bytes as f64 / s / 1_048_576.0,
})
}
#[cfg(test)]
mod tests {
use super::*;
/// Red-when-broken correctness gate for the bench: the real audit append path
/// produces a positive record rate over real on-disk IPC bytes, and the
/// round-trip row-count assertion inside `audit_append_throughput` holds (a
/// broken writer would `ensure!`-fail there rather than reach these asserts).
#[test]
fn audit_append_yields_real_throughput() {
let scale = Scale::sane();
let t = audit_append_throughput(&scale).expect("audit append bench runs");
// event_count clamps to a 10k floor regardless of the sane entry count.
assert!(
t.ops >= 10_000,
"at least the clamped record floor was appended: {}",
t.ops
);
assert!(t.ops_sec > 0.0, "a positive record/sec rate: {}", t.ops_sec);
assert!(t.bytes > 0, "real IPC bytes hit disk: {}", t.bytes);
assert!(
t.mb_per_sec > 0.0,
"a positive on-disk MB/s: {}",
t.mb_per_sec
);
}
}