crates/gunnar-lfs/tests/concurrent_allowance.rs
//! **Two uploads at once are two doors, and they were both measuring the same
//! "before".**
//!
//! `tests/namespace_allowance.rs` next door proves the LFS wall works: a tenant
//! with 1 GiB left is refused a 10 GiB object, and an upload split into
//! individually-legal pieces is refused at the piece that crosses. Every one of
//! those tests is SINGLE THREADED, and a single-threaded test proves nothing
//! about a limit that several connections share.
//!
//! The door measures, then admits. Under concurrency every request measures the
//! same state:
//!
//! * HTTP LFS passes `taken = 0` (`gunnar-http/src/lfs.rs`, *"one request being
//! one measurement"*), so every request is judged against the allowance
//! measured when its own `open` ran.
//! * Two simultaneous SSH sessions each carry their own `Session::taken`, which
//! counts that session's objects and knows nothing of the other's.
//!
//! A tenant with 1 GiB left who opens ten simultaneous 1 GiB uploads is admitted
//! for all ten. It is a TOCTOU on a resource limit, narrowed but not closed by
//! the walk counting in-flight bytes under `lfs/incomplete/` — narrowed, because
//! the bytes have to have LANDED to be counted, and every admission happens
//! before any of them do.
//!
//! # Red first — MEASURED 2026-09-07
//!
//! Run against the door as it stood, with `Allowance` alone and no ledger. Eight
//! clients, room for four:
//!
//! ```text
//! ---- simultaneous_uploads_are_not_each_admitted_against_the_same_stale_measurement stdout ----
//! thread '…' panicked at crates/gunnar-lfs/tests/concurrent_allowance.rs:185:5:
//! 8 simultaneous uploads put 32768 bytes into a namespace capped at 16384:
//! 8 of them were admitted, each against the same measurement of an empty store
//! ```
//!
//! Twice the plan, on a fixture sized in kibibytes so the arithmetic is the
//! subject. In gibibytes it is the operator's disk.
//!
//! # Why this file spawns real threads
//!
//! Because the bug is in the interleaving and nothing else. A test that called
//! `admit` eight times in a row would pass on the broken code — that is exactly
//! `the_allowance_is_the_sessions_running_total_and_not_a_per_object_limit`,
//! which has been green throughout. The two barriers are what make the failure
//! deterministic rather than a race the machine wins nine times in ten: every
//! client MEASURES before any client WRITES, which is the shape the bug has in
//! production and the shape a passing run must survive.
//!
//! `std::thread` and not the gatling engine (the multicore rule: gatling,
//! never rayon): gatling saturates cores over splittable work, and this is not
//! work — it is eight connections arriving at one door at the same instant.
//! The thing under test IS the interleaving.
//!
//! # The measurement is reproduced here, and that is deliberate
//!
//! `held` below is `gunnar_server::entitlement::EntitlementSlot::lfs_allowance_state`'s
//! arithmetic — `caps["lfs_bytes"]` minus a walk of everything under the
//! namespace's `lfs/` dirs — spelled again in ten lines. `gunnar-lfs` cannot
//! depend on `gunnar-server` (the whole reason `Allowance` is *"a number
//! travelling to where the bytes are"* rather than a policy in this crate), so
//! the alternative was to test the door with no measurement in front of it, and
//! a door tested without its measurement cannot show a TOCTOU at all.
#![cfg(feature = "fs")]
use std::io::BufRead;
use std::path::Path;
use std::sync::{Arc, Barrier};
use gunnar_lfs::{
Allowance, AllowanceLedger, FsLfsStore, Limits, NamespaceAllowance, Oid, Operation, Session,
};
/// The tenant. One namespace throughout: the whole subject is what several
/// transfers against the SAME namespace do to each other.
const NAMESPACE: &str = "team";
/// One object, in bytes. Small on purpose: the arithmetic is the subject and a
/// fixture sized in gibibytes would be a test of the machine's disk.
const OBJECT: u64 = 4096;
/// The plan: room for exactly four of them.
const CAP: u64 = 4 * OBJECT;
/// …and eight clients arriving at once for it.
const CLIENTS: usize = 8;
const REFUSAL: &str = "plan 4k has room left for four objects; this upload does not fit";
/// Everything under `dir`, recursively — the walk `Usage.Read` bills from and
/// the one `lfs_allowance_state` subtracts from the cap. Counts `objects/` and
/// `incomplete/` alike, because both occupy the tenant's disk
/// (`usage.proto`, `lfs_bytes`).
fn dir_bytes(dir: &Path) -> u64 {
let Ok(entries) = std::fs::read_dir(dir) else {
return 0;
};
entries
.flatten()
.map(|entry| match entry.file_type() {
Ok(kind) if kind.is_dir() => dir_bytes(&entry.path()),
Ok(_) => entry.metadata().map(|meta| meta.len()).unwrap_or(0),
Err(_) => 0,
})
.sum()
}
/// Only the bytes that actually LANDED — `lfs/objects/`, not `lfs/incomplete/`.
/// What the tenant now holds and is billed for, and the number the cap is a cap
/// on.
fn committed_bytes(store: &FsLfsStore) -> u64 {
dir_bytes(&store.root().join("objects"))
}
/// A distinct object per client, so no two uploads dedupe into one file and
/// make the total look honest by accident.
fn payload(client: usize) -> Vec<u8> {
let mut bytes = vec![0u8; OBJECT as usize];
bytes[0] = client as u8;
bytes
}
fn real_oid(payload: &[u8]) -> Oid {
let mut hasher = gix_hash::hasher(gix_hash::Kind::Sha256);
hasher.update(payload);
match hasher.try_finalize().unwrap() {
gix_hash::ObjectId::Sha256(digest) => Oid::from_digest(digest),
other => unreachable!("{other:?}"),
}
}
fn script(oid: &Oid, size: u64, payload: &[u8]) -> Vec<u8> {
let mut s = Vec::new();
gunnar_pktline::write_text(&mut s, b"version 1").unwrap();
gunnar_pktline::write_flush(&mut s).unwrap();
gunnar_pktline::write_text(&mut s, format!("put-object {oid}").as_bytes()).unwrap();
gunnar_pktline::write_text(&mut s, format!("size={size}").as_bytes()).unwrap();
gunnar_pktline::write_delim(&mut s).unwrap();
gunnar_pktline::write_data(&mut s, payload).unwrap();
gunnar_pktline::write_flush(&mut s).unwrap();
gunnar_pktline::write_text(&mut s, b"quit").unwrap();
gunnar_pktline::write_flush(&mut s).unwrap();
s
}
fn statuses(wire: &[u8]) -> Vec<u16> {
String::from_utf8_lossy(wire)
.split("status ")
.skip(1)
.filter_map(|rest| rest.split('\n').next()?.trim().parse().ok())
.collect()
}
/// **The door, built the way `Dispatch::ask_plan` builds it**: mark the ledger
/// FIRST, then walk the disk, then bind the two. The order is the contract
/// (`gunnar_lfs::ledger`), and a test that got it backwards would be testing a
/// door production does not have.
fn door(ledger: &Arc<AllowanceLedger>, store: &FsLfsStore) -> NamespaceAllowance {
let mark = ledger.mark(NAMESPACE);
let held = dir_bytes(store.root());
NamespaceAllowance::new(
mark,
Allowance {
bytes: CAP.saturating_sub(held),
refusal: REFUSAL.to_owned(),
},
)
}
/// One client's whole conversation: measure, build the allowance, upload.
fn upload(store: &FsLfsStore, allowance: NamespaceAllowance, client: usize) -> Vec<u16> {
let bytes = payload(client);
let oid = real_oid(&bytes);
let mut input = std::io::BufReader::new(std::io::Cursor::new(script(&oid, OBJECT, &bytes)));
let mut output = Vec::new();
let reader: &mut dyn BufRead = &mut input;
Session::new(store, Operation::Upload)
.with_limits(Limits::default())
.with_allowance(allowance)
.serve(reader, &mut output)
.expect("a well-framed conversation must not fail the session");
statuses(&output)
}
/// **The defect.** Eight clients, room for four, and nothing between them.
#[test]
fn simultaneous_uploads_are_not_each_admitted_against_the_same_stale_measurement() {
let dir = tempfile::tempdir().unwrap();
let store = FsLfsStore::open_or_init(dir.path()).unwrap();
let ledger = AllowanceLedger::detached();
let measured = Barrier::new(CLIENTS);
let admitted: Vec<Vec<u16>> = std::thread::scope(|scope| {
let handles: Vec<_> = (0..CLIENTS)
.map(|client| {
let store = &store;
let measured = &measured;
let ledger = &ledger;
scope.spawn(move || {
// ── T, the check ────────────────────────────────────────
let allowance = door(ledger, store);
// Every client has now measured, and not one byte has been
// written. This is the interleaving, made deterministic.
measured.wait();
// ── O, the use ──────────────────────────────────────────
upload(store, allowance, client)
})
})
.collect();
handles.into_iter().map(|h| h.join().unwrap()).collect()
});
let landed = committed_bytes(&store);
let admitted_count = admitted
.iter()
.filter(|statuses| statuses.contains(&200) && !statuses.contains(&507))
.count();
assert!(
landed <= CAP,
"{CLIENTS} simultaneous uploads put {landed} bytes into a namespace \
capped at {CAP}: {admitted_count} of them were admitted, each against \
the same measurement of an empty store"
);
assert_eq!(
admitted_count,
(CAP / OBJECT) as usize,
"room for {} objects and {admitted_count} were let in",
CAP / OBJECT
);
}
/// **The failure path, run.** A transfer that dies between admit and commit must
/// not cost the tenant room they still own.
///
/// This is the class of bug the reservation design would have had and the reason
/// this one has no `Drop`: a release that does not run leaks bytes charged to a
/// tenant forever, and a crash locks them out of quota they paid for. There is
/// nothing to release here, so the property is structural — but "structural" is
/// what everybody says right before the incident, so it is measured.
///
/// The panic is real: it unwinds out of the middle of a session, past the
/// allowance, and is caught only so the test can carry on and ask the next
/// question.
#[test]
fn a_panic_between_admit_and_commit_does_not_cost_the_next_upload() {
let dir = tempfile::tempdir().unwrap();
let store = FsLfsStore::open_or_init(dir.path()).unwrap();
let ledger = AllowanceLedger::detached();
// A transfer that is admitted for the WHOLE plan and then dies.
let died = std::panic::catch_unwind(std::panic::AssertUnwindSafe(|| {
let allowance = door(&ledger, &store);
allowance.admit(CAP).expect("the whole plan is available");
panic!("the connection dropped");
}));
assert!(died.is_err(), "the test's own panic must have happened");
// A fresh transfer marks at the counter as it now stands, so everything
// charged before it is invisible to it, and what it is held to is what its
// own walk of the disk found — nothing.
let next = door(&ledger, &store);
next.admit(CAP).expect(
"a panicked upload that committed nothing has locked the tenant out of the plan they \
still own",
);
}
/// The same property one step further out: a whole SESSION whose object fails
/// verification. Nothing lands, and the next transfer gets the full plan.
///
/// `VerifyingSink` refuses an object whose bytes do not hash to its oid, which
/// is a failure AFTER the door admitted it — the exact window a reservation
/// would have had to unwind.
///
/// # The session now ENDS, and that is the other half of the same finding
///
/// It used to survive its own 422 and answer the next command. Since
/// 2026-09-11 (GUN-017) it does not: the door admitted `declared` bytes and the
/// client delivered none of them, and a 422 the connection survives lets it
/// declare again immediately — charging a namespace's whole plan, as fast as it
/// can loop, without a byte landing. See `Session::charged_but_not_delivered`.
///
/// So `serve` returning `Err` here is the fix working, not a regression, and
/// this test asserts it rather than working around it. What the test is ABOUT
/// is unchanged and is the two assertions below: nothing landed, and the charge
/// does not follow the tenant into the next transfer.
#[test]
fn an_upload_that_fails_verification_does_not_cost_the_next_upload() {
let dir = tempfile::tempdir().unwrap();
let store = FsLfsStore::open_or_init(dir.path()).unwrap();
let ledger = AllowanceLedger::detached();
let bytes = payload(0);
// A real oid, of a DIFFERENT object: the declaration passes the door and the
// payload fails the hash.
let lying_oid = real_oid(&payload(1));
let mut input =
std::io::BufReader::new(std::io::Cursor::new(script(&lying_oid, OBJECT, &bytes)));
let mut output = Vec::new();
let reader: &mut dyn BufRead = &mut input;
let err = Session::new(&store, Operation::Upload)
.with_limits(Limits::default())
.with_allowance(door(&ledger, &store))
.serve(reader, &mut output)
.expect_err(
"a session that charged the allowance and delivered nothing must end; see \
GUN-017",
);
assert!(
format!("{err}").contains("does not get to declare another"),
"the session ended, but not over the unpaid-for charge: {err}"
);
assert_eq!(
committed_bytes(&store),
0,
"an object that does not hash to its oid must not land"
);
let next = door(&ledger, &store);
next.admit(CAP)
.expect("an upload that landed nothing has taken the tenant's room with it");
}
/// **The single-session property `taken` used to carry, kept.** Removing the
/// per-session running total from the door would have been a silent regression
/// of `namespace_allowance.rs`'s central assertion, so it is asserted again
/// here, through the ledger: one holder, four objects that fit, and the fifth
/// refused.
#[test]
fn one_transfers_own_earlier_objects_are_still_counted_against_it() {
let dir = tempfile::tempdir().unwrap();
let store = FsLfsStore::open_or_init(dir.path()).unwrap();
let ledger = AllowanceLedger::detached();
let allowance = door(&ledger, &store);
for object in 0..(CAP / OBJECT) {
allowance
.admit(OBJECT)
.unwrap_or_else(|err| panic!("object {object} fits in the plan: {err}"));
}
let err = allowance
.admit(OBJECT)
.expect_err("the object that crosses the plan must be refused");
assert!(
err.to_string().contains(REFUSAL),
"the client is owed the plan's own sentence: {err}"
);
}
/// **Two transfers, one namespace, no threads** — the arithmetic of the fix
/// written out, so a failure in the concurrent test above can be read as either
/// "the sums are wrong" or "the interleaving is wrong" rather than guessed at.
#[test]
fn a_second_transfer_sees_what_the_first_was_admitted_for() {
let dir = tempfile::tempdir().unwrap();
let store = FsLfsStore::open_or_init(dir.path()).unwrap();
let ledger = AllowanceLedger::detached();
// Both measure the same empty store, exactly as two connections that
// arrived together would.
let first = door(&ledger, &store);
let second = door(&ledger, &store);
first.admit(3 * OBJECT).expect("three of the four fit");
second
.admit(OBJECT)
.expect("the fourth fits: 3 + 1 is the whole plan and not a byte more");
second
.admit(OBJECT)
.expect_err("the fifth object is past the plan, whichever transfer asks for it");
}
/// A ledger keyed by namespace, and one tenant's uploads must not spend
/// another's plan. The cheapest way to get the whole design wrong is a single
/// global counter, and it would pass every other test in this file.
#[test]
fn one_namespaces_uploads_do_not_spend_anothers_plan() {
let dir = tempfile::tempdir().unwrap();
let store = FsLfsStore::open_or_init(dir.path()).unwrap();
let ledger = AllowanceLedger::detached();
let ours = NamespaceAllowance::new(
ledger.mark(NAMESPACE),
Allowance {
bytes: CAP,
refusal: REFUSAL.to_owned(),
},
);
let theirs = NamespaceAllowance::new(
ledger.mark("someone-else"),
Allowance {
bytes: CAP,
refusal: REFUSAL.to_owned(),
},
);
ours.admit(CAP).expect("our whole plan");
theirs
.admit(CAP)
.expect("their plan is theirs; we did not spend it");
let _ = store;
}