crates/storage-git-gix/parked-gix-2/gunnar-wire-gix/tests/fetch_memory.rs
//! **Does a fetch stay inside a ceiling the *server* chose?**
//!
//! Every other memory assertion in this crate is about a shape the server
//! controls — `S-004`'s exclusion closure, the walk budget. These are about the
//! two sizes the server does **not** control:
//!
//! * **what the peer sends.** A v2 argument section and a v0 `want`/`have`
//! exchange each end when the *client* sends a flush-pkt. Nothing else ends
//! them. A client that sends `have <oid>` forever and never flushes made the
//! server allocate forever, before a single object was read;
//! * **how big one object is.** The whole-object pack writer materialises an
//! object's inflated content, so peak memory was the largest object in the
//! transfer.
//!
//! # Why the big blob is compressible, which is the opposite of `streaming.rs`
//!
//! `tests/streaming.rs` needs an **in**compressible blob, because it measures
//! bytes leaving the writer and a blob of zeroes would slip under any bound
//! while the writer buffered every one of them.
//!
//! Here the instrument is the opposite one, so the fixture is too. What is being
//! bounded is the **inflated** size, and a blob of a repeating byte costs
//! kilobytes on disk and its full size in memory. That is precisely the object
//! that gets past a ceiling written against a file size, and it is what makes a
//! multi-gigabyte proof runnable at all: the fixture is cheap to build and
//! expensive to hold, which is the shape of the attack.
//!
//! # The big arms are `#[ignore]`d, and env-driven
//!
//! `PROOF_SIZE_MB` and `PROOF_CEILING_MB` drive
//! [`the_ceiling_binds_not_luck`], which is the arm run under a hard
//! `ulimit -v` outside the ordinary suite. Left alone the whole file runs in the
//! workspace suite at fixture size, because an assertion that only ever runs by
//! hand is an assertion that stops running.
mod common;
use std::io::Cursor;
use gix_hash::{Kind, ObjectId};
use gunnar_store::fs::FsStore;
use gunnar_store::{FullName, ObjectStore, RefEdit, RefStore, Target};
use gunnar_wire::upload_pack::{Config, UploadPack, WantPolicy};
use gunnar_wire::{Error, Limits, ProtocolVersion};
// ── the fixture ────────────────────────────────────────────────────────────
/// A bare on-disk repository holding exactly one commit, whose tree holds one
/// blob of `blob_bytes` bytes.
///
/// On disk rather than in memory, because `MemObjectStore` *is* memory: a
/// fixture that holds the blob in a `HashMap` would put the very bytes under
/// test into the resident set of the process doing the measuring, and no
/// ceiling could then be told from the fixture.
struct BigBlobRepo {
_dir: Option<tempfile::TempDir>,
store: FsStore,
blob: ObjectId,
tip: ObjectId,
}
impl BigBlobRepo {
fn build(blob_bytes: usize) -> Self {
BigBlobRepo::at(None, blob_bytes)
}
/// Build — or **reopen** — the fixture.
///
/// Reopening exists because of a measurement problem, and it is worth
/// naming rather than looking like convenience. Building this fixture costs
/// `blob_bytes` of resident memory, because writing a loose object hands
/// gix the whole thing. So the *build* and the *serve* have the same peak,
/// and any `ulimit` low enough to make the unbounded serve die is also low
/// enough to make the build die — which would prove nothing about serving.
/// Splitting the two into separate processes is what lets the ceiling be
/// the only thing between the fetch and the limit.
fn at(root: Option<&std::path::Path>, blob_bytes: usize) -> Self {
let (dir, git_dir) = match root {
Some(root) => {
std::fs::create_dir_all(root).expect("the proof repository's directory");
(None, root.join("repo.git"))
}
None => {
let dir = tempfile::tempdir().expect("a temporary directory");
let git_dir = dir.path().join("repo.git");
(Some(dir), git_dir)
}
};
let store = FsStore::open_or_init(&git_dir, Kind::Sha1).expect("create the repository");
let objects = store.objects();
let main = FullName::new("refs/heads/main").expect("a valid ref name");
if let Some(tip) = store.refs().resolve(&main).expect("resolve main") {
// Already built by an earlier process. The blob is recovered from
// the commit's tree rather than re-hashed, because re-hashing it
// would need the very allocation this reopen exists to avoid.
let blob = single_tree_entry(&objects, tip);
return BigBlobRepo {
_dir: dir,
store,
blob,
tip,
};
}
// One repeating byte: a few kilobytes deflated, `blob_bytes` inflated.
let content = vec![b'g'; blob_bytes];
let blob = objects
.write(gix_object::Kind::Blob, &content)
.expect("write the blob");
drop(content);
let mut tree = Vec::new();
tree.extend_from_slice(b"100644 big.bin\0");
tree.extend_from_slice(blob.as_slice());
let tree = objects
.write(gix_object::Kind::Tree, &tree)
.expect("write the tree");
let mut commit = Vec::new();
commit.extend_from_slice(format!("tree {tree}\n").as_bytes());
commit.extend_from_slice(b"author g <g@example.invalid> 1700000000 +0000\n");
commit.extend_from_slice(b"committer g <g@example.invalid> 1700000000 +0000\n\n");
commit.extend_from_slice(b"one big blob\n");
let tip = objects
.write(gix_object::Kind::Commit, &commit)
.expect("write the commit");
let refs = store.refs();
refs.transaction(vec![RefEdit::create(
FullName::new("refs/heads/main").expect("a valid ref name"),
Target::Object(tip),
)])
.expect("install refs/heads/main");
refs.set_head(Target::Symbolic(
FullName::new("refs/heads/main").expect("a valid ref name"),
))
.expect("point HEAD at main");
BigBlobRepo {
_dir: dir,
store,
blob,
tip,
}
}
/// Serve one v2 clone of the whole repository under `limits`, and report
/// what came back.
fn clone_under(&self, limits: Limits) -> gunnar_wire::Result<gunnar_wire::UploadSummary> {
let objects = self.store.objects();
let refs = self.store.refs();
let config = Config {
// `Any` keeps this test about the ceiling: `ReachableFromRefs` would
// walk the repository before the pack was ever selected, and a
// refusal from that walk is a different fact.
want_policy: WantPolicy::Any,
limits,
..Config::default()
};
let upload = UploadPack::new(&objects, &refs).with_config(config);
let mut request = Vec::new();
write_line(&mut request, b"command=fetch\n");
write_line(&mut request, b"object-format=sha1\n");
request.extend_from_slice(b"0001");
write_line(&mut request, format!("want {}\n", self.tip).as_bytes());
write_line(&mut request, b"ofs-delta\n");
write_line(&mut request, b"done\n");
request.extend_from_slice(b"0000");
let mut out = Vec::new();
upload.serve(ProtocolVersion::V2, &mut Cursor::new(request), &mut out)
}
}
/// The object id of the one entry in `tip`'s tree.
fn single_tree_entry(objects: &dyn ObjectStore, tip: ObjectId) -> ObjectId {
let mut buf = Vec::new();
let commit = objects
.read(&tip, &mut buf)
.expect("read the tip")
.expect("the tip is here");
let tree = gix_object::CommitRef::from_bytes(commit.data, Kind::Sha1)
.expect("parse the tip")
.tree();
let mut buf = Vec::new();
let tree = objects
.read(&tree, &mut buf)
.expect("read the tree")
.expect("the tree is here");
gix_object::TreeRefIter::from_bytes(tree.data, Kind::Sha1)
.next()
.expect("the tree has an entry")
.expect("the entry parses")
.oid
.to_owned()
}
fn write_line(out: &mut Vec<u8>, payload: &[u8]) {
out.extend_from_slice(format!("{:04x}", payload.len() + 4).as_bytes());
out.extend_from_slice(payload);
}
fn megabytes(mb: usize) -> usize {
mb * 1024 * 1024
}
/// How many bytes of a peer's stream a `ceiling`-byte budget may consume.
///
/// The budget charges what the server **holds** — a pkt-line's payload — and
/// the stream carries four framing bytes per line on top of that. So consuming
/// a little more than the ceiling is correct, and the difference is a property
/// of the peer's line length rather than of the guard. At the shortest line a
/// flood is worth sending (`have <sha1>`, 45 payload bytes) that is 9%; the
/// bound here is generous at 50% plus a whole line, and still an order of
/// magnitude below "read everything, then complain", which is what it exists to
/// catch.
fn framing_bound(ceiling: usize) -> usize {
ceiling + ceiling / 2 + 65_536
}
/// A field out of `/proc/self/status`, in kibibytes.
///
/// `VmHWM` is the kernel's own peak resident set, which is the number a
/// `/usr/bin/time -v` around the whole process reports — and around the whole
/// process it is useless here, because **building** a multi-gigabyte fixture
/// costs multi-gigabytes and would be the peak whatever the fetch did. Writing
/// `5` to `/proc/self/clear_refs` resets `VmHWM`, so the serve can be measured
/// on its own. That is the difference between a number that proves something
/// and a number that reports the size of the test's own setup.
#[cfg(target_os = "linux")]
fn proc_status_kb(field: &str) -> Option<u64> {
let status = std::fs::read_to_string("/proc/self/status").ok()?;
status
.lines()
.find(|line| line.starts_with(field))
.and_then(|line| line.split_whitespace().nth(1))
.and_then(|value| value.parse().ok())
}
#[cfg(target_os = "linux")]
fn reset_peak_rss() {
let _ = std::fs::write("/proc/self/clear_refs", "5");
}
#[cfg(not(target_os = "linux"))]
fn proc_status_kb(_field: &str) -> Option<u64> {
None
}
#[cfg(not(target_os = "linux"))]
fn reset_peak_rss() {}
fn env_mb(name: &str, fallback: usize) -> usize {
std::env::var(name)
.ok()
.and_then(|value| value.parse().ok())
.unwrap_or(fallback)
}
// ── the object ceiling ─────────────────────────────────────────────────────
/// The ceiling is **on by default**, and it is the same number the receive path
/// already applies.
///
/// A value assertion rather than a behavioural one, because the behavioural
/// version of it would need a 1 GiB blob. It is here because the whole point of
/// `gunnar-pack`'s guard — *a guard that is off by default is not a guard* — is
/// a claim about a default, and a default nothing reads is a default that can be
/// changed without anything going red.
#[test]
fn the_object_ceiling_is_on_by_default_and_is_the_receive_paths_number() {
let limits = Config::default().limits;
assert_eq!(
limits.object_bytes,
Some(gunnar_pack::DEFAULT_ALLOC_LIMIT_BYTES),
"a server built from Config::default() would serve with no object ceiling, \
or with one that has drifted from the receive path's"
);
assert_eq!(
limits.request_bytes,
Some(gunnar_wire::DEFAULT_REQUEST_LIMIT_BYTES),
"a server built from Config::default() would read a peer's request with no ceiling"
);
assert_eq!(
Limits::default().no_alloc_limit(),
Limits {
request_bytes: None,
object_bytes: None,
walk_bytes: None
},
"no_alloc_limit() left a ceiling in place, so the opt-out does not opt out"
);
}
/// A blob **under** the ceiling is served, whole, and the ceiling is not what
/// decided it.
///
/// The green half, and it is not decoration. A ceiling that refused everything
/// would satisfy every assertion in the arm below it and would be a far worse
/// bug: a server that cannot clone.
#[test]
fn a_blob_under_the_ceiling_is_served() {
let repo = BigBlobRepo::build(megabytes(4));
let summary = repo
.clone_under(Limits::default().object_bytes(megabytes(8)))
.expect("a 4 MiB blob under an 8 MiB ceiling");
assert_eq!(
summary.objects_sent, 3,
"the commit, its tree and the blob should all be in the pack"
);
}
/// Raise the input past the ceiling and **the ceiling is what binds** — by
/// name, with the size and the limit in the error.
///
/// The two arms differ in one number, the blob's size. Nothing else moves: same
/// fixture shape, same request, same ceiling. So a green here cannot be the
/// fixture being small, the request being odd or the allocator being lucky.
#[test]
fn a_blob_over_the_ceiling_is_refused_by_name() {
let repo = BigBlobRepo::build(megabytes(16));
let ceiling = megabytes(8);
match repo.clone_under(Limits::default().object_bytes(ceiling)) {
Err(Error::ObjectTooLarge { oid, size, limit }) => {
assert_eq!(oid, repo.blob, "the wrong object was named");
assert_eq!(
size,
megabytes(16) as u64,
"the size in the error is not the blob's real inflated size, so it was \
not read from the header"
);
assert_eq!(
limit, ceiling as u64,
"the error names a ceiling nobody set"
);
}
Err(other) => panic!("the object ceiling should have refused this, not {other}"),
Ok(summary) => panic!(
"a 16 MiB blob went into a pack under an 8 MiB ceiling: {} objects sent",
summary.objects_sent
),
}
}
/// …and with the ceiling **off**, the same fetch succeeds.
///
/// This is the RED direction expressed in-process: it proves the refusal above
/// came from the ceiling and not from anything else about a 16 MiB blob. The
/// out-of-process half — the one where taking the ceiling off costs the box its
/// memory — is [`the_ceiling_binds_not_luck`], run under a hard `ulimit -v`.
#[test]
fn without_the_ceiling_the_same_fetch_goes_through() {
let repo = BigBlobRepo::build(megabytes(16));
let summary = repo
.clone_under(Limits::default().no_alloc_limit())
.expect("no ceiling, so nothing to refuse");
assert_eq!(summary.objects_sent, 3);
}
/// The arm run under a hard memory ceiling, at a size the fixture suite cannot
/// afford.
///
/// ```text
/// PROOF_SIZE_MB=2048 PROOF_CEILING_MB=256 \
/// (ulimit -v 3145728; cargo test -p gunnar-wire --test fetch_memory -- --ignored --nocapture)
/// ```
///
/// `PROOF_UNBOUNDED=1` takes the ceiling off, which is the red run: the same
/// blob, the same `ulimit`, and the process dies in the allocator instead of
/// returning an error.
#[test]
#[ignore = "driven by PROOF_SIZE_MB / PROOF_CEILING_MB under an explicit ulimit"]
fn the_ceiling_binds_not_luck() {
let size_mb = env_mb("PROOF_SIZE_MB", 2048);
let ceiling_mb = env_mb("PROOF_CEILING_MB", 256);
let unbounded = std::env::var("PROOF_UNBOUNDED").is_ok();
let root = std::env::var("PROOF_REPO")
.ok()
.map(std::path::PathBuf::from);
println!("building a {size_mb} MiB blob (deflates to kilobytes, inflates to {size_mb} MiB)");
let repo = BigBlobRepo::at(root.as_deref(), megabytes(size_mb));
if std::env::var("PROOF_BUILD_ONLY").is_ok() {
// Phase one. The fixture costs its own size to *write*, so it is built
// by a process with room for it and served by one without.
println!("BUILT ONLY, at {:?}", root);
return;
}
println!(
"built; serving with ceiling = {}",
if unbounded {
"OFF".to_string()
} else {
format!("{ceiling_mb} MiB")
}
);
let limits = if unbounded {
Limits::default().no_alloc_limit()
} else {
Limits::default().object_bytes(megabytes(ceiling_mb))
};
// The fixture is built; forget its cost, so what is reported next is the
// fetch's and nothing else.
reset_peak_rss();
let before = proc_status_kb("VmRSS:").unwrap_or(0);
let outcome = repo.clone_under(limits);
let peak = proc_status_kb("VmHWM:").unwrap_or(0);
println!(
"serve: rss before {before} kB, PEAK RSS during the fetch {peak} kB, \
blob {size_mb} MiB, ceiling {}",
if unbounded {
"OFF".to_string()
} else {
format!("{ceiling_mb} MiB")
}
);
if unbounded {
// Reaching this line at all is the point: the process was expected to
// die in the allocator. Printing what it did instead is what tells a
// reader whether the ulimit was too generous to be a test.
println!("UNBOUNDED run returned: {outcome:?}");
panic!(
"with the ceiling off, serving a {size_mb} MiB blob was expected to exhaust the \
ulimit and abort. It did not, so the ulimit for this run was above what the \
fetch needed and the red half proves nothing — lower it and run again"
);
}
if size_mb <= ceiling_mb {
// The under-the-ceiling data point. Same fixture, same ceiling, one
// number changed — which is what makes the refusal below attributable
// to the size rather than to the size of the fixture.
let summary = outcome.expect("a blob under the ceiling should be served");
assert_eq!(summary.objects_sent, 3);
println!("SERVED {size_mb} MiB under a {ceiling_mb} MiB ceiling: 3 objects");
return;
}
match outcome {
Err(Error::ObjectTooLarge { size, limit, .. }) => {
assert_eq!(size, megabytes(size_mb) as u64);
assert_eq!(limit, megabytes(ceiling_mb) as u64);
println!("REFUSED {size} bytes against a {limit}-byte ceiling, and still running");
}
other => panic!("expected the ceiling to refuse a {size_mb} MiB blob, got {other:?}"),
}
}
// ── the request ceiling ────────────────────────────────────────────────────
/// The ceiling the two flood arms serve under — **and the switch that takes it
/// off**.
///
/// `PROOF_UNBOUNDED=1` is the red half of LAW 2, kept as a knob rather than as a
/// source edit somebody has to redo: with the ceiling off, the endless peer is
/// answered by an endless `Vec`, and under a `ulimit -v` the process dies in the
/// allocator instead of returning an error. Run it as
///
/// ```text
/// PROOF_UNBOUNDED=1 (ulimit -v 1048576; \
/// cargo test -p gunnar-wire --test fetch_memory -- flood --nocapture)
/// ```
///
/// The knob is safe to leave in place because it can only ever turn a test
/// **red**: every assertion below expects a refusal, and removing the ceiling
/// removes the refusal. There is no setting of it that makes a broken guard look
/// green.
fn flood_limits(ceiling: usize) -> Limits {
if std::env::var("PROOF_UNBOUNDED").is_ok() {
Limits::default().no_alloc_limit()
} else {
Limits::default().request_bytes(ceiling)
}
}
/// A peer that sends `have` lines and **never stops**.
///
/// A `Cursor` over a big `Vec` would not do. The bug is that the loop ends when
/// the *peer* sends a flush-pkt, so a finite fixture ends the loop for the
/// server and the assertion becomes "did it read 200 KiB", which a broken
/// server passes as easily as a fixed one. This reader has no end: with the
/// ceiling off the only thing that stops it is the allocator, which is the whole
/// claim.
struct EndlessHaves {
prefix: Vec<u8>,
line: Vec<u8>,
offset: usize,
produced: std::sync::Arc<std::sync::atomic::AtomicUsize>,
}
impl EndlessHaves {
fn new(prefix: Vec<u8>, have: ObjectId) -> Self {
let mut line = Vec::new();
write_line(&mut line, format!("have {have}\n").as_bytes());
EndlessHaves {
prefix,
line,
offset: 0,
produced: std::sync::Arc::new(std::sync::atomic::AtomicUsize::new(0)),
}
}
}
impl std::io::Read for EndlessHaves {
fn read(&mut self, buf: &mut [u8]) -> std::io::Result<usize> {
let source = if self.offset < self.prefix.len() {
&self.prefix[self.offset..]
} else {
let within = (self.offset - self.prefix.len()) % self.line.len();
&self.line[within..]
};
let take = buf.len().min(source.len());
buf[..take].copy_from_slice(&source[..take]);
self.offset += take;
self.produced
.fetch_add(take, std::sync::atomic::Ordering::Relaxed);
Ok(take)
}
}
/// A v0 client that never sends `done` is refused **at the ceiling**, not at
/// the allocator.
///
/// The applied output is *how many of the peer's bytes the server took before
/// it said no*. The peer is endless, so without the ceiling this number has no
/// upper bound at all; with it, the bound is the ceiling plus the framing the
/// ceiling does not count.
#[test]
fn a_v0_have_flood_is_refused_at_the_ceiling() {
let fixture = common::Fixture::build(Kind::Sha1);
let ceiling = 64 * 1024;
let mut prefix = Vec::new();
write_line(
&mut prefix,
format!("want {} side-band-64k ofs-delta\n", fixture.four).as_bytes(),
);
prefix.extend_from_slice(b"0000");
let peer = EndlessHaves::new(prefix, fixture.one);
let produced = std::sync::Arc::clone(&peer.produced);
let config = Config {
limits: flood_limits(ceiling),
..Config::default()
};
let upload = UploadPack::new(&fixture.objects, &fixture.refs).with_config(config);
let mut input = std::io::BufReader::with_capacity(4096, peer);
let err = upload
.serve(ProtocolVersion::V0, &mut input, &mut Vec::new())
.expect_err("an endless `have` flood should be refused");
match err {
Error::RequestTooLarge { limit, .. } => assert_eq!(limit, ceiling),
other => panic!("expected the request ceiling to refuse this, not {other}"),
}
let taken = produced.load(std::sync::atomic::Ordering::Relaxed);
assert!(
taken <= framing_bound(ceiling),
"the server took {taken} bytes from an endless peer before refusing at a \
{ceiling}-byte ceiling"
);
assert!(
taken > ceiling / 2,
"the server refused after only {taken} bytes, well under the {ceiling}-byte \
ceiling — something other than the ceiling ended this, and the bound above is \
not measuring what it claims"
);
}
/// The same, for protocol v2's argument section.
///
/// v2 is the path every modern client takes, so a ceiling that only covered v0
/// would cover the path nobody uses.
#[test]
fn a_v2_argument_flood_is_refused_at_the_ceiling() {
let fixture = common::Fixture::build(Kind::Sha1);
let ceiling = 64 * 1024;
let mut prefix = Vec::new();
write_line(&mut prefix, b"command=fetch\n");
write_line(&mut prefix, b"object-format=sha1\n");
prefix.extend_from_slice(b"0001");
write_line(&mut prefix, format!("want {}\n", fixture.four).as_bytes());
let peer = EndlessHaves::new(prefix, fixture.one);
let produced = std::sync::Arc::clone(&peer.produced);
let config = Config {
limits: flood_limits(ceiling),
..Config::default()
};
let upload = UploadPack::new(&fixture.objects, &fixture.refs).with_config(config);
let mut input = std::io::BufReader::with_capacity(4096, peer);
let err = upload
.serve(ProtocolVersion::V2, &mut input, &mut Vec::new())
.expect_err("an argument section that never ends should be refused");
match err {
Error::RequestTooLarge { limit, .. } => assert_eq!(limit, ceiling),
other => panic!("expected the request ceiling to refuse this, not {other}"),
}
let taken = produced.load(std::sync::atomic::Ordering::Relaxed);
assert!(
taken <= framing_bound(ceiling),
"the server took {taken} bytes from an endless peer before refusing at a \
{ceiling}-byte ceiling"
);
assert!(taken > ceiling / 2, "only {taken} bytes were taken");
}
/// An ordinary negotiation is **not** refused, and the default `want` policy
/// still resolves a non-tip `want`.
///
/// The floor under both flood arms. A ceiling of eight megabytes that happened
/// to refuse a normal fetch would be caught here and nowhere else, and the two
/// flood tests would stay green while the server served nobody.
///
/// It also drives `graph::reachable_closure`, which is the other half of this
/// change: the `want` is an **interior** commit rather than a reference tip, so
/// `WantPolicy::ReachableFromRefs` has to build the repository's whole object
/// closure to answer. That is the call that used to hold the closure twice.
#[test]
fn an_ordinary_negotiation_fits_inside_the_default_ceiling() {
let fixture = common::Fixture::build(Kind::Sha1);
let tip = fixture.extend(fixture.four, 400, "history");
fixture
.refs
.transaction(vec![RefEdit::update(
FullName::new("refs/heads/main").expect("a valid ref name"),
Target::Object(fixture.four),
Target::Object(tip),
)])
.expect("move main onto the extended history");
let haves = fixture.first_parent_haves(tip, 300);
// Not `tip`: an id that is reachable from a reference but is not itself one.
let want = haves[8];
let mut stream = Vec::new();
write_line(&mut stream, b"command=fetch\n");
write_line(&mut stream, b"object-format=sha1\n");
stream.extend_from_slice(b"0001");
write_line(&mut stream, format!("want {want}\n").as_bytes());
for have in &haves {
write_line(&mut stream, format!("have {have}\n").as_bytes());
}
write_line(&mut stream, b"done\n");
stream.extend_from_slice(b"0000");
stream.extend_from_slice(b"0000");
let upload = UploadPack::new(&fixture.objects, &fixture.refs);
let summary = upload
.serve(
ProtocolVersion::V2,
&mut Cursor::new(stream),
&mut Vec::new(),
)
.expect("300 haves for a reachable want is an ordinary negotiation, not a flood");
assert_eq!(
summary.fetch_rounds, 1,
"the round that carries the pack should have been served"
);
assert_eq!(
summary.common,
haves.len(),
"every `have` is in this repository, so every one of them should be common — a \
smaller number means the negotiation stopped reading them"
);
}