gunnar-wire 1.7.0

//! **Eight pushes at once are eight doors, and they were all measuring the
//! same "before".**
//!
//! `tests/pack_ceiling.rs` next door proves the wall works: a push that crosses
//! its byte ceiling is cut mid-transfer, one byte past the number, with the
//! plan's own sentence. Every test in that file is SINGLE THREADED, and a
//! single-threaded test proves nothing about a limit that several connections
//! share.
//!
//! The door measures, then admits. `gunnar_server::entitlement`'s
//! `pack_ceiling` reads `caps["pack_bytes"]`, subtracts what the namespace
//! already holds and adds slack; `Dispatch::ask_plan` asks it once per push.
//! Two pushes that arrive together therefore get the same number, and each is
//! held to the whole of it. A tenant with 1 GiB left who opens ten simultaneous
//! pushes is admitted for ten times their remaining room — a TOCTOU on a
//! resource limit, bounded by N × the ceiling and narrowed, not closed, by the
//! fact that the measurement counts what has LANDED: every admission happens
//! before any of the bytes do.
//!
//! # Red first — MEASURED 2026-09-08
//!
//! Run against the door as it stood, with `PackCeiling { bytes, refusal }` and
//! no ledger behind it. Eight clients, room for one:
//!
//! ```text
//! ---- simultaneous_pushes_are_not_each_admitted_against_the_same_measurement stdout ----
//! thread 'simultaneous_pushes_are_not_each_admitted_against_the_same_measurement'
//! panicked at crates/gunnar-wire/tests/concurrent_pack_ceiling.rs:224:5:
//! 8 simultaneous pushes put 328128 pack bytes into a namespace with room for 65536:
//! 8 of them were admitted, each against the same measurement of the same store
//! ```
//!
//! Five times the room — eight packs of 41016 bytes into 65536 — on a fixture
//! sized in kibibytes so the arithmetic is the subject. In gibibytes it is the operator's disk.
//!
//! # What this file can and cannot see
//!
//! `gunnar-wire` has no filesystem and no idea what a plan is, so the
//! measurement in front of the door is a constant here rather than a walk: the
//! ceiling every client is handed is the one number they would all have got
//! from measuring the same namespace at the same instant, which is exactly what
//! production does. Each client pushes into its own in-memory store, because
//! what is under test is the DOOR and not two pushes racing for one reference
//! name — a ref collision would refuse a push for a second reason and make a
//! refusal prove nothing.
//!
//! What ties them together is therefore the only thing that ties them together
//! in production either: the namespace's ledger of bytes admitted since each of
//! them looked (`gunnar_lfs::ledger`, reached here through
//! [`gunnar_wire::PackLedger`], which is the seam that lets this crate ask the
//! question without linking the crate that answers it).
//!
//! # Why real threads
//!
//! Because the bug is in the interleaving and nothing else. Eight pushes ONE
//! AFTER ANOTHER pass on the broken code — that is
//! `a_push_of_exactly_the_ceiling_lands` next door, green throughout. The
//! barrier is what makes the failure deterministic rather than a race the
//! machine wins nine times in ten: every client MEASURES before any client
//! SENDS, 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 LAW's parallelism rule):
//! 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.

mod common;

use std::sync::{Arc, Barrier, Mutex};

use gix_hash::Kind;
use gunnar_wire::receive_pack::{AllowAll, Config};
use gunnar_wire::{PackCeiling, PackLedger, ReceivePack};

use common::{incompressible_pack, push_bytes, validating_ingest, Fixture, Tracked};

/// The namespace's whole remaining room, slack included — what
/// `EntitlementSlot::pack_ceiling` hands one push and, before this file, handed
/// every push at once.
const CEILING: u64 = 64 * 1024;

/// One client's push, on the wire. Two of these do not fit under [`CEILING`],
/// which is the whole point: room for one, and eight asking.
const PACK_BYTES: usize = 40 * 1024;

/// …and eight clients arriving for it together.
const CLIENTS: usize = 8;

const REFUSAL: &str = "plan 10gb would pass its pack quota with this push: the transfer was \
                       stopped and nothing was stored; delete or buy more at \
                       https://gunnar.rs/billing/team";

/// **The namespace's ledger, spelled here rather than linked.**
///
/// In production this is `gunnar_lfs::Mark::admit_upto` behind
/// `gunnar_server::transport`'s adapter — a monotone `u64` per namespace and a
/// per-holder mark. `gunnar-wire` cannot name it (it does not depend on
/// `gunnar-lfs`, and the trait exists precisely so it does not have to), so the
/// test implements the same contract in ten lines: charge what fits inside
/// `room`, counting everything admitted since this holder marked, and return
/// what was charged.
///
/// The arithmetic being re-spelled is not the arithmetic under test — that is
/// `gunnar-lfs`'s own `tests/concurrent_allowance.rs`. What is under test here
/// is that the RECEIVE LOOP asks before it takes, and asks once per chunk.
#[derive(Debug)]
struct TestLedger {
    admitted: Mutex<u64>,
}

/// One holder's standing: the ledger, and the value it read when it measured.
#[derive(Debug)]
struct Holder {
    ledger: Arc<TestLedger>,
    at: u64,
}

impl PackLedger for Holder {
    fn admit_upto(&self, want: u64, room: u64) -> u64 {
        let mut admitted = self.ledger.admitted.lock().unwrap();
        let since = admitted.saturating_sub(self.at);
        let grant = want.min(room.saturating_sub(since));
        *admitted += grant;
        grant
    }
}

/// The door, built the way `Dispatch::ask_plan` builds it: **mark FIRST, then
/// measure**. The order is the contract (`gunnar_lfs::ledger`) — an admission
/// that lands during the measurement is then charged, where marking afterwards
/// would let it through twice — and a test that got it backwards would be
/// testing a door production does not have.
fn door(ledger: &Arc<TestLedger>) -> PackCeiling {
    let at = *ledger.admitted.lock().unwrap();
    PackCeiling {
        // The measurement: what the namespace has room for, right now, for
        // everyone who asks at this instant.
        bytes: CEILING,
        refusal: REFUSAL.to_owned(),
        ledger: Some(Arc::new(Holder {
            ledger: Arc::clone(ledger),
            at,
        })),
    }
}

/// What one client's push cost the namespace.
struct Pushed {
    /// Packfile bytes the server took off this client's wire.
    pack_taken: u64,
    /// Whether the push landed, as the client reads it: `unpack ok`.
    admitted: bool,
}

/// One client's whole conversation: build a repository, offer a pack, and
/// report what the server actually took.
fn push(pack: &[u8], ceiling: PackCeiling) -> Pushed {
    let fixture = Fixture::build(Kind::Sha1);
    let zero = Kind::Sha1.null().to_string();
    // `refs/heads/fresh` at an object the fixture ALREADY holds: the pack's
    // contents are not the subject, and a connectivity failure would be a
    // second reason for the push to be refused.
    let input = push_bytes(
        &[(&zero, &fixture.merge.to_string(), "refs/heads/fresh")],
        "report-status object-format=sha1",
        pack,
    );
    let commands = (input.len() - pack.len()) as u64;
    let mut reader = Tracked::new(input);
    let rp = ReceivePack::new(&fixture.git, fixture.kind).with_config(Config {
        pack_ceiling: Some(ceiling),
        ..Config::default()
    });
    let mut out = Vec::new();
    let outcome = rp
        .serve(
            &mut reader,
            &mut out,
            &validating_ingest(Kind::Sha1),
            &AllowAll,
        )
        .expect("serve")
        .expect("a push happened");
    Pushed {
        pack_taken: reader.taken - commands,
        admitted: outcome.unpack.is_ok(),
    }
}

// ── the defect: eight pushes, room for one ─────────────────────────────────

/// **The row.** Eight clients measure the same namespace at the same instant
/// and then all send. The bytes they are collectively allowed to spend are the
/// namespace's, not eight times the namespace's.
#[test]
fn simultaneous_pushes_are_not_each_admitted_against_the_same_measurement() {
    let ledger = Arc::new(TestLedger {
        admitted: Mutex::new(0),
    });
    let pack = incompressible_pack(Kind::Sha1, 1, PACK_BYTES);
    assert!(
        2 * pack.len() as u64 > CEILING,
        "two of these packs must not fit under the ceiling or this test proves nothing: \
         {} bytes each, ceiling {CEILING}",
        pack.len()
    );

    let measured = Barrier::new(CLIENTS);
    let pushes: Vec<Pushed> = std::thread::scope(|scope| {
        let handles: Vec<_> = (0..CLIENTS)
            .map(|_| {
                let ledger = &ledger;
                let measured = &measured;
                let pack = &pack;
                scope.spawn(move || {
                    // ── T, the check ────────────────────────────────────────
                    let ceiling = door(ledger);
                    // Every client has now measured, and not one pack byte has
                    // moved. This is the interleaving, made deterministic.
                    measured.wait();
                    // ── O, the use ──────────────────────────────────────────
                    push(pack, ceiling)
                })
            })
            .collect();
        handles.into_iter().map(|h| h.join().unwrap()).collect()
    });

    let landed: u64 = pushes.iter().map(|pushed| pushed.pack_taken).sum();
    let admitted = pushes.iter().filter(|pushed| pushed.admitted).count();

    assert!(
        landed <= CEILING + 1,
        "{CLIENTS} simultaneous pushes put {landed} pack bytes into a namespace with room for \
         {CEILING}: {admitted} of them were admitted, each against the same measurement of the \
         same store"
    );
    assert!(
        admitted <= 1,
        "room for one push and {admitted} were let in"
    );
}

/// **The other half: the ledger must not refuse a push that fits.**
///
/// A door that refused everything would satisfy the row above and nothing else.
/// Eight clients, and room for every one of them: all eight land, whole, and
/// the namespace is charged exactly what they sent.
#[test]
fn simultaneous_pushes_that_all_fit_are_all_admitted() {
    let ledger = Arc::new(TestLedger {
        admitted: Mutex::new(0),
    });
    let pack = incompressible_pack(Kind::Sha1, 1, PACK_BYTES);
    // The same eight clients, under a ceiling with room for all of them.
    let roomy = (CLIENTS as u64 + 1) * pack.len() as u64;

    let measured = Barrier::new(CLIENTS);
    let pushes: Vec<Pushed> = std::thread::scope(|scope| {
        let handles: Vec<_> = (0..CLIENTS)
            .map(|_| {
                let ledger = &ledger;
                let measured = &measured;
                let pack = &pack;
                scope.spawn(move || {
                    let at = *ledger.admitted.lock().unwrap();
                    let ceiling = PackCeiling {
                        bytes: roomy,
                        refusal: REFUSAL.to_owned(),
                        ledger: Some(Arc::new(Holder {
                            ledger: Arc::clone(ledger),
                            at,
                        })),
                    };
                    measured.wait();
                    push(pack, ceiling)
                })
            })
            .collect();
        handles.into_iter().map(|h| h.join().unwrap()).collect()
    });

    assert_eq!(
        pushes.iter().filter(|pushed| pushed.admitted).count(),
        CLIENTS,
        "every one of these pushes fits under the ceiling; the ledger must not refuse a push \
         because a sibling exists"
    );
    for pushed in &pushes {
        assert_eq!(
            pushed.pack_taken,
            pack.len() as u64,
            "a push that fits must be read WHOLE: a ledger that hands out short of the pack is \
             a corrupt push, not a refused one"
        );
    }
}

/// **The failure path, run.** A push that dies between admit and transfer must
/// not cost the tenant room they still own.
///
/// This is the class of bug the reservation-with-`Drop` design would have had,
/// and the reason this one has nothing to give back: a release that does not
/// run leaks bytes charged to a tenant forever and locks them out of quota they
/// paid for. Nothing is held here, so nothing can leak — a holder's charge ages
/// out because the NEXT holder marks at the current value.
///
/// The panic is real: it unwinds out of the middle of a push, past the ceiling,
/// and is caught only so the test can carry on and ask the next question.
#[test]
fn a_panic_mid_push_does_not_cost_the_next_push() {
    let ledger = Arc::new(TestLedger {
        admitted: Mutex::new(0),
    });
    let pack = incompressible_pack(Kind::Sha1, 1, PACK_BYTES);

    let died = std::panic::catch_unwind(std::panic::AssertUnwindSafe(|| {
        let ceiling = door(&ledger);
        // Charged for the whole namespace, and then the connection drops.
        let ledger = ceiling.ledger.as_ref().expect("a standing");
        assert_eq!(ledger.admit_upto(CEILING, CEILING), CEILING);
        panic!("the connection dropped");
    }));
    assert!(died.is_err(), "the test's own panic must have happened");

    let next = push(&pack, door(&ledger));
    assert!(
        next.admitted,
        "a push that died before it stored anything has locked the tenant out of room they \
         still own"
    );
    assert_eq!(
        next.pack_taken,
        pack.len() as u64,
        "the next push must be read whole, not held to what a dead one was charged"
    );
}

/// **An aborted transfer does not charge the namespace forever.**
///
/// The push that is cut costs the ledger the bytes it actually took — which is
/// the honest, over-conservative direction: those bytes really did arrive. What
/// it must NOT do is charge the ceiling it was granted, or the pack it meant to
/// send. A fresh holder, marking after it, gets the whole namespace back.
#[test]
fn a_cut_push_charges_what_it_took_and_not_what_it_meant_to_send() {
    let ledger = Arc::new(TestLedger {
        admitted: Mutex::new(0),
    });
    // A pack that dwarfs the ceiling: this push is cut mid-transfer.
    let huge = incompressible_pack(Kind::Sha1, 8, 128 * 1024);
    let cut = push(&huge, door(&ledger));
    assert!(!cut.admitted, "a push far over the ceiling must be cut");
    assert_eq!(
        cut.pack_taken,
        CEILING + 1,
        "the cut push must cost the server the ceiling and the one byte that trips it, no more"
    );
    assert_eq!(
        *ledger.admitted.lock().unwrap(),
        CEILING + 1,
        "the namespace is charged what arrived, not the ceiling it was granted and not the \
         pack the client meant to send"
    );

    // The next holder marks where the counter now stands, so the cut push's
    // charge is invisible to it and what it is held to is its own measurement.
    let small = incompressible_pack(Kind::Sha1, 1, PACK_BYTES);
    let next = push(&small, door(&ledger));
    assert!(
        next.admitted,
        "an aborted transfer has charged the tenant for ever; the next push is refused room \
         nobody is using"
    );
}