gunnar-ssh 3.0.0

//! **The two load-shedding gates**: a CoDel delay governor (time) and a memory
//! reservation ledger (space). Both refuse *at admission*, before a single byte
//! of the operation's work has been allocated.
//!
//! [`Admission`](crate::Admission) is the one door; this module is the pair of
//! locks on it. An operation is admitted only if **both** pass:
//!
//! ```text
//! admit(class):
//!     if occupied >= governed_limit          -> DENY (time:  the queue is standing)
//!     if reserved + est(class) > budget      -> DENY (space: the budget is spoken for)
//!     occupied += 1;  reserved += est(class)     // released by Drop, on every path
//! ```
//!
//! # Gate 1 — CoDel, because it is the one that does not oscillate
//!
//! ## Sojourn, not queue length
//!
//! Queue *length* is not a saturation signal. A hundred-deep queue draining in
//! a millisecond is a healthy server doing its job; a three-deep queue that has
//! not moved in a second is a server that has stopped. What separates them is
//! how long an item **waited**, so that is what is measured: the interval
//! between admission (enqueue) and the moment the operation's work actually
//! starts running ([`AdmissionGuard::entered_service`](crate::AdmissionGuard::entered_service)).
//!
//! In gunnar that interval is a real queue and not a bookkeeping fiction. Every
//! transfer is `tokio::spawn`ed and then bridged onto a blocking-pool thread
//! (`crate::admission` module docs); the gap between "admitted" and "the task is
//! running" is the runtime's own ready queue plus the blocking-pool handoff.
//! That is the *invisible* ceiling behind the unbounded listener, measured directly
//! instead of inferred.
//!
//! ## Minimum, not average
//!
//! Over each [`INTERVAL`] window the controller keeps the **minimum** sojourn
//! per class, never the mean. One item that got through quickly is proof the
//! queue is not standing — it proves a path existed. An average is moved by a
//! single outlier, so an averaging controller shrinks on noise, then grows back
//! when the outlier ages out, then shrinks again: the oscillation CoDel exists
//! to remove.
//!
//! ## Per class, and this is the crux
//!
//! **One queue. One shared limit. Sojourn measured per class.**
//!
//! The trap is head-of-line blocking of the *signal*, not of the queue. Hold
//! one target for everything and a 25-minute `linux.git` clone — which does not
//! care about a 150 ms wait and whose client would never notice one — reads as a
//! catastrophically standing queue against a 5 ms bar, and the controller sheds
//! load that the box could have served. The clone poisons the delay signal for
//! the 10 ms `ls-remote` sharing its queue.
//!
//! So each class carries its own [`OpClass::target`], and the rule that sets it
//! is **half the class's measured median service time** — a client may be made
//! to wait at most long enough to add 50 % to its own latency. Those medians are
//! measured, not picked round: the reference desktop, 2026-08-13, gunnar in an
//! 8 GiB / 8-cpu cgroup.
//!
//! | class | measured median | target | note |
//! |---|---:|---:|---|
//! | ls-remote | 9–12 ms | **5 ms** | the base [`TARGET`] falls out of the measurement |
//! | fetch | 90 ms | 45 ms | `fetch`/`diverge` |
//! | push | 340 ms | 170 ms | `push`/`base` |
//! | clone | 372 ms | 186 ms | `clone`/`large`; xl was 4350 ms |
//! | gc | **NOT MEASURED** | 186 ms | no bakeoff row; borrows clone's, and says so |
//!
//! `ls-remote` is therefore the canary: it trips first, at exactly the 5 ms the
//! design asked for, and the slower classes trip only when the wait is large
//! enough to matter *to them*.
//!
//! ## The control law
//!
//! When a class's window closes, that class's window-minimum decides, and what
//! it moves is the **one shared limit**:
//!
//! * minimum above the class's target for the whole window → the queue is
//!   standing → shrink multiplicatively, `limit -= max(limit/8, 1)`;
//! * otherwise → grow additively, `limit += 1`.
//!
//! Clamped to [`LimitBand`]: **[10, 100]** when no ceiling is configured, and
//! **[10, ceiling]** when one is — an operator's explicit number is never
//! lowered to the design's default. The `max(_, 1)` is
//! not cosmetic: without it a limit below 8 has an integer shrink step of zero
//! and the controller can never come down again.
//!
//! It starts at the top of the band. A load shedder has nothing to shed on an
//! idle server, and starting at the floor would make a cold server refuse work
//! it could do for the ~90 intervals (9 s) the additive climb takes.
//!
//! # Gate 2 — memory as a RESERVATION you own, not a measurement you take
//!
//! [`MemoryLedger`] is a budget and a counter. It does not ask the operating
//! system anything, and that is the design rather than a limitation of it:
//!
//! * a free-memory reading is **stale the instant you take it**; a reservation
//!   you own is exact;
//! * it denies **before** the allocation, which is the whole principle — asking
//!   "is memory low?" after an operation has built a multi-gigabyte buffer is
//!   asking after the damage;
//! * it is the fixed-pool shape: admission is "is a slab free?".
//!
//! ## Why this gate is not theoretical for gunnar — measured
//!
//! From the same bakeoff table, peak anonymous RSS on `vs_forge_clone`, on
//! identical wire bytes: **gunnar 2314 MB, gitea 122 MB, Forgejo 167 MB**.
//! Memory is the resource gunnar actually exhausts. The cause is known —
//! `EmitEntry.stored` in `znippy-plugin-git/src/pack_walk.rs:931` is an owned
//! `Vec<u8>` per object, so a clone materialises the whole pack in RAM before
//! writing a byte — and it is being fixed elsewhere. It is not fixed here.
//!
//! ## The honest limitation, stated rather than papered over
//!
//! **A memory budget is only meaningful when an operation's footprint is bounded
//! and predictable, and a clone's is neither today.** It is proportional to the
//! pack, and for `linux.git` that is ~6.4 GB across ~27.6 M allocations. So
//! [`OpClass::estimated_bytes`] is a per-class **constant**, and for
//! [`OpClass::Clone`] that constant is a **guess** — the marginal figure from
//! one corpus, not a bound. Until `stored` becomes an extent into the mapped
//! archive rather than an owned copy, this gate does not bound clone memory
//! correctly, and nothing here should be read as claiming it does. What it does
//! do today is stop *unbounded concurrency* of a heavyweight class, which is a
//! smaller but real property.
//!
//! # OS-less, by law
//!
//! No `/proc`, no `sysinfo`, no `getrusage`, no CPU/disk/memory probing. The
//! only inputs are [`std::time::Instant`] and in-process counters, so this works
//! on bare metal and in a container with equal honesty. The corollary is stated
//! plainly on [`DEFAULT_MEMORY_BUDGET_BYTES`]: this code **cannot** know how much
//! RAM the box has, so the default is a constant an operator on a small machine
//! must lower.
//!
//! No rayon, no thread pool, no allocation on the decision path (the multicore
//! rule: gatling, never rayon): the
//! per-class windows are a fixed `[ClassWindow; OpClass::COUNT]` array behind one
//! mutex touched once per transfer, not once per byte.

use std::sync::atomic::{AtomicBool, AtomicU64, AtomicUsize, Ordering};
use std::sync::Mutex;
use std::time::{Duration, Instant};

use crate::command::GitService;
use gunnar_lfs::Operation;

/// The base sojourn a queue may stand at before it counts as standing: the
/// target for the *fastest* class, [`OpClass::LsRemote`], whose measured median
/// service time is 9–12 ms.
///
/// Every other class scales off its own measurement; see [`OpClass::target`].
pub const TARGET: Duration = Duration::from_millis(5);

/// How long a class's minimum sojourn must stay above its target before the
/// shared limit moves.
///
/// One window, one decision. Shorter and the controller reacts to a single slow
/// scheduling turn; longer and a genuinely standing queue keeps standing while
/// the controller thinks about it.
pub const INTERVAL: Duration = Duration::from_millis(100);

/// The floor the governed limit may never go below.
///
/// A load shedder that can shed itself to zero is a self-inflicted outage. Ten
/// concurrent git operations is a server that is still a server.
pub const MIN_LIMIT: usize = 10;

/// The ceiling the governed limit may never go above **when no budget was
/// declared**. A declared budget is its own ceiling; see
/// [`LimitBand::for_budget`].
pub const MAX_LIMIT: usize = 100;

/// A declared budget's floor is `budget / this`.
///
/// Half. The governor may shed a saturated box down to half of what the box was
/// measured to be able to hold, and no further: the budget's three terms —
/// cores, descriptors, `MemAvailable` — are hard resources, and a controller
/// reading a *latency* signal has no evidence about them. Below half, the
/// honest answer to a client is to serve it.
pub const BUDGET_FLOOR_DENOMINATOR: usize = 2;

/// The default self-imposed memory budget for admitted operations: 8 GiB.
///
/// **This is a number about gunnar, not about the machine.** By the OS-less law
/// this code cannot read the box's RAM, so it cannot size itself; 8 GiB is the
/// cgroup the bakeoff measurement above was taken in
/// (measured 2026-08-13), which makes it the one
/// figure with evidence behind it rather than a round guess.
///
/// An operator on a smaller box **must** lower it, and one on a larger box
/// should raise it. It covers only the footprints of operations admitted through
/// this gate: gunnar's resident baseline (~330 MB, the `ls-remote` rows, which
/// send no pack at all) sits outside the budget and must be left as headroom.
pub const DEFAULT_MEMORY_BUDGET_BYTES: u64 = 8 * 1024 * 1024 * 1024;

const MIB: u64 = 1024 * 1024;

/// How long an operation may **wait** for a slot before it is refused.
///
/// # Why there is a wait at all, when this module argued there must not be
///
/// The original text (still above, and still right about *unbounded* queues)
/// said a git client cannot see a queue, so a fast refusal beats a silent
/// wait. Two of its four premises did not survive measurement:
///
/// * **"Queueing keeps the expensive things alive."** So does refusing. A
///   refused client does not go away: it reconnects, and a reconnect is a
///   *whole* SSH handshake — a new TCP connection, a new key exchange, a new
///   signature verification — where a wait is one parked future. MEASURED on
///   the rig (2026-08-30): at 64 concurrent clients the refusal arm
///   spent more CPU per **served** operation than the waiting arm, because
///   three quarters of the handshakes bought nothing.
/// * **"A queue makes saturation invisible again."** Only an *unmeasured* one
///   does. This queue is bounded, counted
///   ([`crate::AdmissionStats::admitted_after_wait`],
///   [`crate::AdmissionStats::peak_wait_ms`]) and it is the CoDel controller's
///   input, so it is strictly more visible than the refusal it replaces.
///
/// Two seconds because that is well under what a client pays to retry — git
/// itself has no backoff, so the retry is the user's or CI's, in whole seconds
/// — and well under [`crate::SshConfig::exec_grace`] (60 s) and
/// [`crate::SshConfig::inactivity_timeout`] (600 s), so a waiting session can
/// never be disconnected by the timeouts that exist to kill idle ones.
///
/// `Duration::ZERO` restores the 2026-08-05 behaviour exactly: refuse on the
/// spot. That is what `--admission-wait-ms 0` is for, and it is the arm the
/// before/after table was measured against.
pub const DEFAULT_ADMISSION_WAIT: Duration = Duration::from_secs(2);

/// **How long an AUTHENTICATED caller may be queued for a slot**: 30 s.
///
/// The two seconds above are right for the anonymous caller this wait was
/// argued against, and wrong for a key. MEASURED on gunnar.rs, 2026-10-25:
/// every CI lane pushes and fetches as one key, many at once, and git-lfs's
/// pure-SSH `git-lfs-transfer` has no retry — a refused session falls back to
/// `git-lfs-authenticate` and the whole push fails with *"batch request: exec
/// request failed"*. A key's parked sessions are bounded by its own share, so
/// queueing it costs one future per session and saves a failed CI job.
///
/// Thirty, not more: well under [`crate::SshConfig::exec_grace`] (60 s), so a
/// queued session is never cut off by the timeout that kills idle ones.
pub const DEFAULT_KEYED_ADMISSION_WAIT: Duration = Duration::from_secs(30);

/// The share of a transfer's **wire bytes** charged to the memory budget as it
/// moves ([`crate::AdmissionGuard::reserve_more`]).
///
/// A quarter. MEASURED, not guessed: the spool bounds mean the pushed pack
/// itself is spooled past 64 MiB and read back as a read-only mmap (page cache,
/// reclaimable, and `MemAvailable` counts it as free), so what a transfer
/// really holds is the per-object bookkeeping — the walk's entry table
/// (~56 B/object), the closure check's offset set (~16 B), gunnar-wire's
/// visited sets (~100 B) — which came to ~2.5 GB against `linux.git`'s 8.58 GiB
/// pack (the memory limit). 2.5 / 8.58 is 0.29; a quarter
/// is that, rounded to a shift.
///
/// It is a *charge*, not a prediction. Its job is that a transfer which turns
/// out to be enormous stops **new** admissions long before the box is in
/// trouble, without a three-object push-create having to pay for the
/// possibility.
pub const WIRE_MEMORY_NUMERATOR: u64 = 1;
/// The denominator of [`WIRE_MEMORY_NUMERATOR`].
pub const WIRE_MEMORY_DENOMINATOR: u64 = 4;

/// How many bytes must cross the wire between two reservation top-ups.
///
/// The ledger is one atomic compare-and-swap, so this is not about the cost of
/// the call; it is about not turning a per-byte quantity into a per-byte
/// *decision*. 16 MiB is a quarter of the 64 MiB spool threshold, so a
/// transfer that is about to spool has already told the ledger about itself
/// three times.
pub const WIRE_RESERVE_STEP: u64 = 16 * MIB;

/// The three cost shapes a transfer can have, and the thing a lane floor
/// protects.
///
/// See [`OpClass::lane`] for why the split exists at all, and
/// [`Governor::lane_ceiling`] for what a floor actually reserves.
#[derive(Debug, Clone, Copy, PartialEq, Eq, PartialOrd, Ord, Hash)]
pub enum Lane {
    /// A ref advertisement: no pack, milliseconds, 4 MiB.
    Cheap,
    /// A pack **out** of the store — clone and fetch. Mostly page cache.
    Read,
    /// A pack **in**, and the indexer behind it — push and gc.
    Write,
}

impl Lane {
    /// Every lane, in index order; the per-lane arrays are sized from this.
    pub const ALL: [Lane; 3] = [Lane::Cheap, Lane::Read, Lane::Write];
    /// How many lanes there are.
    pub const COUNT: usize = Lane::ALL.len();

    /// Index into the per-lane arrays.
    pub const fn index(self) -> usize {
        match self {
            Lane::Cheap => 0,
            Lane::Read => 1,
            Lane::Write => 2,
        }
    }

    /// The name a log or a counter uses.
    pub const fn name(self) -> &'static str {
        match self {
            Lane::Cheap => "cheap",
            Lane::Read => "read",
            Lane::Write => "write",
        }
    }

    /// The slots of the governed limit this lane is **guaranteed**, as
    /// `limit * numerator / denominator`, never below [`Lane::floor_min`].
    ///
    /// An eighth for `cheap` and a quarter for `read`: a `write` storm can take
    /// at most five eighths of the box, so an `ls-remote` and a clone always
    /// have somewhere to land. `write` reserves nothing, because it is the lane
    /// that would starve the others and it is never the one that is starved —
    /// whatever the other two are not using is its.
    pub const fn floor_share(self) -> (usize, usize) {
        match self {
            Lane::Cheap => (1, 8),
            Lane::Read => (1, 4),
            Lane::Write => (0, 1),
        }
    }

    /// The smallest floor a lane keeps however small the limit is.
    ///
    /// Two, not one: one slot means a single stuck operation closes the lane,
    /// and a lane floor exists precisely so that a lane cannot be closed by
    /// what is happening in another one.
    pub const fn floor_min(self) -> usize {
        match self {
            Lane::Cheap | Lane::Read => 2,
            Lane::Write => 0,
        }
    }

    /// This lane's guaranteed slots under a governed limit of `limit`.
    pub fn floor(self, limit: usize) -> usize {
        let (n, d) = self.floor_share();
        (limit * n / d).max(self.floor_min()).min(limit)
    }
}

/// Which kind of user-level git operation is being admitted.
///
/// The classes exist for **one** reason: a sojourn measured across a mixed queue
/// is a number about whichever operation happened to be slowest, so the delay
/// signal must be split even though the queue is not.
#[derive(Debug, Clone, Copy, PartialEq, Eq, PartialOrd, Ord, Hash)]
pub enum OpClass {
    /// A ref advertisement and nothing else. The canary: fastest, cheapest, and
    /// the first to notice a queue that has started to stand.
    LsRemote,
    /// An incremental fetch — a negotiated pack over an existing history.
    Fetch,
    /// A full clone. The expensive one, in time and (today, badly) in memory.
    Clone,
    /// A push, including LFS upload.
    Push,
    /// Repository maintenance.
    Gc,
}

impl OpClass {
    /// Every class, in index order. The per-class arrays are sized from this,
    /// so a new variant that is not added here fails to compile rather than
    /// silently sharing another class's window.
    pub const ALL: [OpClass; 5] = [
        OpClass::LsRemote,
        OpClass::Fetch,
        OpClass::Clone,
        OpClass::Push,
        OpClass::Gc,
    ];

    /// How many classes there are; the width of every per-class array here.
    pub const COUNT: usize = OpClass::ALL.len();

    /// Index into the per-class arrays.
    pub const fn index(self) -> usize {
        match self {
            OpClass::LsRemote => 0,
            OpClass::Fetch => 1,
            OpClass::Clone => 2,
            OpClass::Push => 3,
            OpClass::Gc => 4,
        }
    }

    /// The name a log or a counter uses.
    pub const fn name(self) -> &'static str {
        match self {
            OpClass::LsRemote => "ls-remote",
            OpClass::Fetch => "fetch",
            OpClass::Clone => "clone",
            OpClass::Push => "push",
            OpClass::Gc => "gc",
        }
    }

    /// The sojourn **this class** may stand at before the shared limit shrinks.
    ///
    /// Half the class's measured median service time
    /// (measured on the reference desktop,
    /// 2026-08-13): a client of
    /// this class may be made to wait at most long enough to add 50 % to its own
    /// latency. That rule, not a table of round numbers, is what stops a slow
    /// class from shrinking the limit on a fast class's behalf — and it lands
    /// [`OpClass::LsRemote`] exactly on [`TARGET`], because its measured median
    /// is 10 ms.
    pub const fn target(self) -> Duration {
        match self {
            // measured 9–12 ms → 5 ms, which IS `TARGET`.
            OpClass::LsRemote => TARGET,
            // measured 90 ms (`fetch`/`diverge`).
            OpClass::Fetch => Duration::from_millis(45),
            // measured 372 ms (`clone`/`large`; `xl` was 4350 ms, so this is the
            // conservative end of the measured range).
            OpClass::Clone => Duration::from_millis(186),
            // measured 340 ms (`push`/`base`).
            OpClass::Push => Duration::from_millis(170),
            // NOT MEASURED: the bakeoff run of 2026-08-13 has no `vs_forge_gc`
            // row, so this borrows clone's rather than inventing a figure. Say
            // so here rather than let a reader assume it was measured.
            OpClass::Gc => Duration::from_millis(186),
        }
    }

    /// What this class reserves **at admission**, before it knows its size.
    ///
    /// # This used to be the whole footprint, and that was the ceiling
    ///
    /// Until 2026-08-30 this returned a per-class **constant equal to the whole
    /// operation's measured peak** — 320 MiB for a push, 2 GiB for a clone —
    /// and [`MemoryLedger`] refused anything that did not fit in
    /// [`DEFAULT_MEMORY_BUDGET_BYTES`]. That is an arithmetic ceiling nobody
    /// chose: 8 GiB / 320 MiB = **25 concurrent pushes on any box**, whatever
    /// its RAM, and a three-object push-create reserved exactly as much as a
    /// `linux.git` mirror.
    ///
    /// MEASURED on the native desktop rig (gunnar-solo pinned to 4 cpus / 8 GiB,
    /// 2026-08-30): 500 push-creates at 64 concurrent clients → **118–132 of
    /// them refused `MemoryBudget`** while the server's RSS never passed
    /// 1.6 GB of its 8 GiB. The gate was refusing three quarters of the work to
    /// protect a budget that was 20 % used.
    ///
    /// # So the reservation starts small and GROWS with the bytes on the wire
    ///
    /// This is the *entry* cost — the session, the state machine, the ref
    /// advertisement, the first buffers — and it is what admission is decided
    /// on. Everything proportional to the transfer is added afterwards by
    /// [`crate::AdmissionGuard::reserve_more`], which the transport calls as
    /// the pack actually arrives ([`WIRE_MEMORY_NUMERATOR`]). A gate that
    /// charges a small push for a large one refuses work the box can do; a
    /// gate that never charges the large one takes the box down. Charging what
    /// has actually moved is the only rule that is right in both directions.
    ///
    /// The numbers below are marginal RSS per concurrent transfer, measured on
    /// the same rig by walking the concurrency up with the gate wide open
    /// (`rss-slope.tsv`, a scratch capture outside the repository), rounded up
    /// to the next power of
    /// two.
    pub const fn admission_bytes(self) -> u64 {
        match self {
            // No pack at all — the ref advertisement and its framing.
            OpClass::LsRemote => 4 * MIB,
            // A negotiation and the first window of a pack.
            OpClass::Fetch => 16 * MIB,
            // Same entry cost as a fetch; a clone's real footprint is its pack,
            // and its pack is charged as it is written.
            OpClass::Clone => 16 * MIB,
            // The receive state machine plus the first spool window. MEASURED
            // 39 MB marginal per concurrent tiny push-create on the rig.
            OpClass::Push => 32 * MIB,
            // NOT MEASURED — no bakeoff row, no rig row. Sized at a push's,
            // and flagged rather than asserted.
            OpClass::Gc => 32 * MIB,
        }
    }

    /// The **lane** this class is metered in.
    ///
    /// One queue is still one queue, but a lane may not be starved by another:
    /// an `ls-remote` that costs 4 MiB and 10 ms must not wait behind an 8 GiB
    /// mirror push, and on a saturated box it did — every class shared one
    /// counter against one limit, so a push storm took every slot.
    ///
    /// Three lanes, because there are three cost shapes: the ref advertisement
    /// (no pack), a read (a pack out of the store, mostly page cache), and a
    /// write (a pack in, plus the indexer behind it).
    pub const fn lane(self) -> Lane {
        match self {
            OpClass::LsRemote => Lane::Cheap,
            OpClass::Fetch | OpClass::Clone => Lane::Read,
            OpClass::Push | OpClass::Gc => Lane::Write,
        }
    }

    /// The class an SSH `exec` request belongs to.
    ///
    /// # What SSH cannot tell you, and it matters
    ///
    /// `clone`, `fetch` and `ls-remote` are **the same exec line**:
    /// `git-upload-pack '<path>'`. Which of the three it turns out to be is
    /// decided later, by the protocol-v2 command the client sends inside the
    /// stream (`ls-refs` versus `fetch`), and by whether the client had any
    /// history to negotiate from. The admission gate runs before any of that,
    /// by design — the entire value of this gate is refusing before the cost is
    /// paid.
    ///
    /// So over SSH, `git-upload-pack` is classified [`OpClass::Fetch`]: the
    /// middle of the three, chosen deliberately rather than by accident.
    /// Classifying it as `Clone` would give every `ls-remote` a 2 GiB
    /// reservation and a 186 ms delay target; as `LsRemote` it would give every
    /// clone a 4 MiB reservation and a 5 ms target, which is the poisoned signal
    /// this module exists to avoid.
    ///
    /// Callers that **do** know — the HTTP smart transport, which sees
    /// `GET /info/refs` separately from `POST /git-upload-pack`, and `gunnar gc`,
    /// which is not a transport at all — pass the exact class to
    /// [`Admission::try_admit`](crate::Admission::try_admit) instead of going
    /// through this function.
    pub fn for_ssh_service(service: GitService) -> OpClass {
        match service {
            GitService::UploadPack => OpClass::Fetch,
            GitService::ReceivePack => OpClass::Push,
            GitService::LfsTransfer(Operation::Upload) => OpClass::Push,
            GitService::LfsTransfer(Operation::Download) => OpClass::Fetch,
        }
    }
}

impl std::fmt::Display for Lane {
    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
        f.write_str(self.name())
    }
}

impl std::fmt::Display for OpClass {
    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
        f.write_str(self.name())
    }
}

/// The range the governed limit is clamped to.
///
/// Spelled as a type rather than two constants because it has to be **narrowed
/// by an operator's configured ceiling**: a server told `--max-concurrent-transfers 4`
/// must serve 4, not the design's floor of 10. See [`LimitBand::for_ceiling`].
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub struct LimitBand {
    /// The smallest the controller may shrink to.
    pub min: usize,
    /// The largest it may grow to, and where it starts.
    pub max: usize,
}

impl LimitBand {
    /// The design's band: **[10, 100]**.
    pub const DEFAULT: LimitBand = LimitBand {
        min: MIN_LIMIT,
        max: MAX_LIMIT,
    };

    /// The band for a server whose operator pinned a hard ceiling.
    ///
    /// The operator's number is the ceiling, and the FLOOR is clamped under it:
    /// pinning 4 gives `[4, 4]`, a governor that cannot move, which is exactly
    /// what "I told you 4" should mean. A configured ceiling below
    /// [`MIN_LIMIT`] must not be silently RAISED — a gate admitting more than
    /// it was told to is the one failure mode a limit may not have.
    ///
    /// **The ceiling is NOT capped at [`MAX_LIMIT`], and that was a bug.** The
    /// first form did `MAX_LIMIT.min(ceiling)`, so a server configured with the
    /// shipped default of 256 got a band of `[10, 100]` and quietly served 100.
    /// Lowering an operator's explicit number is the same defect as raising it,
    /// in the other direction: both mean the configured value is not the value.
    /// `MAX_LIMIT` is the design's default for a server that has NOT been told a
    /// number — it is not an opinion that outranks one that has.
    pub fn for_ceiling(ceiling: usize) -> LimitBand {
        LimitBand {
            min: MIN_LIMIT.min(ceiling),
            max: ceiling,
        }
    }

    /// The band for a server whose ceiling is a **declared, measured budget**
    /// rather than a round number an operator typed.
    ///
    /// # Why the floor moves with the ceiling — MEASURED
    ///
    /// [`LimitBand::for_ceiling`] pins the floor at [`MIN_LIMIT`] = 10. With the
    /// shipped ceiling of 256 that is a band of `[10, 256]`, and the controller
    /// walks the whole of it: multiplicative decrease is `limit -= limit/8`, so
    /// 256 → 224 → 196 → … → 24 → 21 → **19** → 17 … On the appliance under a
    /// 64-client push storm every 100 ms window stands, and the measurement
    /// took the gate refusing by name at exactly that — `already serving its
    /// limit of 19`, 268 of 1000 push-creates admitted. **19 was not a constant
    /// anybody chose; it is a point on a descent.**
    ///
    /// The descent does not stop because shedding does not fix what the
    /// controller is measuring. The sojourn it reads was the gap between
    /// admission and the spawned task's first line — the tokio ready queue and
    /// the blocking-pool handoff — and that gap is set by how many clients are
    /// *connected and doing crypto*, not by how many transfers are admitted. A
    /// refused client reconnects, so a refusal makes the input worse. A
    /// controller with no authority over its own input runs to its floor.
    ///
    /// Both halves are fixed here. The controller's input is now the
    /// **admission wait** — the queue this gate actually owns
    /// ([`DEFAULT_ADMISSION_WAIT`]) — and its floor is a share of the budget
    /// rather than a constant: below [`BUDGET_FLOOR_DENOMINATOR`]`⁻¹` of a
    /// number derived from cores, descriptors and `MemAvailable`, the box is
    /// provably not the bottleneck and refusing costs more than serving.
    pub fn for_budget(budget: usize) -> LimitBand {
        LimitBand {
            min: (budget / BUDGET_FLOOR_DENOMINATOR).clamp(MIN_LIMIT.min(budget), budget),
            max: budget,
        }
    }

    fn clamp(self, limit: usize) -> usize {
        limit.clamp(self.min, self.max)
    }
}

/// One class's sliding window.
///
/// `opened` is `None` until the first sample, so construction never reads the
/// clock and a test can drive the whole controller off `Instant`s of its own.
#[derive(Debug, Clone, Copy)]
struct ClassWindow {
    opened: Option<Instant>,
    /// The **minimum** sojourn seen since the window opened. Never a sum, never
    /// a count: an average would let one outlier shrink the limit.
    min: Option<Duration>,
}

impl ClassWindow {
    const EMPTY: ClassWindow = ClassWindow {
        opened: None,
        min: None,
    };
}

/// The CoDel controller: per-class delay windows, one shared limit.
#[derive(Debug)]
pub struct Codel {
    band: LimitBand,
    /// The one shared limit. Read on every admission, written once per closed
    /// window.
    limit: AtomicUsize,
    /// Fixed-size and preallocated: no map, no allocation, no growth.
    windows: Mutex<[ClassWindow; OpClass::COUNT]>,
    shrinks: AtomicU64,
    grows: AtomicU64,
}

impl Codel {
    /// A controller starting at the top of its band.
    pub fn new(band: LimitBand) -> Codel {
        Codel {
            band,
            limit: AtomicUsize::new(band.max),
            windows: Mutex::new([ClassWindow::EMPTY; OpClass::COUNT]),
            shrinks: AtomicU64::new(0),
            grows: AtomicU64::new(0),
        }
    }

    /// The band this controller is clamped to.
    pub fn band(&self) -> LimitBand {
        self.band
    }

    /// The shared limit right now.
    pub fn limit(&self) -> usize {
        self.limit.load(Ordering::Relaxed)
    }

    /// How many times a standing queue has shrunk the limit.
    pub fn shrinks(&self) -> u64 {
        self.shrinks.load(Ordering::Relaxed)
    }

    /// How many times a draining queue has grown it.
    pub fn grows(&self) -> u64 {
        self.grows.load(Ordering::Relaxed)
    }

    /// Record one item's sojourn, and adjust the shared limit if this closed
    /// `class`'s window.
    pub fn record(&self, class: OpClass, sojourn: Duration) {
        self.record_at(class, sojourn, Instant::now());
    }

    /// [`Codel::record`] with the clock supplied.
    ///
    /// The seam a test drives: `Instant` cannot be constructed, but it can be
    /// offset, so a test captures one `Instant::now()` and hands this window
    /// boundaries to the microsecond instead of sleeping for them.
    pub fn record_at(&self, class: OpClass, sojourn: Duration, now: Instant) {
        let mut windows = self.windows.lock().unwrap_or_else(|e| e.into_inner());
        let window = &mut windows[class.index()];

        let opened = *window.opened.get_or_insert(now);
        window.min = Some(match window.min {
            Some(previous) => previous.min(sojourn),
            None => sojourn,
        });

        if now.saturating_duration_since(opened) < INTERVAL {
            return;
        }

        // The window is closed. Its MINIMUM decides, against THIS class's
        // target — the head-of-line guard, in one line.
        let standing = window.min.is_some_and(|min| min > class.target());
        *window = ClassWindow {
            opened: Some(now),
            min: None,
        };
        drop(windows);

        if standing {
            self.shrink();
        } else {
            self.grow();
        }
    }

    /// Multiplicative decrease: `limit -= max(limit/8, 1)`.
    ///
    /// The `max(_, 1)` is load-bearing. `limit/8` is integer division, so at a
    /// limit of 7 the step would be 0 and a standing queue could never bring the
    /// limit down — the controller would be dead in exactly the state it exists
    /// for.
    fn shrink(&self) {
        let band = self.band;
        let _ = self
            .limit
            .fetch_update(Ordering::Relaxed, Ordering::Relaxed, |limit| {
                Some(band.clamp(limit.saturating_sub((limit / 8).max(1))))
            });
        self.shrinks.fetch_add(1, Ordering::Relaxed);
    }

    /// Additive increase: `limit += 1`.
    fn grow(&self) {
        let band = self.band;
        let _ = self
            .limit
            .fetch_update(Ordering::Relaxed, Ordering::Relaxed, |limit| {
                Some(band.clamp(limit.saturating_add(1)))
            });
        self.grows.fetch_add(1, Ordering::Relaxed);
    }
}

/// Gate 2: a budget and a counter. Never a probe.
#[derive(Debug)]
pub struct MemoryLedger {
    budget: u64,
    reserved: AtomicU64,
}

impl MemoryLedger {
    /// A ledger with `budget` bytes to give out.
    pub fn new(budget: u64) -> MemoryLedger {
        MemoryLedger {
            budget,
            reserved: AtomicU64::new(0),
        }
    }

    /// The budget.
    pub fn budget(&self) -> u64 {
        self.budget
    }

    /// Bytes currently reserved by operations in flight.
    pub fn reserved(&self) -> u64 {
        self.reserved.load(Ordering::Relaxed)
    }

    /// Reserve `bytes`, or refuse.
    ///
    /// A compare-and-swap loop rather than "load, check, add": between a load
    /// and an add, two concurrent admissions both see room and both take it, and
    /// a budget that can be exceeded by the number of racing threads is not a
    /// budget. Nothing is allocated here and nothing is allocated by the caller
    /// before here — the refusal arrives before the cost.
    pub fn try_reserve(&self, bytes: u64) -> bool {
        self.reserved
            .fetch_update(Ordering::Relaxed, Ordering::Relaxed, |reserved| {
                let wanted = reserved.saturating_add(bytes);
                (wanted <= self.budget).then_some(wanted)
            })
            .is_ok()
    }

    /// Charge `bytes` **whether or not they fit**, and say which it was.
    ///
    /// The reserve path for an operation that is already running: see
    /// [`Growth`] for why this cannot refuse. The counter is allowed to exceed
    /// the budget, and while it does, [`MemoryLedger::try_reserve`] refuses
    /// every new admission — which is the whole mechanism by which one enormous
    /// transfer quiets the box instead of taking it down.
    pub fn charge(&self, bytes: u64) -> bool {
        let before = self.reserved.fetch_add(bytes, Ordering::Relaxed);
        before.saturating_add(bytes) <= self.budget
    }

    /// Give `bytes` back.
    ///
    /// Saturating, because a ledger that can go negative under a double release
    /// would hand out the whole budget twice.
    pub fn release(&self, bytes: u64) {
        let _ = self
            .reserved
            .fetch_update(Ordering::Relaxed, Ordering::Relaxed, |reserved| {
                Some(reserved.saturating_sub(bytes))
            });
    }
}

/// Why the governor shed an operation.
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub enum Shed {
    /// Gate 1: the governed concurrency limit is full.
    Concurrency {
        /// The limit in force at the moment of the refusal — the *governed*
        /// one, which is not the configured one and moves on its own. Reported
        /// rather than assumed, or an operator reads a number in a log that
        /// matches nothing in a config file.
        limit: usize,
        /// How many operations were in flight.
        in_flight: usize,
    },
    /// Gate 1b: this operation's **lane** is full, though the server as a
    /// whole is not. The other lanes' floors are what is holding the slots; see
    /// [`Lane::floor`].
    LaneFull {
        /// The lane that was full.
        lane: Lane,
        /// What that lane may hold under the governed limit right now.
        ceiling: usize,
        /// The governed server-wide limit.
        limit: usize,
    },
    /// Gate 1c: the claim asked for a slot the caller had told this governor to
    /// **hold back** ([`Governor::claim_reserving`]), though the server as a
    /// whole is not full.
    ///
    /// Its own variant rather than [`Shed::Concurrency`] with a smaller `limit`
    /// in it, because reporting a reduced limit as *the* limit is the exact
    /// failure the unbounded listener is about: an operator would read `limit: 2` out of a log
    /// on a box configured for 256 and conclude the gate was broken. Both
    /// numbers travel, and the caller that supplied the reserve is the one that
    /// gets to name the rule.
    Reserved {
        /// Operations in flight when the claim failed.
        occupied: usize,
        /// The **governed** server-wide limit — the real one, unreduced.
        limit: usize,
        /// How many slots of it the caller asked to be held back from this
        /// claim.
        reserve: usize,
    },
    /// Gate 2: the memory budget is spoken for.
    Memory {
        /// The class whose estimate did not fit.
        class: OpClass,
        /// What [`OpClass::estimated_bytes`] asked for.
        wanted: u64,
        /// What was already reserved.
        reserved: u64,
        /// The budget.
        budget: u64,
    },
}

/// What an admitted operation holds: a concurrency slot and a memory
/// reservation, plus the instant it was admitted.
///
/// Deliberately **not** a `Drop` guard. The one guard is
/// [`AdmissionGuard`](crate::AdmissionGuard), which owns this; two `Drop`s over
/// one reservation is how a ledger comes to be released twice.
#[derive(Debug, Clone, Copy)]
pub struct Claim {
    /// The class, needed to release the right reservation and to file the
    /// sojourn under the right window.
    pub class: OpClass,
    /// What was reserved **at admission** ([`OpClass::admission_bytes`]).
    /// Everything charged afterwards is counted by the guard that owns this
    /// claim, so the release is `bytes + grown` and cannot drift.
    pub bytes: u64,
    /// Enqueue time — when the operation first asked for a slot, before any
    /// wait. The sojourn is measured from here.
    pub enqueued: Instant,
}

/// What a top-up of a claim's reservation did.
///
/// It is deliberately not a `Result`: an operation that is already running
/// cannot be refused, and pretending otherwise would put a failure on a path
/// with nothing to do about it. What an overdraw does instead is close the door
/// behind it — `reserved > budget` makes [`MemoryLedger::try_reserve`] refuse
/// every *new* admission until this transfer finishes, which is the graceful
/// degradation the budget exists for.
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub enum Growth {
    /// The charge fitted inside the budget.
    Fitted {
        /// What this transfer now holds.
        held: u64,
    },
    /// The charge did not fit and was taken anyway. New admissions are refused
    /// until it is given back.
    Overdrawn {
        /// What this transfer now holds.
        held: u64,
        /// What the whole server now holds.
        reserved: u64,
        /// The budget it is over.
        budget: u64,
    },
}

/// Both gates, and the occupancy they meter.
#[derive(Debug)]
pub struct Governor {
    codel: Codel,
    memory: MemoryLedger,
    /// Operations in flight. This *is* the queue occupancy — incremented inside
    /// the same compare-and-swap that checks the limit, so there is no window in
    /// which two admissions both see room.
    occupied: AtomicUsize,
    /// The same count, split by [`Lane`], so a lane floor can be enforced.
    /// Fixed-size and preallocated: no map, no allocation on the decision path.
    per_lane: [AtomicUsize; Lane::COUNT],
    /// How many operations of each lane are **parked** in
    /// [`crate::Admission::admit_within`] right now.
    ///
    /// This is the demand signal, and it is the whole reason lane floors are
    /// work-conserving rather than a second wall: see
    /// [`Governor::lane_ceiling`].
    waiting: [AtomicUsize; Lane::COUNT],
    shed_concurrency: AtomicU64,
    shed_lane: AtomicU64,
    shed_memory: AtomicU64,
    overdrawn: AtomicU64,
}

impl Governor {
    /// A governor over `band` with `budget` bytes of memory to give out.
    pub fn new(band: LimitBand, budget: u64) -> Governor {
        Governor {
            codel: Codel::new(band),
            memory: MemoryLedger::new(budget),
            occupied: AtomicUsize::new(0),
            per_lane: [const { AtomicUsize::new(0) }; Lane::COUNT],
            waiting: [const { AtomicUsize::new(0) }; Lane::COUNT],
            shed_concurrency: AtomicU64::new(0),
            shed_lane: AtomicU64::new(0),
            shed_memory: AtomicU64::new(0),
            overdrawn: AtomicU64::new(0),
        }
    }

    /// The delay controller.
    pub fn codel(&self) -> &Codel {
        &self.codel
    }

    /// The memory ledger.
    pub fn memory(&self) -> &MemoryLedger {
        &self.memory
    }

    /// The governed limit right now.
    pub fn limit(&self) -> usize {
        self.codel.limit()
    }

    /// Operations in flight right now.
    pub fn occupied(&self) -> usize {
        self.occupied.load(Ordering::Relaxed)
    }

    /// Operations in flight in one lane.
    pub fn occupied_in(&self, lane: Lane) -> usize {
        self.per_lane[lane.index()].load(Ordering::Relaxed)
    }

    /// **How many slots `lane` may hold right now**: the governed limit, less
    /// the part of every *other* lane's floor that somebody is actually
    /// **waiting** for.
    ///
    /// # A floor that reserves an idle lane's slots is the old bug again
    ///
    /// The per-principal cap used to bind unconditionally, and measured on the
    /// 32-core desktop that meant 32 served and 32 refused against
    /// **224 idle slots**
    /// (`crate::admission`). A lane floor that reserved an eighth of the box for
    /// `ls-remote` whether or not any `ls-remote` existed would be the same
    /// mistake with a different noun: it would refuse a push the box could
    /// serve, to keep a slot warm for a request nobody had made.
    ///
    /// So the floor is claimed **on demand**, and demand is now observable:
    /// since 2026-08-30 an operation over the limit *parks*
    /// ([`crate::Admission::admit_within`]) instead of being refused, so the
    /// number of parked operations per lane is a live, exact count of who is
    /// asking. A lane with nobody waiting reserves nothing; the instant one
    /// `ls-remote` parks, one slot of the cheap floor is held back and the next
    /// release goes to it rather than to the push storm that would otherwise
    /// take every slot for as long as it lasted.
    ///
    /// `min(floor, waiting)` is what makes it a floor and not a queue: a lane
    /// can never reserve more than [`Lane::floor`] however many of its
    /// operations are waiting, so `cheap` cannot starve `write` either.
    pub fn lane_ceiling(&self, lane: Lane, limit: usize) -> usize {
        let reserved_elsewhere: usize = Lane::ALL
            .iter()
            .filter(|other| **other != lane)
            .map(|other| other.floor(limit).min(self.waiting_in(*other)))
            .sum();
        limit.saturating_sub(reserved_elsewhere).max(1)
    }

    /// Operations of `lane` parked waiting for a slot right now.
    pub fn waiting_in(&self, lane: Lane) -> usize {
        self.waiting[lane.index()].load(Ordering::Relaxed)
    }

    /// Say that one operation of `lane` has started waiting.
    ///
    /// Paired with [`Governor::left_wait`] by
    /// [`crate::Admission::admit_within`], which is the only caller and which
    /// pairs them on every path including cancellation.
    pub fn entered_wait(&self, lane: Lane) {
        self.waiting[lane.index()].fetch_add(1, Ordering::Relaxed);
    }

    /// Say that one operation of `lane` has stopped waiting.
    pub fn left_wait(&self, lane: Lane) {
        let _ = self.waiting[lane.index()].fetch_update(
            Ordering::Relaxed,
            Ordering::Relaxed,
            |waiting| Some(waiting.saturating_sub(1)),
        );
    }

    /// Refusals by gate: (concurrency, lane, memory).
    pub fn shed_counts(&self) -> (u64, u64, u64) {
        (
            self.shed_concurrency.load(Ordering::Relaxed),
            self.shed_lane.load(Ordering::Relaxed),
            self.shed_memory.load(Ordering::Relaxed),
        )
    }

    /// How many transfers have charged the ledger past its budget while
    /// running ([`Growth::Overdrawn`]).
    pub fn overdrawn(&self) -> u64 {
        self.overdrawn.load(Ordering::Relaxed)
    }

    /// **The admission decision.** Time first, then the lane, then space, then
    /// take all three.
    ///
    /// Nothing is allocated on any refusal path, and nothing the operation
    /// needs is allocated on the success path either: a [`Claim`] is three
    /// `Copy` words. That is the point of the gate — the refusal has to arrive
    /// before the cost, or it is only a slower way of paying it.
    pub fn try_claim(&self, class: OpClass) -> Result<Claim, Shed> {
        self.claim(class, 0, true)
    }

    /// [`Governor::try_claim`] or [`Governor::attempt`], chosen by the caller.
    pub fn claim_for(&self, class: OpClass, count: bool) -> Result<Claim, Shed> {
        self.claim(class, 0, count)
    }

    /// [`Governor::claim_for`] against a limit `reserve` slots **lower** than
    /// the governed one, so that this claim can never take one of the last
    /// `reserve` slots.
    ///
    /// # Why the reserve is enforced here and not by the caller
    ///
    /// A caller can read [`Governor::occupied`], compare it to a threshold and
    /// only then ask for a claim — and that is a race, for exactly the reason
    /// written over the compare-and-swap below: `N` racing admissions all read
    /// the same occupancy and all decide they fit. Measured, on a four-slot gate
    /// with two hundred anonymous readers: strangers holding *transient* claims
    /// while they made that decision filled the server-wide gate, and a keyed
    /// caller arriving into that window was refused by gate 1 with
    /// `AtCapacity` — the reserve that existed to protect it having never been
    /// consulted. A limit that is checked outside the atomic that enforces it is
    /// not a limit.
    ///
    /// So the reserve rides in the same `fetch_update`. `N` racing strangers can
    /// then no more collectively exceed `limit - reserve` than they can exceed
    /// `limit`, and a refused one never holds a slot at all — which matters
    /// twice, because the refusal path of a flood is the path an attacker is
    /// trying to make expensive.
    ///
    /// # Why this moves no shed counter
    ///
    /// The three [`Governor::shed_counts`] counters count *this* governor's own
    /// gates — the limit it controls, the lane floors it computes, the ledger it
    /// keeps. The reserve is none of those: it is a number the caller handed in,
    /// for a policy the caller owns, and the caller counts it
    /// (`AdmissionStats::refused_keyed_reserve`). Counting it here as a
    /// concurrency shed would tell an operator the CoDel limit was full when it
    /// was nowhere near, and counting it here under a fourth name would give the
    /// same event two owners.
    pub fn claim_reserving(
        &self,
        class: OpClass,
        reserve: usize,
        count: bool,
    ) -> Result<Claim, Shed> {
        self.claim(class, reserve, count)
    }

    /// [`Governor::try_claim`] without moving the shed counters.
    ///
    /// The retry inside [`crate::Admission::admit_within`] asks this door many
    /// times for **one** operation. Counting every ask would make
    /// `shed_concurrency` a count of *polls* rather than of refusals — the
    /// counter would read 21 where one client was refused once, which is worse
    /// than no counter. So the wait loop attempts without counting and counts
    /// once, at the end, when the refusal is final.
    pub fn attempt(&self, class: OpClass) -> Result<Claim, Shed> {
        self.claim(class, 0, false)
    }

    fn claim(&self, class: OpClass, reserve: usize, count: bool) -> Result<Claim, Shed> {
        let limit = self.codel.limit();
        // What THIS claim may take. `reserve` is 0 for every ordinary caller,
        // in which case the two are the same number and nothing below changes.
        // See [`Governor::claim_reserving`] for why the reserve is applied here
        // rather than checked by the caller beforehand.
        let available = limit.saturating_sub(reserve);
        // Check and take in ONE compare-and-swap. "Load, compare, then add"
        // lets N racing admissions all read `occupied == limit - 1`.
        let taken = self
            .occupied
            .fetch_update(Ordering::Relaxed, Ordering::Relaxed, |occupied| {
                (occupied < available).then_some(occupied + 1)
            });
        if let Err(occupied) = taken {
            if reserve > 0 {
                // The caller's rule refused this, not the governor's, so the
                // governor's counters do not move and the governor's variant is
                // not used — even when the box happens to ALSO be full. Reported
                // this way round on purpose: a caller that supplied a reserve
                // has a policy reason for it, and a refusal that flipped between
                // two reasons depending on how full the box happened to be would
                // be unreadable in a log and, in `Admission`'s case, would decide
                // whether the caller gets parked.
                return Err(Shed::Reserved {
                    occupied,
                    limit,
                    reserve,
                });
            }
            if count {
                self.shed_concurrency.fetch_add(1, Ordering::Relaxed);
            }
            return Err(Shed::Concurrency {
                limit,
                in_flight: occupied,
            });
        }

        // The lane, second: the server-wide limit is the resource that actually
        // runs out, and a lane floor is only interesting once it does.
        let lane = class.lane();
        let ceiling = self.lane_ceiling(lane, limit);
        let lane_taken = self.per_lane[lane.index()].fetch_update(
            Ordering::Relaxed,
            Ordering::Relaxed,
            |held| (held < ceiling).then_some(held + 1),
        );
        if lane_taken.is_err() {
            self.occupied.fetch_sub(1, Ordering::Relaxed);
            if count {
                self.shed_lane.fetch_add(1, Ordering::Relaxed);
            }
            return Err(Shed::LaneFull {
                lane,
                ceiling,
                limit,
            });
        }

        let bytes = class.admission_bytes();
        if !self.memory.try_reserve(bytes) {
            // Hand the concurrency slot AND the lane slot straight back: a slot
            // leaked on the memory path is a server that refuses everything an
            // hour after start-up.
            self.per_lane[lane.index()].fetch_sub(1, Ordering::Relaxed);
            self.occupied.fetch_sub(1, Ordering::Relaxed);
            if count {
                self.shed_memory.fetch_add(1, Ordering::Relaxed);
            }
            return Err(Shed::Memory {
                class,
                wanted: bytes,
                reserved: self.memory.reserved(),
                budget: self.memory.budget(),
            });
        }

        Ok(Claim {
            class,
            bytes,
            enqueued: Instant::now(),
        })
    }

    /// **Charge a running transfer for what has actually crossed the wire.**
    ///
    /// Called by [`crate::AdmissionGuard::reserve_more`] and by nothing else,
    /// so the guard stays the single owner of "what this operation holds" and
    /// the release cannot drift. See [`Growth`] for why it cannot refuse and
    /// [`WIRE_MEMORY_NUMERATOR`] for what fraction of the wire is charged.
    pub fn charge(&self, held_after: u64, extra: u64) -> Growth {
        if self.memory.charge(extra) {
            Growth::Fitted { held: held_after }
        } else {
            self.overdrawn.fetch_add(1, Ordering::Relaxed);
            Growth::Overdrawn {
                held: held_after,
                reserved: self.memory.reserved(),
                budget: self.memory.budget(),
            }
        }
    }

    /// Give a claim's slot and reservation back. Called from exactly one place:
    /// [`AdmissionGuard`](crate::AdmissionGuard)'s `Drop`, which passes what it
    /// grew the reservation to on top of [`Claim::bytes`].
    pub fn release(&self, claim: &Claim, grown: u64) {
        self.memory.release(claim.bytes.saturating_add(grown));
        let lane = claim.class.lane();
        let _ = self.per_lane[lane.index()].fetch_update(
            Ordering::Relaxed,
            Ordering::Relaxed,
            |held| Some(held.saturating_sub(1)),
        );
        let _ = self
            .occupied
            .fetch_update(Ordering::Relaxed, Ordering::Relaxed, |occupied| {
                Some(occupied.saturating_sub(1))
            });
    }

    /// File one **admission wait** against its class's window.
    ///
    /// This is the controller's input, and the change of input is the point:
    /// see [`LimitBand::for_budget`] for what it used to be (the runtime's
    /// spawn-to-run gap) and why a controller that cannot move its own input
    /// runs to its floor. A wait of zero — the ordinary case on an unloaded
    /// server — is a sample too, and it is the one that grows the limit back.
    pub fn record_sojourn(&self, class: OpClass, sojourn: Duration) {
        self.codel.record(class, sojourn);
    }
}

/// A one-shot latch for "this operation's work has actually started".
///
/// Its own type so the "was the sojourn ever measured?" question has one
/// answer-holder rather than a bool copied into two structs.
#[derive(Debug, Default)]
pub struct ServiceMark(AtomicBool);

impl ServiceMark {
    /// Claim the mark. `true` the first time, `false` for every later call, so a
    /// caller that marks twice files one sample rather than two.
    pub fn claim(&self) -> bool {
        !self.0.swap(true, Ordering::Relaxed)
    }

    /// Whether the mark was ever claimed.
    pub fn claimed(&self) -> bool {
        self.0.load(Ordering::Relaxed)
    }
}

#[cfg(test)]
mod tests {
    use super::*;

    /// Drive `class` through exactly one closed window at `sojourn`.
    ///
    /// Two samples: one to open the window, one at `INTERVAL` later to close it.
    /// Returns the instant the window closed, so a caller can chain windows.
    fn one_window(codel: &Codel, class: OpClass, sojourn: Duration, from: Instant) -> Instant {
        codel.record_at(class, sojourn, from);
        let closed = from + INTERVAL;
        codel.record_at(class, sojourn, closed);
        closed
    }

    /// **A standing queue shrinks the limit.**
    ///
    /// # Seen red
    ///
    /// By changing `Codel::record_at`'s decision to
    /// `let standing = window.min.is_some_and(|min| min > INTERVAL);` — i.e.
    /// comparing the window minimum against the window length instead of the
    /// class's target, the natural "off by one constant" of this controller.
    /// Failure:
    ///
    /// ```text
    /// a queue standing at 50ms against a 5ms target did not shrink the limit:
    /// it is still 100 after 8 closed windows (shrinks=0)
    /// ```
    #[test]
    fn a_standing_queue_shrinks_the_limit() {
        let codel = Codel::new(LimitBand::DEFAULT);
        assert_eq!(
            codel.limit(),
            MAX_LIMIT,
            "a governor starts at the top of its band"
        );

        // Ten times the ls-remote target, for eight windows.
        let mut at = Instant::now();
        for _ in 0..8 {
            at = one_window(&codel, OpClass::LsRemote, Duration::from_millis(50), at);
        }

        assert!(
            codel.limit() < MAX_LIMIT,
            "a queue standing at 50ms against a {:?} target did not shrink the limit: it is \
             still {} after 8 closed windows (shrinks={})",
            OpClass::LsRemote.target(),
            codel.limit(),
            codel.shrinks()
        );
        assert_eq!(
            codel.shrinks(),
            8,
            "every closed standing window must shrink"
        );
        // 100 → 88 → 78 → 69 → 61 → 54 → 48 → 42 → 37: multiplicative, not -8.
        assert_eq!(
            codel.limit(),
            37,
            "the decrease must be MULTIPLICATIVE (limit -= limit/8); an additive one would \
             leave a different number here and would not back off fast enough to matter"
        );
        assert_eq!(codel.grows(), 0);
    }

    /// **A draining queue grows the limit.**
    ///
    /// # Seen red
    ///
    /// By making `Codel::grow` a no-op (`fn grow(&self) { }`). Failure:
    ///
    /// ```text
    /// a queue draining at 1ms did not grow the limit: 10 after 20 closed
    /// windows, and a controller that only shrinks is a ratchet
    /// ```
    #[test]
    fn a_draining_queue_grows_the_limit() {
        let codel = Codel::new(LimitBand::DEFAULT);
        // Start from the floor so growth is visible: shrink it there first.
        let mut at = Instant::now();
        for _ in 0..40 {
            at = one_window(&codel, OpClass::LsRemote, Duration::from_millis(50), at);
        }
        assert_eq!(
            codel.limit(),
            MIN_LIMIT,
            "the standing phase must reach the floor"
        );

        for _ in 0..20 {
            at = one_window(&codel, OpClass::LsRemote, Duration::from_millis(1), at);
        }
        assert!(
            codel.limit() > MIN_LIMIT,
            "a queue draining at 1ms did not grow the limit: {} after 20 closed windows, and a \
             controller that only shrinks is a ratchet",
            codel.limit()
        );
        assert_eq!(
            codel.limit(),
            MIN_LIMIT + 20,
            "growth must be ADDITIVE, one per closed draining window"
        );
    }

    /// **The limit never leaves its band**, in either direction, however long
    /// the load runs.
    ///
    /// # Seen red
    ///
    /// By dropping the clamp from `Codel::shrink`
    /// (`Some(limit.saturating_sub((limit / 8).max(1)))`). Failure:
    ///
    /// ```text
    /// the limit left the band: 9 is outside [10, 100]
    /// ```
    ///
    /// It fails on the FIRST step past the floor, not after drifting to zero,
    /// which is what makes the assertion inside the loop worth having: a check
    /// only at the end would report `0` and hide when it started.
    #[test]
    fn the_limit_never_leaves_the_band() {
        let codel = Codel::new(LimitBand::DEFAULT);
        let mut at = Instant::now();

        // Far past the floor.
        for _ in 0..200 {
            at = one_window(&codel, OpClass::LsRemote, Duration::from_secs(1), at);
            let limit = codel.limit();
            assert!(
                (MIN_LIMIT..=MAX_LIMIT).contains(&limit),
                "the limit left the band: {limit} is outside [{MIN_LIMIT}, {MAX_LIMIT}]"
            );
        }
        assert_eq!(codel.limit(), MIN_LIMIT);

        // …and far past the ceiling.
        for _ in 0..500 {
            at = one_window(&codel, OpClass::LsRemote, Duration::from_micros(1), at);
            let limit = codel.limit();
            assert!(
                (MIN_LIMIT..=MAX_LIMIT).contains(&limit),
                "the limit left the band: {limit} is outside [{MIN_LIMIT}, {MAX_LIMIT}]"
            );
        }
        assert_eq!(codel.limit(), MAX_LIMIT);
    }

    /// **THE HEAD-OF-LINE TEST.** A slow class does not shrink the limit on a
    /// fast class's behalf.
    ///
    /// A `linux.git` clone waiting 150 ms is a clone that has not noticed. An
    /// `ls-remote` waiting 150 ms has had its latency multiplied by fifteen.
    /// One number cannot say both, so the target is per class — and this test is
    /// the difference between the two designs.
    ///
    /// # Seen red
    ///
    /// By replacing the per-class bar in `Codel::record_at` with the single
    /// shared one the naive design has:
    /// `let standing = window.min.is_some_and(|min| min > TARGET);`
    /// Failure:
    ///
    /// ```text
    /// a clone waiting 150ms — well inside its own 186ms target, and a wait its
    /// client cannot perceive — shrank the shared limit to 42. That is the
    /// slow class poisoning the fast class's delay signal, which is the exact
    /// trap per-class sojourn exists to close (shrinks=8)
    /// ```
    #[test]
    fn a_slow_class_does_not_shrink_the_limit_on_a_fast_class_behalf() {
        let codel = Codel::new(LimitBand::DEFAULT);
        let mut at = Instant::now();

        // Eight windows of clones waiting 150 ms — thirty times the ls-remote
        // target, and comfortably INSIDE clone's own 186 ms target.
        assert!(Duration::from_millis(150) > OpClass::LsRemote.target());
        assert!(Duration::from_millis(150) < OpClass::Clone.target());
        for _ in 0..8 {
            at = one_window(&codel, OpClass::Clone, Duration::from_millis(150), at);
        }
        assert_eq!(
            codel.limit(),
            MAX_LIMIT,
            "a clone waiting 150ms — well inside its own {:?} target, and a wait its client \
             cannot perceive — shrank the shared limit to {}. That is the slow class poisoning \
             the fast class's delay signal, which is the exact trap per-class sojourn exists to \
             close (shrinks={})",
            OpClass::Clone.target(),
            codel.limit(),
            codel.shrinks()
        );
        assert_eq!(codel.shrinks(), 0);

        // The other half of the bar, or the test above would also pass on a
        // controller that had simply stopped shrinking: the SAME 150 ms, filed
        // under ls-remote, must shrink hard.
        for _ in 0..8 {
            at = one_window(&codel, OpClass::LsRemote, Duration::from_millis(150), at);
        }
        assert!(
            codel.limit() < MAX_LIMIT,
            "150ms of ls-remote sojourn is thirty times its {:?} target and MUST shed load; a \
             controller that shrinks for nobody is not a controller",
            OpClass::LsRemote.target()
        );
        assert_eq!(codel.shrinks(), 8);

        // …and the clone class is still healthy while the shared limit came
        // down. One queue, one limit, two signals.
        let before = codel.limit();
        let shrinks_before = codel.shrinks();
        for _ in 0..4 {
            at = one_window(&codel, OpClass::Clone, Duration::from_millis(150), at);
        }
        assert!(
            codel.limit() > before,
            "the clone class must still read as DRAINING and grow the shared limit back from \
             {before}, even though its absolute sojourn is thirty times the ls-remote target; \
             it is {} instead",
            codel.limit()
        );
        assert_eq!(
            codel.shrinks(),
            shrinks_before,
            "not one of those clone windows may be read as standing"
        );
    }

    /// Each class keeps its **own** window: a sample filed under one class may
    /// not open, close or lower another's.
    ///
    /// # Seen red
    ///
    /// By indexing every class into slot 0 in `Codel::record_at`
    /// (`&mut windows[0]`). Failure:
    ///
    /// ```text
    /// assertion `left == right` failed: push's own windows must close and grow
    ///   left: 10
    ///  right: 9
    /// ```
    ///
    /// The tenth growth is the ls-remote sample's own window being closed by a
    /// push, which is exactly the crossing this test forbids — one class's
    /// sample opening and closing another's window.
    #[test]
    fn a_sample_lands_in_its_own_class_window_only() {
        let codel = Codel::new(LimitBand::DEFAULT);
        let at = Instant::now();

        // Open ls-remote's window, then push a full INTERVAL of PUSH samples
        // through. Nothing may close ls-remote's window but an ls-remote sample.
        codel.record_at(OpClass::LsRemote, Duration::from_millis(50), at);
        for step in 0..10 {
            codel.record_at(
                OpClass::Push,
                Duration::from_millis(1),
                at + INTERVAL * (step + 1),
            );
        }
        // Push closed its own windows and they were healthy, so it GREW.
        assert_eq!(codel.grows(), 9, "push's own windows must close and grow");
        assert_eq!(
            codel.shrinks(),
            0,
            "a push sample closed the ls-remote window: shrinks={} grows={} after one ls-remote \
             sample that could not have closed anything",
            codel.shrinks(),
            codel.grows()
        );
    }

    /// **Reserving up to the budget admits; the next operation is denied.**
    ///
    /// # Seen red
    ///
    /// By making `MemoryLedger::try_reserve` compare `reserved` rather than
    /// `reserved + bytes` against the budget — the classic off-by-one-operation
    /// budget check, which lets the last admission overshoot by a whole
    /// estimate. Failure:
    ///
    /// ```text
    /// the ledger admitted 5 clones against a 4-clone budget: 10737418240
    /// reserved of 8589934592
    /// ```
    #[test]
    fn the_budget_admits_up_to_itself_and_then_denies() {
        let clone = OpClass::Clone.admission_bytes();
        let ledger = MemoryLedger::new(clone * 4);

        let mut admitted = 0;
        while ledger.try_reserve(clone) {
            admitted += 1;
            assert!(
                admitted <= 4,
                "the ledger admitted {admitted} clones against a 4-clone budget: {} reserved of {}",
                ledger.reserved(),
                ledger.budget()
            );
        }
        assert_eq!(admitted, 4, "a 4-clone budget must admit exactly 4 clones");
        assert_eq!(ledger.reserved(), ledger.budget());

        // …and a SMALLER operation still fits nowhere, which is the property a
        // "there is some room" check would lose.
        assert!(
            !ledger.try_reserve(1),
            "a full ledger handed out one more byte"
        );
    }

    /// **A completed operation releases its reservation and the next one is
    /// then admitted.**
    ///
    /// # Seen red
    ///
    /// By making `MemoryLedger::release` a no-op. Failure:
    ///
    /// ```text
    /// the 4th clone finished and gave its 2147483648 bytes back, but the next
    /// clone was still refused: 8589934592 reserved of 8589934592. A budget
    /// that is never released is a server that wedges after N operations
    /// ```
    #[test]
    fn a_completed_operation_releases_its_reservation() {
        let clone = OpClass::Clone.admission_bytes();
        let ledger = MemoryLedger::new(clone * 4);
        for _ in 0..4 {
            assert!(ledger.try_reserve(clone));
        }
        assert!(!ledger.try_reserve(clone), "the budget is full");

        ledger.release(clone);
        assert!(
            ledger.try_reserve(clone),
            "the 4th clone finished and gave its {clone} bytes back, but the next clone was \
             still refused: {} reserved of {}. A budget that is never released is a server that \
             wedges after N operations",
            ledger.reserved(),
            ledger.budget()
        );
    }

    /// **The budget is never exceeded, however hard it is raced.**
    ///
    /// # This test was HOLLOW twice, and both failures are worth keeping
    ///
    /// 1. First written as 64 `thread::spawn`s in a loop, it **passed against a
    ///    deliberately broken `try_reserve`**: spawning a thread costs orders of
    ///    magnitude more than the three instructions being raced, so the racers
    ///    arrived one at a time. A concurrency test whose threads never overlap
    ///    is a test of `thread::spawn`.
    /// 2. Adding a `Barrier` was not enough either — five runs, still green.
    ///    One release of 64 threads gives the bug exactly one chance to happen
    ///    inside a window a few nanoseconds wide.
    ///
    /// What makes it fail is **many chances at the boundary**: a budget of four
    /// slabs, every thread cycling reserve/release, so the "is there room for
    /// one more" edge is crossed millions of times instead of once. The
    /// invariant is checked by the racers themselves, not after the fact.
    ///
    /// # Seen red
    ///
    /// By rewriting `try_reserve` as load / compare / `fetch_add`, which is what
    /// it looks like it could be. On the reference desktop, 2026-08-14:
    ///
    /// ```text
    /// 32 racers cycling 20000 times each drove the ledger to 23068672 bytes
    /// against a budget of 4194304: it was exceeded by 18874368 bytes, which
    /// is a budget that the number of concurrent admissions can walk straight
    /// past
    /// ```
    ///
    /// Five and a half times the budget. The naive version does not overshoot
    /// by one slab, it overshoots by however many threads are in the window.
    #[test]
    fn the_budget_is_not_exceeded_by_racing_admissions() {
        use std::sync::{Arc, Barrier};

        let slab = 1024 * 1024u64;
        let slabs = 4u64;
        let racers = 32;
        let rounds = 20_000;
        let budget = slab * slabs;
        let ledger = Arc::new(MemoryLedger::new(budget));
        // The high-water mark any racer ever SAW, which is the applied output:
        // a budget is not a budget if a reader can observe it exceeded.
        let peak = Arc::new(AtomicU64::new(0));
        let gate = Arc::new(Barrier::new(racers));

        // Plain std threads: these are one-shot racers, not data-parallel work,
        // so there is nothing for gatling to rotate over and rayon is forbidden
        // outright (the multicore rule: gatling, never rayon).
        let threads: Vec<_> = (0..racers)
            .map(|_| {
                let ledger = Arc::clone(&ledger);
                let peak = Arc::clone(&peak);
                let gate = Arc::clone(&gate);
                std::thread::spawn(move || {
                    gate.wait();
                    for _ in 0..rounds {
                        if ledger.try_reserve(slab) {
                            peak.fetch_max(ledger.reserved(), Ordering::Relaxed);
                            ledger.release(slab);
                        }
                    }
                })
            })
            .collect();
        for thread in threads {
            thread.join().expect("racer");
        }

        assert_eq!(
            ledger.reserved(),
            0,
            "every reservation was released, so the ledger must be empty"
        );
        assert!(
            peak.load(Ordering::Relaxed) <= budget,
            "{} racers cycling {rounds} times each drove the ledger to {} bytes against a budget \
             of {budget}: it was exceeded by {} bytes, which is a budget that the number of \
             concurrent admissions can walk straight past",
            racers,
            peak.load(Ordering::Relaxed),
            peak.load(Ordering::Relaxed) - budget
        );
        // …and the racers really did reach the boundary, or the assertion above
        // passed on a ledger nobody ever filled.
        assert_eq!(
            peak.load(Ordering::Relaxed),
            budget,
            "the ledger never actually filled ({} of {budget}), so the boundary this test is \
             about was never crossed",
            peak.load(Ordering::Relaxed)
        );
    }

    /// A memory refusal must give the **concurrency** slot straight back.
    ///
    /// # Seen red
    ///
    /// By deleting the `self.occupied.fetch_sub(1, ...)` from the memory arm of
    /// `Governor::try_claim`. Failure:
    ///
    /// ```text
    /// 40 memory refusals leaked 40 concurrency slots: occupied is 44 with 4
    /// operations actually in flight, so the gate now refuses on TIME for a
    /// reason that has nothing to do with time
    /// ```
    #[test]
    fn a_memory_refusal_does_not_leak_a_concurrency_slot() {
        let clone = OpClass::Clone.admission_bytes();
        let governor = Governor::new(LimitBand::DEFAULT, clone * 4);
        let held: Vec<_> = (0..4)
            .map(|i| {
                governor
                    .try_claim(OpClass::Clone)
                    .unwrap_or_else(|shed| panic!("clone {i} of 4 was shed: {shed:?}"))
            })
            .collect();

        for _ in 0..40 {
            let shed = governor
                .try_claim(OpClass::Clone)
                .expect_err("the budget is full");
            assert!(matches!(shed, Shed::Memory { .. }), "{shed:?}");
        }
        assert_eq!(
            governor.occupied(),
            4,
            "40 memory refusals leaked {} concurrency slots: occupied is {} with 4 operations \
             actually in flight, so the gate now refuses on TIME for a reason that has nothing \
             to do with time",
            governor.occupied() - 4,
            governor.occupied()
        );

        for claim in &held {
            governor.release(claim, 0);
        }
        assert_eq!(governor.occupied(), 0);
        assert_eq!(governor.memory().reserved(), 0);
    }

    /// The operator's ceiling is the upper bound of the whole band, floor
    /// included.
    ///
    /// # Seen red
    ///
    /// By writing `LimitBand::for_ceiling` as `{ min: MIN_LIMIT, max: MAX_LIMIT.min(ceiling) }`
    /// — the obvious version. Failure:
    ///
    /// ```text
    /// assertion `left == right` failed: a server told to serve 4 has a band of
    /// [10, 4] and a limit of 4: the governor admits more than the operator
    /// configured
    ///   left: LimitBand { min: 10, max: 4 }
    ///  right: LimitBand { min: 4, max: 4 }
    /// ```
    ///
    /// An inverted band — `min` above `max` — which `usize::clamp` panics on
    /// the moment the controller tries to move. The band has to be well formed
    /// at construction, not corrected later.
    #[test]
    fn a_configured_ceiling_below_the_floor_is_still_the_ceiling() {
        let band = LimitBand::for_ceiling(4);
        assert_eq!(
            band,
            LimitBand { min: 4, max: 4 },
            "a server told to serve 4 has a band of [{}, {}] and a limit of {}: the governor \
             admits more than the operator configured",
            band.min,
            band.max,
            Codel::new(band).limit()
        );
        assert_eq!(Codel::new(band).limit(), 4);

        // ★ A ceiling ABOVE the design default is honoured, not capped.
        //
        // The first form did `MAX_LIMIT.min(ceiling)`, so the shipped default of
        // 256 became a band of [10, 100] and the server quietly served 100.
        // Lowering an operator's explicit number is the same defect as raising
        // it. Seen RED by restoring that `.min()`: left `LimitBand { min: 10,
        // max: 100 }`, right `LimitBand { min: 10, max: 256 }`.
        assert_eq!(
            LimitBand::for_ceiling(256),
            LimitBand { min: 10, max: 256 },
            "a configured ceiling of 256 was lowered to the design's default"
        );
        assert_eq!(LimitBand::for_ceiling(32), LimitBand { min: 10, max: 32 });
        // With no ceiling configured, the design's band still applies.
        assert_eq!(LimitBand::DEFAULT, LimitBand { min: 10, max: 100 });
    }

    /// Every class has a distinct index, or two classes share a window and the
    /// head-of-line guard is silently off.
    #[test]
    fn every_class_has_its_own_window_slot() {
        let mut seen = [false; OpClass::COUNT];
        for class in OpClass::ALL {
            assert!(
                !std::mem::replace(&mut seen[class.index()], true),
                "{class} shares window slot {} with another class",
                class.index()
            );
        }
        assert!(seen.iter().all(|&s| s));
    }

    /// The service mark is one-shot, so a caller that marks twice files one
    /// sojourn sample rather than two.
    #[test]
    fn the_service_mark_is_one_shot() {
        let mark = ServiceMark::default();
        assert!(!mark.claimed());
        assert!(mark.claim(), "the first claim wins");
        assert!(!mark.claim(), "the second claim must not");
        assert!(mark.claimed());
    }
}