crates/gunnar-lfs/tests/streaming.rs
//! **A gigabyte moves through this server in a constant amount of memory.**
//!
//! Asserted against the kernel's own high-water mark for this process
//! (`VmHWM` in `/proc/self/status`), not against a counter gunnar keeps, and not
//! against `Vec::capacity` somewhere. "It streams" is exactly the claim a green
//! unit test cannot make: every buffer in the module could be replaced by one
//! sized from the header and every other test in this crate would still pass.
//!
//! # Nothing here materialises a gigabyte, on either side
//!
//! * the **input** is [`Zeros`], a `Read` that synthesises the pkt-line frames
//! as they are asked for — the test process never holds the payload either,
//! so a failure means the *server* grew, not the harness;
//! * the **store** is [`DiscardingLfsStore`], which holds nothing at all, so
//! the store cannot be what makes the measurement pass;
//! * the **oid** is the published SHA-256 of 2^30 zero bytes, so the upload
//! genuinely commits rather than being abandoned at the verifier — the full
//! success path is what is measured, including the hash.
//!
//! A test in this tree once OOM'd a 499 GB machine by allocating what a header
//! claimed. This is the path most likely to repeat that.
use std::io::{BufReader, Read};
use gunnar_lfs::{DiscardingLfsStore, Operation, Session, CHUNK_BYTES};
/// One gibibyte.
const GIB: u64 = 1024 * 1024 * 1024;
/// `sha256` of 2^30 zero bytes. From coreutils —
/// `head -c 1073741824 /dev/zero | sha256sum` — not from gunnar, so a hasher
/// that agreed only with itself would not satisfy it.
const GIB_OF_ZEROS: &str = "49bc20df15e412a64472421e13fe86ff1c5165e18b2afccf160d4dc19fe68a14";
/// How much resident memory the server is allowed to grow by while a gigabyte
/// passes. Generous — the honest bound is a few hundred KiB (one `CHUNK_BYTES`
/// slab each way, plus the hasher) — because the point is to separate
/// *constant* from *proportional*, and 1 GiB is 16384× this.
const RSS_BUDGET_BYTES: u64 = 64 * 1024 * 1024;
/// The process's peak resident set size, in bytes.
///
/// Hard error rather than `Option`: a probe that quietly returned "unknown"
/// would turn this whole file into a test that passes having measured nothing.
fn peak_rss_bytes() -> u64 {
let status = std::fs::read_to_string("/proc/self/status").expect(
"this test measures real process memory and needs /proc/self/status; \
it is not skipped, because a skipped memory proof is worse than none",
);
let line = status
.lines()
.find(|line| line.starts_with("VmHWM:"))
.expect("VmHWM is not in /proc/self/status");
let kib: u64 = line
.split_whitespace()
.nth(1)
.and_then(|value| value.parse().ok())
.unwrap_or_else(|| panic!("could not parse {line:?}"));
kib * 1024
}
/// The whole upload conversation for a `total`-byte object of zeros, generated
/// as it is read. Peak memory: one pkt-line.
struct Zeros {
/// Bytes still to hand out of the current frame.
frame: Vec<u8>,
at: usize,
/// Payload bytes not yet framed.
remaining: u64,
stage: Stage,
}
#[derive(PartialEq)]
enum Stage {
Preamble,
Payload,
Trailer,
Done,
}
impl Zeros {
fn new(total: u64, oid: &str) -> Zeros {
let mut preamble = Vec::new();
gunnar_pktline::write_text(&mut preamble, b"version 1").unwrap();
gunnar_pktline::write_flush(&mut preamble).unwrap();
gunnar_pktline::write_text(&mut preamble, format!("put-object {oid}").as_bytes()).unwrap();
gunnar_pktline::write_text(&mut preamble, format!("size={total}").as_bytes()).unwrap();
gunnar_pktline::write_delim(&mut preamble).unwrap();
Zeros {
frame: preamble,
at: 0,
remaining: total,
stage: Stage::Preamble,
}
}
fn refill(&mut self) {
self.frame.clear();
self.at = 0;
match self.stage {
Stage::Preamble | Stage::Payload if self.remaining > 0 => {
self.stage = Stage::Payload;
let take = self.remaining.min(CHUNK_BYTES as u64) as usize;
// The frame is reused; it never grows past one pkt-line.
gunnar_pktline::write_data(&mut self.frame, &vec![0u8; take]).unwrap();
self.remaining -= take as u64;
}
Stage::Preamble | Stage::Payload => {
self.stage = Stage::Trailer;
gunnar_pktline::write_flush(&mut self.frame).unwrap();
gunnar_pktline::write_text(&mut self.frame, b"quit").unwrap();
gunnar_pktline::write_flush(&mut self.frame).unwrap();
}
Stage::Trailer => self.stage = Stage::Done,
Stage::Done => {}
}
}
}
impl Read for Zeros {
fn read(&mut self, buf: &mut [u8]) -> std::io::Result<usize> {
if self.at == self.frame.len() {
if self.stage == Stage::Done {
return Ok(0);
}
self.refill();
if self.at == self.frame.len() {
return Ok(0);
}
}
let n = buf.len().min(self.frame.len() - self.at);
buf[..n].copy_from_slice(&self.frame[self.at..self.at + n]);
self.at += n;
Ok(n)
}
}
#[test]
fn a_gibibyte_uploads_in_constant_memory() {
// Warm the allocator and the harness so the baseline is not measuring
// startup. VmHWM only ever rises, so anything already peaked is excluded
// from the delta by construction.
let store = DiscardingLfsStore::new();
let before = peak_rss_bytes();
let mut input = BufReader::new(Zeros::new(GIB, GIB_OF_ZEROS));
let mut output = Vec::new();
let summary = Session::new(&store, Operation::Upload)
.serve(&mut input, &mut output)
.expect("the session must complete");
let after = peak_rss_bytes();
let grew = after.saturating_sub(before);
// APPLIED OUTPUT FIRST. A session that read nothing would have the best
// memory profile in the world, so the transfer is asserted before the
// memory is.
assert_eq!(
summary.bytes_received, GIB,
"the session did not move a gibibyte"
);
assert_eq!(summary.objects_received, 1);
assert!(summary.clean_quit);
assert_eq!(store.accepted(), GIB, "the store was not handed a gibibyte");
assert_eq!(
store.committed(),
1,
"the object never committed — the hash or the size did not check out, \
so the success path was not the path measured"
);
let wire = String::from_utf8_lossy(&output);
assert!(
wire.contains("status 200"),
"the upload was not answered 200: {wire}"
);
assert!(
grew < RSS_BUDGET_BYTES,
"peak RSS grew by {grew} bytes ({:.1} MiB) while moving {GIB} bytes; \
the budget is {RSS_BUDGET_BYTES}. Something on this path is sized from \
the object rather than from CHUNK_BYTES.",
grew as f64 / (1024.0 * 1024.0)
);
eprintln!(
"a gibibyte moved; peak RSS grew {grew} bytes ({:.2} MiB), budget {:.0} MiB",
grew as f64 / (1024.0 * 1024.0),
RSS_BUDGET_BYTES as f64 / (1024.0 * 1024.0)
);
// ── the probe's own identity-value check ───────────────────────────────
//
// Everything above rests on `peak_rss_bytes()` being able to MOVE. A probe
// stuck at its first reading would report `grew == 0` for any server,
// including one that buffered the whole gibibyte. So: allocate a quarter of
// a gibibyte, touch every page so it is resident, and require the probe to
// notice. This runs last, so it cannot inflate the measurement above.
let baseline = peak_rss_bytes();
let mut ballast = vec![0u8; 256 * 1024 * 1024];
for page in ballast.chunks_mut(4096) {
page[0] = 1;
}
std::hint::black_box(&ballast);
let moved = peak_rss_bytes().saturating_sub(baseline);
drop(ballast);
assert!(
moved > 200 * 1024 * 1024,
"the RSS probe reported only {moved} bytes of growth for a resident \
256 MiB allocation — it is not measuring anything, so neither is the \
assertion above"
);
}