gunnar-lfs 1.6.0

//! **One namespace's room, shared by every upload in flight against it.**
//!
//! [`Allowance`] is a number measured off the disk and handed to one transfer.
//! That is correct for one transfer and wrong for two, because both measure the
//! same "before":
//!
//! * The HTTP object endpoint passes `taken = 0` — *"one request being one
//!   measurement"* — so every `PUT` is judged against the allowance measured
//!   when its own `open` ran.
//! * Two simultaneous SSH sessions each carry their own [`crate::Session`]'s
//!   `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 was
//! admitted for all ten. It is a time-of-check/time-of-use on a resource limit,
//! bounded by N × object size and narrowed — not closed — by the fact that the
//! measuring walk counts in-flight bytes under `lfs/incomplete/`: the bytes have
//! to have LANDED to be counted, and every admission happens before any of them
//! do. `tests/concurrent_allowance.rs` measured it at 2× with eight clients and
//! room for four.
//!
//! # The shape: a monotone counter per namespace, and NOTHING to give back
//!
//! Every admission adds the object's declared size to one `u64` per namespace.
//! The counter only ever goes up. A holder ([`Mark`]) records the value it saw
//! at the moment it was created, and every later admission is judged against the
//! bytes admitted **since that mark**:
//!
//! ```text
//! admitted_since(mark) + this_object  <=  allowance.bytes
//! ```
//!
//! `allowance.bytes` is already `cap - held`, measured off the disk just after
//! the mark was taken, so the sum is the whole truth: what was on disk, plus
//! everything admitted since, plus what is being asked for now.
//!
//! ## Why the mark is taken BEFORE the disk is walked
//!
//! An admission that lands *during* the walk may or may not be visible to it.
//! Marking first puts such an admission after the mark, so it is charged — and
//! if the walk happened to see its bytes as well, it is charged twice, which
//! refuses slightly early. Marking after the walk would put it before the mark
//! AND possibly outside the walk, which admits too much. Of the two roundings
//! only one of them is a quota that leaks, so the order is not a matter of
//! taste. `gunnar_ssh::server::Dispatch::ask_plan` takes the mark before it asks
//! any gate for an allowance, and that ordering is the contract.
//!
//! ## Why there is no reservation to release, and why that is the point
//!
//! The obvious design is a reservation taken at admit and released on
//! completion or abort, guarded by `Drop`. It is also the design with a
//! failure mode nobody wants: a release that does not run — a panic in the
//! wrong place, a connection dropped where the guard was already forgotten, a
//! `process::abort` — leaks bytes that are charged to the tenant **forever**,
//! and the tenant is locked out of quota they own by a crash. Exactly that
//! class of bug was found in this repository the same week ("a purge that
//! fails still releases the lockout").
//!
//! Nothing is held here, so nothing can be leaked. A holder's charge ages out on
//! its own: the NEXT holder marks at the current value of the counter, so
//! everything charged before it is invisible to it, and what is on disk is what
//! its own walk measured. There is no `Drop` impl in this file. A panic between
//! admit and commit costs the panicking transfer's own charge for the life of
//! that one holder and nothing after it, and
//! `tests/concurrent_allowance.rs::a_panic_between_admit_and_commit_does_not_cost_the_next_upload`
//! is that failure path, run.
//!
//! ## What it costs, honestly
//!
//! An aborted upload stays charged against the holders that were alive when it
//! was admitted. On the HTTP wire a holder is one request, so it clears within
//! milliseconds. On a long-lived SSH session it lasts to the end of that
//! session: that session sees less room than truly free, by at most the bytes
//! of the uploads that failed beside it. **Over-conservative, transient, and
//! self-healing** — the safe direction, and it is a refusal a client retries
//! rather than a byte the operator pays for.
//!
//! The alternative that would remove even that — decrementing on abort — buys a
//! narrower window in exchange for the exact leak-a-reservation hazard above,
//! and the arithmetic that makes a holder's snapshot survive a decrement is a
//! second thing to get right. Not worth it. If the SSH session case ever bites,
//! the cheap fix is a fresh mark per `put-object` rather than per session, which
//! is four lines and no new hazard.
//!
//! # Why the PUSH door borrows this file, and why the file did not move
//!
//! Since 2026-09-08 the pushed-packfile ceiling uses the same counter shape:
//! `gunnar_ssh::server::Dispatch::ask_plan` marks [`AllowanceLedger::shared_pack`]
//! before the gate chain walks anything, `gunnar_server::transport` binds that
//! mark to the number the tightest gate asked for, and `gunnar-wire`'s receive
//! loop charges it per chunk through [`Mark::admit_upto`]. Two doors, one
//! shape, one file.
//!
//! The obvious tidy-up is to move this file to `gunnar-core`, where both doors
//! could name it without an eyebrow. **It would cost the appliance a hundred
//! crates.** `gunnar-lfs` is a LEAF — `gunnar-pktline`, `gix-hash` and nothing
//! else — and `gunnar-ssh`'s manifest depends on it saying so out loud;
//! `gunnar-core` carries the identity table, the control database (redb), the
//! crypto backend, the certificate issuer and the legacy Arrow log. MEASURED
//! 2026-09-08 with `cargo tree -e normal`: `gunnar-lfs` resolves 44 packages,
//! `gunnar-core` 141. Moving the ledger up would drag those 141 into
//! `gunnar-lfs-serve`, whose whole argument is that it is small.
//!
//! `gunnar-wire` — the crate that does the charging — reaches it through
//! neither: it names a one-method trait of its own (`gunnar_wire::PackLedger`)
//! and `gunnar-server`, which already holds both halves of the `PackCeiling`
//! twin, implements it in four lines. That is the dependency edge inverted
//! rather than added, and it is why `gunnar-wire` still depends on exactly what
//! it depended on before (51 packages, unchanged).
//!
//! # Why not serialise uploads per namespace instead
//!
//! Because it would cost more than the bug. One upload at a time per namespace
//! turns a tenant's eight parallel LFS uploads into eight sequential ones — the
//! git-lfs client defaults to three concurrent transfers and people raise it —
//! and the overshoot it buys back is at most one object's worth over what this
//! already gives. The lock here is held across a hash lookup and an integer add,
//! never across the disk walk, never across the transfer. Two uploads still run
//! at full speed; only their bookkeeping is ordered.

use std::collections::HashMap;
use std::sync::{Arc, Mutex, OnceLock};

use crate::allowance::Allowance;
use crate::error::{Error, Result};

/// **Bytes admitted per namespace, monotonically.**
///
/// One entry per namespace the process has ever admitted an upload for, and
/// entries are never removed: the map is bounded by the size of the catalog,
/// each entry is a namespace name and eight bytes, and removing them would mean
/// counting live holders, which is the reservation bookkeeping this design
/// exists to avoid. A box with ten thousand namespaces spends well under a
/// megabyte on it, once.
#[derive(Debug, Default)]
pub struct AllowanceLedger {
    admitted: Mutex<HashMap<String, u64>>,
}

/// The one ledger the whole process shares **for `lfs_bytes`**.
///
/// A `static` rather than a field on `Control` or `Dispatch`, and the reason is
/// the property being defended: a namespace's room is a fact about the BOX, and
/// two doors that shared a tenant but not a ledger would be exactly the bug
/// again, one wire over. `gunnar-server` builds a `Dispatch` on more than one
/// path, and nothing in the type system would have caught a second one holding
/// a second ledger. Tests that want isolation build their own with
/// [`AllowanceLedger::detached`].
static SHARED: OnceLock<Arc<AllowanceLedger>> = OnceLock::new();

/// The same, for `pack_bytes`. See [`AllowanceLedger::shared_pack`] for why
/// this is a second counter and not the same one.
static SHARED_PACK: OnceLock<Arc<AllowanceLedger>> = OnceLock::new();

impl AllowanceLedger {
    /// The process-wide ledger for **`lfs_bytes`**. See [`SHARED`].
    pub fn shared() -> &'static Arc<AllowanceLedger> {
        SHARED.get_or_init(|| Arc::new(AllowanceLedger::default()))
    }

    /// The process-wide ledger for **`pack_bytes`** — the push door's counter,
    /// used by `gunnar_ssh::server::Dispatch::ask_plan` for the mid-transfer
    /// pack ceiling exactly as [`Self::shared`] is used for the LFS allowance.
    ///
    /// # Why a SECOND counter, when a push and an LFS upload fill the same disk
    ///
    /// Because they are billed against different numbers. `pack_bytes` and
    /// `lfs_bytes` are separate caps on the signed entitlement fact and are
    /// reported separately by `Usage.Read`; `gunnar_ssh::PlanGate::pack_ceiling`
    /// is `caps["pack_bytes"] - held_pack + slack` and
    /// `PlanGate::lfs_allowance` is `caps["lfs_bytes"] - held_lfs`. Two
    /// admissions judged against *different rooms* cannot share one counter: an
    /// LFS upload charged into the pack counter would shrink a push's ceiling
    /// by bytes that never moved the pack cap, and the tenant would be cut
    /// mid-push against room nobody had spent. That is a quota that refuses
    /// wrongly, which is the same defect as one that admits wrongly, pointing
    /// the other way.
    ///
    /// The map is keyed by namespace either way, so this costs one more `u64`
    /// per namespace that has ever pushed and nothing else. If the two meters
    /// are ever merged on the entitlement fact, these two statics merge with
    /// them and every holder's arithmetic is unchanged.
    pub fn shared_pack() -> &'static Arc<AllowanceLedger> {
        SHARED_PACK.get_or_init(|| Arc::new(AllowanceLedger::default()))
    }

    /// A ledger of its own, for a test that wants one namespace's arithmetic
    /// without every other test in the binary sharing it.
    pub fn detached() -> Arc<AllowanceLedger> {
        Arc::new(AllowanceLedger::default())
    }

    /// **Snapshot the counter for `namespace`. Call this BEFORE measuring the
    /// disk** — see the module comment for why the order is load-bearing and
    /// not a matter of taste.
    pub fn mark(self: &Arc<Self>, namespace: &str) -> Mark {
        let at = self.lock().get(namespace).copied().unwrap_or(0);
        Mark {
            ledger: Arc::clone(self),
            namespace: namespace.to_owned(),
            at,
        }
    }

    /// **Read the charge and apply it under ONE lock.** Two acquisitions — read
    /// the total, decide, add — would be the very check-then-use this file
    /// exists to close, moved down one level.
    ///
    /// `true` when the object was admitted and charged; `false` when it does not
    /// fit, in which case nothing was charged and a refused upload costs the
    /// tenant nothing.
    fn charge(&self, namespace: &str, at: u64, size: u64, room: u64) -> bool {
        let mut admitted = self.lock();
        let total = admitted.entry(namespace.to_owned()).or_insert(0);
        // `saturating_sub`: a mark can be AHEAD of the counter only if the map
        // entry were ever reset, which nothing does — and 0 ("nothing has been
        // admitted since you looked") is the conservative reading either way.
        let since = total.saturating_sub(at);
        // `saturating_add` because `size` is a number a hostile client
        // declared. `u64::MAX` must refuse, not wrap into room that is not
        // there — the same argument `Allowance::admit` makes about the same
        // untrusted integer.
        if since.saturating_add(size) > room {
            return false;
        }
        *total = total.saturating_add(size);
        true
    }

    /// **Charge as much of `want` as still fits, and say how much that was.**
    ///
    /// [`Self::charge`] answers a yes/no about a size the client DECLARED, which
    /// is the LFS shape: an object announces its length before its bytes.
    /// A packfile announces an object count and no byte count at all, so the
    /// push door cannot ask "may I take these N bytes" about the transfer —
    /// only about the chunk in its hand. This is that question: partial credit
    /// instead of a refusal, so a push that fits in the room left is read to its
    /// end rather than turned away for asking about a buffer bigger than the
    /// remainder.
    ///
    /// Read and apply under ONE lock, for the same reason [`Self::charge`] is:
    /// two acquisitions would be the very check-then-use this file exists to
    /// close, moved down one level.
    fn charge_upto(&self, namespace: &str, at: u64, want: u64, room: u64) -> u64 {
        let mut admitted = self.lock();
        let total = admitted.entry(namespace.to_owned()).or_insert(0);
        let since = total.saturating_sub(at);
        // `saturating_sub` on both: a mark ahead of the counter and a room
        // already overspent both read as "nothing left", which is the
        // conservative answer and the one that cannot hand out room twice.
        let grant = want.min(room.saturating_sub(since));
        *total = total.saturating_add(grant);
        grant
    }

    /// The map, whether or not a thread panicked while holding it.
    ///
    /// A panic in the four lines this lock covers cannot leave a `u64` in a
    /// state anyone has to distrust, and refusing every LFS upload on the box
    /// because one connection panicked would be precisely the lockout this
    /// design is built to make impossible.
    fn lock(&self) -> std::sync::MutexGuard<'_, HashMap<String, u64>> {
        self.admitted
            .lock()
            .unwrap_or_else(std::sync::PoisonError::into_inner)
    }
}

/// **Where one transfer's holder stands in the namespace's ledger** — the value
/// the counter had when this transfer measured the disk.
///
/// `Clone` is a real clone of the same standing, not a new one: the request, the
/// [`crate::Session`] built from it and `LfsAccess`'s copy are one transfer that
/// happens to be held in three places, and giving a clone a fresh mark would
/// hide every byte charged in between from it.
#[derive(Clone, Debug)]
pub struct Mark {
    ledger: Arc<AllowanceLedger>,
    namespace: String,
    at: u64,
}

impl Mark {
    /// The namespace this mark stands in, for a log line.
    pub fn namespace(&self) -> &str {
        &self.namespace
    }

    /// **Charge up to `want` bytes against this standing, inside a total of
    /// `room`, and return what was charged.**
    ///
    /// The streaming twin of [`NamespaceAllowance::admit`], for a door that
    /// learns the size of a transfer one chunk at a time instead of being told
    /// it up front: the pushed-packfile ceiling
    /// (`gunnar_wire::PackLedger`, wired up in `gunnar_server::transport`).
    /// `room` is passed per call rather than held here because a `Mark` is a
    /// standing in a counter and not a policy — the number comes from whichever
    /// gate was tightest, and the one byte that makes a ceiling *reachable* is
    /// the receive loop's own arithmetic, spelled where that loop's `room()`
    /// spells it and nowhere else.
    ///
    /// A grant of `0` means the namespace's room is gone: everything admitted
    /// since this holder marked already fills it. Nothing is reserved and
    /// nothing is given back — see the module comment for why that is the
    /// design and not an omission.
    pub fn admit_upto(&self, want: u64, room: u64) -> u64 {
        self.ledger
            .charge_upto(&self.namespace, self.at, want, room)
    }
}

/// **An [`Allowance`] that knows what else is in flight against the same
/// namespace.**
///
/// Everything the plain [`Allowance`] is — a byte count and the sentence a
/// tenant who runs out reads — plus the one thing it could not have: the other
/// uploads. `gunnar-server` still computes the number, this still carries it to
/// where the bytes are, and the only new fact is which namespace's room it is a
/// slice of.
#[derive(Clone, Debug)]
pub struct NamespaceAllowance {
    allowance: Allowance,
    mark: Mark,
}

impl NamespaceAllowance {
    /// Bind a measured allowance to the ledger standing that was taken before
    /// it was measured.
    pub fn new(mark: Mark, allowance: Allowance) -> Self {
        NamespaceAllowance { allowance, mark }
    }

    /// The measured number and the refusal, for the surfaces that only report:
    /// the batch endpoint's advisory answer and the transport's log line. A
    /// caller that is about to take bytes wants [`Self::admit`] instead — this
    /// one cannot charge and does not know about anything else in flight.
    pub fn allowance(&self) -> &Allowance {
        &self.allowance
    }

    /// **May one more object of `size` bytes land, counting everything admitted
    /// against this namespace since this transfer measured the disk?**
    ///
    /// Charges on success and nothing on refusal. `Err` is
    /// [`Error::NoRoomLeft`] carrying [`Allowance::refusal`] verbatim, exactly
    /// as [`Allowance::admit`]'s is, so the client reads the same 507 and the
    /// same sentence whether it was the tenant's own bytes or a sibling upload's
    /// that filled the plan.
    ///
    /// There is no `taken` parameter, and its absence is the fix. A caller used
    /// to pass what IT had committed — the HTTP wire passed `0`, one session
    /// passed its own running total — and that was the whole bug: every caller
    /// could only ever count itself. The ledger counts them all, including this
    /// transfer's own earlier objects, so a session's running total is not lost,
    /// it is subsumed.
    pub fn admit(&self, size: u64) -> Result<()> {
        if self.mark.ledger.charge(
            &self.mark.namespace,
            self.mark.at,
            size,
            self.allowance.bytes,
        ) {
            Ok(())
        } else {
            Err(Error::NoRoomLeft(self.allowance.refusal.clone()))
        }
    }
}