crates/gunnar-ssh/src/admission.rs
//! **Admission control for git transfers**: an explicit, countable
//! ceiling in place of the accidental one.
//!
//! # What the accidental ceiling was
//!
//! Every git transfer occupies one tokio **blocking-pool** thread for its whole
//! duration (`gunnar-cli`'s `WireTransport::serve` bridges an async duplex into
//! `gunnar-wire`'s blocking `BufRead`/`Write` through `SyncIoBridge`). Nothing
//! bounded how many of those could exist at once, so the real limit was
//! `tokio`'s default of 512 blocking threads, a number nobody chose. Past it,
//! transfers queued **silently**: no log, no counter, no error. The failure an
//! operator saw was "git is slow", with nothing anywhere to look at.
//!
//! # Why this refuses instead of queueing
//!
//! This gate **refuses** over the limit rather than queueing behind it, and the
//! reasoning is not "refusing is simpler":
//!
//! * **A git client retries; a git client cannot see a queue.** `git` reports
//! whatever the server says and the user (or CI) runs it again. A refusal that
//! arrives in milliseconds with a sentence in it is strictly better
//! information than a socket that goes quiet for a minute.
//! * **Queueing keeps the expensive things alive.** A queued transfer still
//! holds a TCP socket, a russh session, its channel and its key material. That
//! is the resource a fork-bomb of clones is trying to exhaust, and a queue
//! hands it to them for free.
//! * **The two existing timeouts fight a queue.** `inactivity_timeout` (600 s)
//! and `exec_grace` (60 s) both measure a session that is not making progress.
//! A transfer parked in a queue looks exactly like one of those, so a long
//! wait ends in a disconnect anyway, after having occupied the slot for the
//! whole wait.
//! * **A queue makes saturation invisible again.** The whole point of the ceiling is
//! that the ceiling must be *visible*. A refusal is an event with a number
//! attached; a wait is the thing that was already happening.
//!
//! So: [`Admission::try_admit`] never blocks and never waits. It either returns
//! a guard or a [`Refusal`], and both outcomes move a counter.
//!
//! # Two limits, because one is not enough
//!
//! [`SshConfig::max_concurrent_transfers`](crate::SshConfig::max_concurrent_transfers)
//! is the server-wide ceiling. On its own it is still exhaustible by a *single*
//! misbehaving client, which would then deny service to everybody else, so
//! [`SshConfig::max_transfers_per_principal`](crate::SshConfig::max_transfers_per_principal)
//! caps what any one identity can hold at once. Identity is the principal id,
//! not the peer address: gunnar authenticates keys, and a per-IP limit would
//! punish everyone behind one CI runner's NAT while a botnet walked past it.
//!
//! # The per-principal limit is a SHARE, not a wall
//!
//! The second limit used to be an unconditional ceiling, and that was wrong in
//! a way nobody could see from the code. **Measured on a 32-core desktop with
//! the shipped defaults, 2026-08-03:**
//!
//! | load | served | refused |
//! |---|---|---|
//! | 1 principal, 64 clients | 32 | 32 (per-principal) |
//! | 32 principals, 320 clients | 256 | the rest |
//!
//! A client holding **one** key could never reach the advertised 256; that took
//! eight identities. Every single-team appliance — which is what gunnar is —
//! silently got one eighth of the capacity written on the box, and the gate that
//! did it was binding on an **empty server**, where there was nobody to be fair
//! to.
//!
//! The fairness mechanism itself is right: one identity must not be able to take
//! a shared host away from everybody else. So the ceiling is now **max-min
//! fair** — it applies only under contention, and what it applies is a share:
//!
//! ```text
//! contended = occupied (this candidate included) ≥ CONTENTION_PERCENT % of the server-wide limit
//! ceiling(p) = if !contended { the server-wide limit }
//! else { max(configured, server_wide / active principals) }
//! admit = server-wide gate has room AND p holds < ceiling(p)
//! ```
//!
//! * **Idle server, one identity** → not contended, so the only ceiling is the
//! server-wide one. 64 clients on one key now get 64 slots, and 256 clients
//! get 256. That is the half the old gate got wrong.
//! * **Contended server, many identities** → each gets `server_wide / active`,
//! with the configured number as a **floor** nobody drops below however many
//! identities show up. A greedy principal is held to its share exactly as
//! before.
//! * **Work-conserving.** Enforcing the share only above the contention
//! threshold is what stops "two identities, one of them idle" from costing the
//! busy one half the server for no benefit. Below the threshold there are
//! free slots, so refusing anything would refuse work the box could do.
//!
//! `active principals` is the number of identities holding at least one
//! transfer *right now* (the candidate included), which is the only demand
//! signal an admission gate has. It is exact, not sampled: it is the length of
//! the same map the counts live in, read under the same lock.
//!
//! # The ANONYMOUS ceiling: the one caller the share above cannot hold
//!
//! Everything above is keyed on a principal id, and that is deliberate: gunnar
//! authenticates keys, so the fair share divides the server between *identities*
//! and a greedy one is held to its slice. Anonymous read then opened a door with no key
//! behind it. Every unauthenticated reader of a public repository arrives as the
//! single principal [`gunnar_core::ANONYMOUS_PRINCIPAL_ID`] — the id is refused
//! at registration ([`gunnar_core::Principal::new`]) so nothing else can ever
//! claim it — which *looks* like the whole internet already sharing one row of
//! the map above, and therefore already bounded.
//!
//! It is not, and the reason is the paragraph before it: **the share is
//! work-conserving on purpose.** Below [`CONTENTION_PERCENT`] of the governed
//! limit the share does not bind at all and every principal's ceiling is the
//! whole server. That rule was measured and is right for a keyed identity —
//! refusing a lone team its own idle appliance is the exact mistake the 2026-08-03
//! table above records. Applied to a stranger it says: *an idle box may be taken
//! entirely by anonymous traffic.* An operator arriving into that does not even
//! get a clean refusal; they get [`Admission::admit_within`]'s wait, because
//! nothing evicts a transfer already in flight.
//!
//! So anonymous gets a **second, different kind of ceiling**, and the ways it
//! differs from the per-principal share are the whole design:
//!
//! * **Hard, not a share.** It binds on an empty server. There is no
//! contention threshold, because the thing being rationed is not fairness
//! between identities — it is the blast radius of a caller gunnar cannot name,
//! bill, revoke or ring up. Work-conserving is the right instinct for a tenant
//! and the wrong one for a stranger.
//! * **Small in absolute terms**, not a fraction. Five
//! ([`DEFAULT_MAX_ANONYMOUS_TRANSFERS`]) against a keyed floor of 32 and a
//! server-wide 256. Anonymous can therefore never be more than ~2 % of the
//! box, which is what makes "an operator is never held up by an anonymous
//! flood" a *structural* statement rather than a hope about timing.
//! * **Refused instantly, never parked.** [`Admission::admit_within`]'s wait
//! exists because a refused keyed client reconnects, and one parked future is
//! cheaper than that reconnect. For anonymous the arithmetic inverts: parking
//! holds the TCP socket, the russh session and its key material, which is
//! precisely the resource a clone fork-bomb is trying to exhaust. So an
//! `AnonymousAtCapacity` refusal is final at the first ask.
//! * **Counted separately** ([`AdmissionStats::refused_anonymous`]). Folded into
//! `refused_per_principal` it would be indistinguishable from a CI runner
//! hitting its share, and the two demand opposite operator actions: one says
//! *raise the limit*, the other says *you are being scraped*.
//!
//! ## The keyed reserve: the other half of the same door (D23b)
//!
//! The ceiling above is **absolute**, and the box it is a share of is **not**.
//! [`Governor`]'s CoDel controller sheds the governed limit down its band
//! whenever the machine is queueing, and the shed was measured reaching a
//! limit of 19. Five out of 19 is a quarter of the machine, held by a caller
//! that cannot be named — and held at exactly the moment there is least to go
//! round. The `~2 %` in the bullet above is a statement about an *unshed* box.
//! The same arithmetic undoes an operator who raised the ceiling on purpose for
//! a browse replica: the number they raised bears no relation to the number the
//! controller is enforcing an hour later.
//!
//! So the top of the box is **keyed-only**. An anonymous read is admitted only
//! while occupancy is *below* [`CONTENTION_PERCENT`] of the governed limit;
//! from there up it is refused with [`Refusal::KeyedReserve`], instantly, and
//! counted as [`AdmissionStats::refused_keyed_reserve`]. What that reserves is
//! [`Admission::keyed_reserve`] slots — 65 of the shipped 256, 2 of a shed 4 —
//! and a keyed caller can therefore never be refused *by a stranger's
//! acquisition*.
//!
//! Three things about it are decisions rather than defaults:
//!
//! * **The threshold is `CONTENTION_PERCENT`, not a fourth number.** That
//! constant is *defined* as the occupancy at which gunnar starts asking keyed
//! identities to share with each other. The moment it does that is exactly the
//! moment it should stop asking them to share with strangers. A second
//! threshold would be a second opinion about when the box is busy, and the
//! order between the two would be an accident of arithmetic.
//! * **It refuses an acquisition; it does not evict.** Taking a slot back from
//! a transfer already running throws away every byte of pack already built,
//! hands a legitimate visitor a torn stream that `git` reports as a network
//! error (indistinguishable from a broken repository), and does not even
//! reliably give the slot to the keyed caller that triggered it — the
//! `slot_freed` wakeup goes to whoever is parked. It would make a
//! stranger cost *more* than serving it, because the evicted client comes back
//! and redoes the work. Refusing before the first byte costs one
//! compare-and-swap.
//! * **It is enforced inside the claim's compare-and-swap**
//! ([`Governor::claim_reserving`](crate::governor::Governor::claim_reserving)),
//! not checked around it. Both orderings of "read occupancy, then claim" were
//! measured and both leak — the failures are quoted in
//! `tests/keyed_reserve.rs`. A limit checked outside the atomic that enforces
//! it is not a limit.
//!
//! ### What that does and does not promise
//!
//! Stated exactly, because the exact statement is weaker than the slogan and the
//! difference is where somebody would otherwise be surprised. A stranger
//! admitted onto a *quiet* box is still holding its slot if the box fills a
//! moment later, and this gate will not take it back. So "a keyed caller is
//! never refused while a stranger holds anything" is **not** true, and the only
//! way to make it true is the eviction ruled out above.
//!
//! What is true is a **ratchet**. From the moment occupancy reaches
//! `CONTENTION_PERCENT`, no stranger is admitted again, so anonymous occupancy
//! from there on only ever falls: every anonymous transfer that finishes hands
//! its slot to the keyed side and cannot take it back. A flood therefore cannot
//! hold a busy box at its anonymous maximum — it decays to zero at the rate
//! transfers complete, and `SshConfig::inactivity_timeout` bounds that rate from
//! above. The residue is bounded too, at `min(anonymous ceiling, limit −
//! reserve)`, which on the shipped appliance is 5 of 256. That is the number a
//! keyed caller could in the worst case be queued behind, and it is the number
//! the ceiling was chosen to make small.
//!
//! What anonymous keeps is therefore the *whole* allowance on a box that is not
//! busy, which is the ordinary state of an appliance and the only state a public
//! repository is interesting in. If keyed demand is permanently above
//! `CONTENTION_PERCENT` the public door is permanently shut — and that is
//! accepted, deliberately: a box that is three-quarters full of paid-for work
//! has nothing to spend on a stranger, the refusal is instant and names its own
//! rule so the operator can see it happening, and the two answers (a bigger
//! limit, or a browse replica) are both things an operator does rather than
//! things gunnar should guess. It is not a livelock: nothing is parked, nothing
//! is retried on gunnar's side, and a stranger costs a string compare.
//!
//! ## Why there is no priority WAKE order
//!
//! The obvious third mechanism — when a slot frees, wake keyed waiters before
//! anonymous ones — is not implemented because it would be a no-op.
//! [`Admission::admit_within`]'s queue **has no anonymous waiters in it**: every
//! refusal that names anonymity short-circuits the wait, so the only callers
//! parked on the gate's own `slot_freed` notify are keyed ones already. Adding
//! a priority queue would mean adding a data structure and a lock in order to
//! sort a queue with one class in it.
//!
//! ## Why not queue the overflow instead
//!
//! Same answer as the module's opening section, only more so. A bounded queue
//! of anonymous readers is a bounded pile of held sockets, and an unbounded one
//! is a slower outage with worse diagnostics. Refusing in milliseconds with a
//! sentence in it is the strictly better outcome for a `git` that will retry
//! anyway, and it is the only outcome that keeps the cost of an anonymous
//! request proportional to what gunnar is willing to spend on it.
//!
//! ## Fairness *among* anonymous readers
//!
//! There is none, and there cannot be: telling two strangers apart requires an
//! identity, and not having one is the definition of this caller. What bounds a
//! single greedy anonymous client is therefore not a share but the **time** any
//! one permit can be held — which is already bounded by
//! `SshConfig::inactivity_timeout` and by the transfer finishing. The ceiling's
//! job is to cap the total, and the existing timeouts recycle it.
//!
//! An operator who wants the old behaviour — a hard per-identity cap that binds
//! whether or not anyone else is there, e.g. a multi-tenant host that bills by
//! identity — sets
//! [`SshConfig::per_principal_limit_is_hard`](crate::SshConfig::per_principal_limit_is_hard).
//! The configured numbers keep their meaning either way.
//!
//! # Saturation is observable, by construction
//!
//! [`AdmissionStats`] is a snapshot of five monotonic-or-live counters, reachable
//! from [`BoundServer::admission`](crate::BoundServer::admission). Every refusal
//! is also logged at `warn` with the limit that was hit and what was in flight.
//! `tests/admission.rs` drives a server past its cap and asserts the counter
//! moved, not merely that nothing crashed.
//!
//! # The server-wide limit is a GOVERNOR now, not a constant
//!
//! `max_concurrent_transfers` used to be the number of permits in a semaphore,
//! and a fixed number is wrong in both directions at once: too high and the box
//! is already thrashing before anything is refused, too low and it refuses work
//! it could have done. It is also unknowable in advance — the right number
//! depends on the repositories, the disks and the other tenants.
//!
//! So the constant is **replaced** — not joined by a second mechanism — with
//! [`crate::governor::Governor`], and the operator's number becomes the band's
//! ceiling rather than the limit itself. Two gates, in this order, both of them
//! *at admission* and before anything has been allocated:
//!
//! 1. **time** — a CoDel controller that measures how long operations *waited*,
//! per class, and moves one shared limit;
//! 2. **space** — a memory reservation the server owns, rather than a
//! measurement of the machine it is taken from.
//!
//! `crate::governor` sets out why sojourn rather than queue length, why the
//! minimum rather than the average, why the target is per class, and what the
//! memory estimate does *not* yet bound. Everything below — the per-principal
//! max-min share, the counters, the refusal messages — is unchanged and now
//! divides the *governed* limit instead of the configured one, which is the
//! right divisor: a share of capacity the server does not have is not a share.
//!
//! # The usage share (`admit-share`): a class from `edda_admit`, a share from here
//!
//! With the `admit-share` feature the gate asks
//! [`edda_admit::Admit::plan`] for every **keyed** transfer, with the
//! namespace's rolling-window bytes out of the flow ledger
//! ([`gunnar_core::FlowLedger::window_bytes`]) as the usage. The plan's class
//! maps onto a share this gate already has:
//!
//! * `Full` → the per-principal max-min share, exactly as above;
//! * `Reduced` / `Floor` → **the anonymous ceiling**: a hard cap that binds on
//! an idle box, plus the keyed reserve. An over-budget namespace is still
//! served — at a stranger's width — and refused FAST, with
//! [`Refusal::ReducedShare`], never parked in [`Admission::admit_within`]'s
//! wait: the same reasoning that refuses anonymous instead of queueing it
//! applies to anybody held to the anonymous share.
//!
//! `record` is fed when the guard drops, with the transfer's wire bytes in each
//! direction, so the estimate `Usage.AdmitPlan` shows learns per kind. The
//! policy's own slot shares are not configured here — this gate is the share.
//!
//! **Without the feature there is no policy field, no call and no branch on
//! the path**: the admission decision is today's, and `tests/admission.rs`,
//! `anonymous_ceiling.rs` and `keyed_reserve.rs` run unchanged to prove it.
//!
//! # What this is NOT
//!
//! It is not intra-request parallelism, and it is deliberately the *opposite*
//! decision. gatling has no shared pool, so fanning a clone out inside the
//! request at forge concurrency spawns tens of thousands of OS threads, and
//! the measured deflate curve is already
//! *slower* at 64 threads than at 32. Throughput at the load target comes from
//! serving many requests at once, each costing about one core, which is exactly
//! what this gate meters.
use std::collections::HashMap;
use std::sync::atomic::{AtomicU64, Ordering};
use std::sync::{Arc, Mutex};
use std::time::Duration;
use crate::governor::{
Claim, Governor, Growth, Lane, LimitBand, OpClass, ServiceMark, Shed, DEFAULT_ADMISSION_WAIT,
DEFAULT_MEMORY_BUDGET_BYTES, WIRE_MEMORY_DENOMINATOR, WIRE_MEMORY_NUMERATOR, WIRE_RESERVE_STEP,
};
/// Why a transfer was not admitted.
///
/// Every variant carries the number that was hit, because a refusal an operator
/// cannot tie to a number is a mystery rather than a signal.
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub enum Refusal {
/// The server-wide concurrent-transfer limit is full.
AtCapacity {
/// The server-wide limit **in force at the moment of the refusal**.
///
/// The *governed* limit, not the configured one: the configured number
/// is the ceiling of
/// [`LimitBand`](crate::governor::LimitBand), and what actually refuses
/// is wherever the CoDel controller has moved to inside it. Reported
/// rather than assumed, for the same reason the per-principal share is:
/// an operator who read one number out of a config file and a different
/// one out of a log would conclude the gate was broken.
limit: usize,
},
/// This principal already holds its share of the server-wide limit.
PrincipalAtCapacity {
/// The per-principal limit **in force at the moment of the refusal**.
///
/// Under [`SshConfig::per_principal_limit_is_hard`] that is the
/// configured number. Otherwise it is the max-min share —
/// `max(configured, server_wide / active)` — which is why it is
/// reported rather than assumed: an operator who read the configured
/// number out of a file and a different number out of a log would
/// conclude the gate was broken.
///
/// [`SshConfig::per_principal_limit_is_hard`]: crate::SshConfig::per_principal_limit_is_hard
limit: usize,
/// How many identities held a transfer when this one was refused.
///
/// `1` means the refusal came from a hard configured cap: a share is
/// never smaller than the whole server when there is only one claimant.
/// Without this an operator cannot tell "you are being throttled
/// because the box is shared" from "you are being throttled because
/// somebody pinned a number".
active_principals: usize,
},
/// This operation's **lane** is full while the rest of the server is held
/// by the other lanes' floors (gate 1b, [`Lane::floor`]).
///
/// Its own variant because the operator action is different: "the box is
/// full" says raise the budget, and this one says a *write* storm is being
/// held to five eighths of the box on purpose so that reads still land.
LaneAtCapacity {
/// The lane that was full.
lane: Lane,
/// What that lane may hold right now.
ceiling: usize,
/// The governed server-wide limit it is a share of.
limit: usize,
},
/// The **anonymous** caller is already holding its whole ceiling.
///
/// Its own variant, and not [`Refusal::PrincipalAtCapacity`] with the
/// anonymous id in it, because the operator action is different in kind. A
/// per-principal refusal names a client that can be found, telephoned or
/// given a bigger share; this one says *strangers are at their cap*, which
/// is either normal (a popular public repository) or the first sign of a
/// scrape, and in neither case is the answer "raise the concurrency limit".
/// It also carries no `active_principals`, because there is exactly one
/// anonymous principal and dividing anything by it would be theatre.
AnonymousAtCapacity {
/// The anonymous ceiling in force, i.e.
/// [`Admission::max_anonymous`]. Unlike every other limit reported
/// here this one is *not* governed or shared — it is the operator's
/// number, unmodified — which is exactly why it is worth printing:
/// an operator who sees a refusal against 5 on a box with 200 free
/// slots must be able to see immediately that a different knob did it.
limit: usize,
},
/// The stranger asked for a slot inside the part of the box that only a
/// **keyed** identity may hold (D23b).
///
/// Its own variant, and specifically **not**
/// [`Refusal::AnonymousAtCapacity`], because the two say opposite things
/// about the machine and are fixed by opposite knobs. `AnonymousAtCapacity`
/// means *strangers are at their cap* — it can fire on an idle box, and the
/// only number that changes it is `--max-anonymous-transfers`. This one
/// means *the box is too busy to spend any of itself on a stranger*: the
/// anonymous ceiling is nowhere near reached, and lowering it would do
/// nothing at all. The knob here is the server-wide limit, or the size of
/// the box, or a browse replica.
///
/// It also carries no principal and no share, because it is not a statement
/// about this caller at all: it is a statement about occupancy, and the only
/// thing about the caller that entered into it was that it has no key.
KeyedReserve {
/// Slots occupied as the gate's own compare-and-swap observed them —
/// the reserve rides inside that atomic
/// ([`Governor::claim_reserving`](crate::governor::Governor::claim_reserving)),
/// so this is the number that actually made the decision and not a
/// second reading taken nearby. Reported so an operator can see the
/// decision rather than reconstruct it: `occupied + reserve >= limit`
/// is the whole rule, visible in one line of log.
occupied: usize,
/// The **governed** server-wide limit those slots are out of, i.e.
/// wherever the CoDel controller has moved to inside its band. The
/// configured number would be actively misleading here: the whole
/// reason this gate exists is that the governed limit moves and the
/// anonymous ceiling does not.
limit: usize,
/// How many slots at the top of that limit are keyed-only, i.e.
/// [`Admission::keyed_reserve`]. Printed because it is derived
/// ([`CONTENTION_PERCENT`]) and therefore appears in no config file an
/// operator could look it up in.
reserve: usize,
},
/// The server's **memory budget** is already spoken for (gate 2).
///
/// A third reason, kept separate from [`Refusal::AtCapacity`] on purpose: an
/// operator who reads "at capacity" raises the concurrency limit, and doing
/// that here makes the problem worse. The numbers travel so the log says
/// which knob to turn.
MemoryBudget {
/// The class whose estimate did not fit.
class: OpClass,
/// What [`OpClass::estimated_bytes`] asked for.
wanted: u64,
/// What was reserved by operations already in flight.
reserved: u64,
/// The configured budget.
budget: u64,
},
/// **The gate is PAUSED** — every transfer is refused, not for want of a
/// slot but because the server asked for a quiet moment: the runtime
/// data-set growth drains the box before it unmounts `/data`
/// (`DATA-SET-GROWTH-FLOW.md` §3 step 5, T1/T12). Carries no number
/// because no number was hit; carries the reason, whose words the client
/// reads. Set and cleared by [`Admission::pause`] / [`Admission::unpause`]
/// — one gate, so all three git fronts pause together.
Paused {
/// Why, and in which words.
pause: Pause,
},
/// **This namespace is over its usage budget** (`admit-share`), so the
/// transfer was held to the anonymous share and that share was full.
///
/// Its own variant, and neither [`Refusal::AnonymousAtCapacity`] nor
/// [`Refusal::KeyedReserve`], because the caller HAS a key and the operator
/// action is different again: the tenant's budget (`--admit-limit`), or the
/// tenant waiting for the window to move. The plan's sentence ("100 GB/30d
/// used 112 GB; this clone ~1.2 GB; Reduced") is in the `warn` row beside it.
#[cfg(feature = "admit-share")]
ReducedShare {
/// The anonymous ceiling the namespace is held to.
ceiling: usize,
/// `true` when the box was past `CONTENTION_PERCENT` and the keyed
/// reserve refused it; `false` when its own ceiling was full.
busy: bool,
},
}
/// **Why the admission gate is paused** — the reason, and the sentence the
/// client is handed for it. An enum rather than a `String` so [`Refusal`]
/// stays `Copy` and the words are fixed in one place, where a test can hold
/// them.
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub enum Pause {
/// The data set is being grown at runtime: drained, unmounted, grown,
/// mounted again — minutes, and the box stays up throughout.
Growing,
}
impl Pause {
/// The client's sentence. **The word `growing` is load-bearing** the way
/// `starting` is in [`crate::STARTING_MESSAGE`]: it is what tells a
/// person, a CI script and the bench that this is a planned pause with an
/// end, not a broken repository and not a full box.
pub const fn words(self) -> &'static str {
match self {
Pause::Growing => GROWING_MESSAGE,
}
}
/// The log's `reason` field.
pub const fn reason(self) -> &'static str {
match self {
Pause::Growing => "paused-growing",
}
}
}
/// What a client is told while the data set grows. The contract's words
/// (`DATA-SET-GROWTH-FLOW.md` §0, ruling 1; the drain design note in
/// `.nornir/data-set-grow-drain.md`), verbatim.
///
/// It names no estate: the same build serves gunnar.rs, test2 and every sold estate, and a
/// test2 push told "gunnar.rs is growing" was told about somebody else's server (2026-10-09).
pub const GROWING_MESSAGE: &str = "The server is growing its storage to make room for new \
members. Pushes are paused — please try again in a few \
minutes.";
impl Refusal {
/// What the client is told, in its own protocol.
///
/// Says *try again* explicitly. "Server busy" with no instruction reads as a
/// permanent failure to a human and to a CI log alike, and this one is
/// transient by definition.
pub fn client_message(&self) -> String {
match self {
Refusal::AtCapacity { limit } => format!(
"the server is already serving its limit of {limit} concurrent git transfers; \
try again in a moment"
),
// Both arms must keep the words "per-principal limit", and neither
// may contain "concurrent git transfers": `installtest`'s
// `refusals_in_log` tells the two refusals apart by exactly those
// substrings, because the log records the refusal by its `Display`
// — the same sentence the client is handed. Two witnesses of one
// event only agree if they are reading the same string.
Refusal::PrincipalAtCapacity {
limit,
active_principals,
} if *active_principals > 1 => format!(
"you already have {limit} git transfers in flight, which is your fair share of \
this server while {active_principals} identities are using it; that is the \
per-principal limit in force right now, and it rises again as the server \
empties; try again in a moment"
),
Refusal::PrincipalAtCapacity { limit, .. } => format!(
"you already have {limit} git transfers in flight on this server, which is the \
per-principal limit; try again in a moment"
),
// Shares neither "concurrent git transfers" nor "per-principal
// limit": `installtest`'s `refusals_in_log` tells refusals apart by
// exactly those, and a fourth reason that collided with either
// would be counted as the wrong one.
Refusal::LaneAtCapacity {
lane,
ceiling,
limit,
} => format!(
"this server is holding {lane} operations to {ceiling} of its {limit} slots so \
that the other kinds of request still land; try again in a moment"
),
// Shares neither "concurrent git transfers" nor "per-principal
// limit", for the reason given twice above. It says *unauthenticated*
// in the first clause so a human reading it in a `git` output knows
// instantly that the fix is a key, not patience: this is the one
// refusal on the box that a caller can make go away by logging in.
Refusal::AnonymousAtCapacity { limit } => format!(
"this server is already serving {limit} unauthenticated readers, which is all it \
offers without a key; try again in a moment, or push a key and clone as \
yourself"
),
// Shares neither "concurrent git transfers" nor "per-principal
// limit", for the reason given twice above — and it must not be the
// same sentence as `AnonymousAtCapacity` either, because the two
// are fixed by different knobs. It says *with a key* so the human
// holding the `git` prompt learns the same thing the operator does:
// this refusal is about not having one.
Refusal::KeyedReserve {
occupied,
limit,
reserve,
} => format!(
"this server is {occupied} of {limit} slots busy and keeps its last {reserve} for \
callers with a key; try again in a moment, or push a key and clone as yourself"
),
// Deliberately shares neither of the two substrings above:
// `installtest`'s `refusals_in_log` tells refusals apart by
// "concurrent git transfers" and "per-principal limit", and a third
// reason that collided with either would be counted as the wrong
// one. It says *memory* so the operator raises the budget rather
// than the concurrency limit.
Refusal::MemoryBudget {
class,
wanted,
reserved,
budget,
} => format!(
"this server holds a memory budget of {} MiB for git operations in flight, {} MiB \
of it is reserved, and a {class} needs {} MiB more than is left; try again in a \
moment",
budget / (1024 * 1024),
reserved / (1024 * 1024),
wanted / (1024 * 1024),
),
// Its own words entirely (see `Pause::words`): none of the three
// substrings the other refusals are told apart by, and a
// sentence that says WHY and for how long.
Refusal::Paused { pause } => pause.words().to_owned(),
// Shares neither "concurrent git transfers" nor "per-principal
// limit" (`installtest`'s `refusals_in_log`), and says *budget* so
// the human at the prompt learns it is about their usage, not the box.
#[cfg(feature = "admit-share")]
Refusal::ReducedShare { ceiling, busy } => {
format!(
"this namespace is over its transfer budget for the window, so it is served at \
the reduced share of {ceiling} git transfers at once{}; try again in a moment, \
or ask for a larger budget",
if *busy { ", and none while the server is this busy" } else { "" }
)
}
}
}
/// The **machine-readable** reason, for a log's `reason` field.
///
/// The sentence above is for the person holding the `git` prompt; this is
/// for the operator filtering a log, and they need different things. Until
/// the anonymous ceiling existed every refusal here was fairly described by
/// one string — the box was out of *something* server-wide — and both
/// fronts hard-coded `admission-gate-full` accordingly. That string is now
/// actively misleading on the one refusal that can fire with the box nearly
/// empty: an operator who greps `admission-gate-full`, sees a hit, and
/// looks at 59 free slots concludes the gate is broken, which is exactly
/// the "the limit exists but produces no signal an operator can act on"
/// failure the unbounded listener was about, wearing a different hat.
///
/// Only the fronts that can actually *see* an anonymous caller use this.
/// The gRPC control plane keeps its literal, because
/// [`gunnar_grpc::auth`] refuses the anonymous principal outright: a
/// `reason` there that could report anonymity would be describing a state
/// that crate makes unreachable.
pub const fn reason(&self) -> &'static str {
match self {
Refusal::AtCapacity { .. } => "admission-gate-full",
Refusal::LaneAtCapacity { .. } => "admission-lane-full",
Refusal::PrincipalAtCapacity { .. } => "admission-principal-share",
Refusal::AnonymousAtCapacity { .. } => "admission-anonymous-ceiling",
Refusal::KeyedReserve { .. } => "admission-keyed-reserve",
Refusal::MemoryBudget { .. } => "admission-memory-budget",
Refusal::Paused { pause } => pause.reason(),
#[cfg(feature = "admit-share")]
Refusal::ReducedShare { .. } => "admission-reduced-share",
}
}
}
impl std::fmt::Display for Refusal {
fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
f.write_str(&self.client_message())
}
}
/// A snapshot of the gate's counters.
///
/// Taken atomically field by field rather than under a lock: these are a
/// **signal**, and a signal that costs a mutex on the request path would be
/// paid for by the thing it is measuring.
#[derive(Debug, Clone, Copy, PartialEq, Eq, Default)]
pub struct AdmissionStats {
/// Transfers let through since the server started.
pub admitted: u64,
/// Transfers refused because the server-wide limit was full.
pub refused_at_capacity: u64,
/// Transfers refused because one principal held its whole share.
pub refused_per_principal: u64,
/// Transfers refused because the **memory budget** was spoken for (gate 2).
pub refused_memory: u64,
/// Reads refused because the **anonymous ceiling** was full.
///
/// Deliberately not folded into
/// [`refused_per_principal`](Self::refused_per_principal): that counter
/// moving means a client gunnar knows about is being throttled, and this one
/// moving means strangers are at their cap. A dashboard that added them
/// together would show one number that can be entirely benign (a popular
/// public repository, working exactly as designed) or entirely alarming (a
/// keyed CI fleet starved of its share) with no way to tell which.
pub refused_anonymous: u64,
/// Reads refused because the slot they wanted was inside the **keyed
/// reserve** (D23b).
///
/// Deliberately not folded into
/// [`refused_anonymous`](Self::refused_anonymous), which is the other half
/// of the same door and means the opposite thing. `refused_anonymous`
/// moving on an otherwise quiet box is *normal* — a popular public
/// repository, the ceiling doing its job — and the operator response is
/// none. This counter moving means the box was over
/// [`CONTENTION_PERCENT`] full of real work at the moment a stranger
/// asked, which is a **capacity** signal: it is the public door being shut
/// by load, and it will keep being shut until either the load falls or the
/// box grows. Summed together the two would show one number that means
/// "everything is fine" and "your public repositories are effectively
/// offline" with no way to tell which.
pub refused_keyed_reserve: u64,
/// Transfers refused at the **reduced share** (`admit-share`): a namespace
/// over its usage budget, held to the anonymous width.
#[cfg(feature = "admit-share")]
pub refused_reduced: u64,
/// Transfers refused because the gate was **paused** ([`Pause`]): the
/// data set was being grown. Its own counter, because this one moving is
/// neither load nor strangers — it is a planned window, and an operator
/// reading it wants to know how many clients met the window, not to
/// raise any limit.
pub refused_paused: u64,
/// Transfers in flight right now.
pub in_flight: u64,
/// The most that were ever in flight at once. This is the number that says
/// how close to the ceiling a server has actually been; `in_flight` alone
/// never shows it, because a sampler always misses the peak.
pub peak_in_flight: u64,
/// The **governed** server-wide limit right now, i.e. where the CoDel
/// controller has moved to inside its band. `0` when there is no limit.
///
/// Without this the shed is invisible: an operator seeing refusals against a
/// configured 100 has no way to know the governor has taken it to 37, which
/// is the same "the limit exists but produces no signal" the unbounded listener was
/// written about.
pub limit: usize,
/// Bytes reserved by operations in flight, against
/// [`memory_budget`](Admission::memory_budget).
pub reserved_bytes: u64,
/// Operations that were admitted only after **waiting** for a slot
/// ([`DEFAULT_ADMISSION_WAIT`]).
///
/// This is the queue made countable. It is not a failure count: every one
/// of these was served, and every one of them would have been a refusal
/// before 2026-08-30.
pub admitted_after_wait: u64,
/// Milliseconds spent waiting, summed over every operation that waited.
pub waited_ms_total: u64,
/// The longest single admission wait, in milliseconds. The number that says
/// whether the bound is doing anything.
pub peak_wait_ms: u64,
/// Waits that ran out and became refusals.
pub refused_after_wait: u64,
/// Refusals because the operation's **lane** was at its ceiling while the
/// other lanes' floors held the rest.
pub refused_lane: u64,
/// Transfers whose reservation grew past the memory budget while they ran.
/// Nonzero means one transfer is now keeping new ones out — which is the
/// budget working, not failing.
pub overdrawn: u64,
/// Bytes charged by running transfers over and above their admission
/// reservations.
pub grown_bytes: u64,
/// Milliseconds between admission and the transfer's work actually
/// starting, summed over every transfer that reported it.
///
/// **Observed, not controlled.** Until 2026-08-30 this gap *was* the CoDel
/// controller's input, and that was the defect: it is the tokio ready queue
/// plus the blocking-pool handoff, which is set by how many clients are
/// connected and doing crypto, not by how many transfers are admitted. A
/// refused client reconnects, so shedding made this number worse and the
/// controller shed until it hit its floor — the `limit of 19`
/// measured. It is kept as a *counter* because it is a real and useful
/// symptom of a box that is out of CPU; it is no longer a control input.
pub handoff_ms_total: u64,
/// The longest single admission-to-work gap, in milliseconds.
pub peak_handoff_ms: u64,
/// Transfers that finished **without ever reporting that their work
/// started**, so they contributed no sojourn sample.
///
/// A blindness counter, not a health one. The CoDel controller only sees
/// what [`AdmissionGuard::entered_service`] files; a caller that forgets to
/// call it leaves the controller with no samples at all, and a controller
/// with no samples grows to the top of its band and stays there — a gate
/// that looks like it is working and is not. This number must be 0 on any
/// path that is actually governed.
pub unmeasured_sojourn: u64,
}
impl AdmissionStats {
/// Every refusal, however it was refused.
pub fn refused(&self) -> u64 {
self.refused_at_capacity
+ self.refused_lane
+ self.refused_per_principal
+ self.refused_memory
+ self.refused_anonymous
+ self.refused_keyed_reserve
+ self.refused_paused
+ self.refused_reduced()
}
#[cfg(feature = "admit-share")]
fn refused_reduced(&self) -> u64 {
self.refused_reduced
}
#[cfg(not(feature = "admit-share"))]
fn refused_reduced(&self) -> u64 {
0
}
/// The mean admission wait over the operations that waited, in
/// milliseconds. `0` when nothing has ever waited.
pub fn mean_wait_ms(&self) -> u64 {
self.waited_ms_total
.checked_div(self.admitted_after_wait.max(1))
.unwrap_or(0)
}
}
#[derive(Debug, Default)]
struct Counters {
admitted: AtomicU64,
refused_at_capacity: AtomicU64,
refused_lane: AtomicU64,
refused_per_principal: AtomicU64,
refused_memory: AtomicU64,
/// Reads refused by the anonymous ceiling.
refused_anonymous: AtomicU64,
/// Reads refused by the keyed reserve (D23b).
refused_keyed_reserve: AtomicU64,
/// Transfers refused while the gate was paused ([`Pause`]).
refused_paused: AtomicU64,
/// Transfers refused at the reduced share (`admit-share`).
#[cfg(feature = "admit-share")]
refused_reduced: AtomicU64,
in_flight: AtomicU64,
peak_in_flight: AtomicU64,
unmeasured_sojourn: AtomicU64,
/// Admissions that had to wait for a slot at all.
admitted_after_wait: AtomicU64,
/// Milliseconds spent waiting, summed over every operation that waited.
waited_ms_total: AtomicU64,
/// The longest single wait, in milliseconds.
peak_wait_ms: AtomicU64,
/// Waits that ran out and became refusals.
refused_after_wait: AtomicU64,
/// Transfers whose growing reservation went past the budget
/// ([`Growth::Overdrawn`]).
overdrawn: AtomicU64,
/// Bytes charged to the ledger by running transfers, over and above their
/// admission reservations.
grown_bytes: AtomicU64,
/// Milliseconds between admission and the transfer's work starting, summed.
handoff_ms_total: AtomicU64,
/// The longest single admission-to-work gap, in milliseconds.
peak_handoff_ms: AtomicU64,
}
/// How full the server-wide gate must be, in percent, before the per-principal
/// share is enforced at all.
///
/// **Chosen from the measurement, not picked round.** With the shipped defaults
/// (256 server-wide, 32 per principal) 75 % is 192 in flight, which leaves 64
/// free slots — exactly **twice** the configured per-principal floor — as
/// headroom in which a newly arriving identity is never asked about fairness.
/// Two floors rather than one, so a second identity can arrive *and* be
/// throttled to its floor without the first having to give anything back.
///
/// Below the threshold there are free slots by definition, and refusing a
/// transfer there refuses work the box could actually do: that is the exact
/// mistake the old unconditional cap made, where 32 clients on one key were
/// refused against 224 idle slots. 192 in flight is also already 6× the core
/// count of the 32-core box the defaults were sized for
/// (Measurement 3 — 63 % efficiency at 32
/// threads), so the machine is deep into queueing before fairness engages, and
/// fairness is only interesting once queueing has started.
///
/// It is deliberately **not** configurable. A third number to get wrong is how
/// the first two ended up unexamined for four sessions; the two an operator
/// sizes are the limit and the floor.
pub const CONTENTION_PERCENT: usize = 75;
/// **How many git reads the server will serve without a key at all.**
///
/// Five. Chosen small and chosen *absolute*, and both halves are the decision:
///
/// * **Small.** Against the shipped 256 server-wide this is under 2 % of the
/// box. That number is not a guess at demand, it is a statement about
/// *liability*: it is the most an operator can lose to a caller who cannot be
/// named, revoked, billed or telephoned. Everything above five anonymous
/// readers is served by them coming back a moment later, which git clients do
/// by themselves; nothing above five is worth an operator's console being
/// slow.
/// * **Absolute, not a fraction of the limit.** A percentage would grow the
/// stranger's allowance every time an operator bought a bigger box, which is
/// exactly backwards — a bigger appliance is bought to serve *the team*
/// better, and the untrusted share should stay where it was. It also keeps
/// this ceiling meaningful when the CoDel controller has shed the governed
/// limit right down: a fraction of a shrinking limit shrinks with it, and the
/// moment the box is in trouble is the moment this cap most needs to hold
/// still.
///
/// It is far below [`DEFAULT_MAX_TRANSFERS_PER_PRINCIPAL`] (32), which is the
/// number *one keyed identity is guaranteed*, and that ordering is asserted by
/// `tests/anonymous_ceiling.rs` rather than left to two constants drifting
/// apart. It is well above 1, because a public repository that can only ever be
/// cloned by one person at a time is not public in any useful sense — and
/// because a ceiling of 1 turns every ordinary CI runner pair into a refusal.
///
/// Raise it on a box whose *purpose* is public serving (a browse replica, see
/// [`crate::control::AnonymousPolicy`]); lower it to 0 — or close the door with
/// [`SshConfig::without_anonymous_access`](crate::SshConfig::without_anonymous_access)
/// — on one whose purpose is not.
///
/// [`DEFAULT_MAX_TRANSFERS_PER_PRINCIPAL`]: crate::config::DEFAULT_MAX_TRANSFERS_PER_PRINCIPAL
pub const DEFAULT_MAX_ANONYMOUS_TRANSFERS: usize = 5;
/// How long a parked admission sleeps before re-asking, when nothing wakes it.
///
/// A waiter is normally woken by a guard being dropped
/// ([`Admission::slot_freed`]). This bound exists for the other way the answer
/// changes: the CoDel limit grows by one every window
/// ([`crate::governor::INTERVAL`], 100 ms) whose delay was under target, so a
/// server that is not finishing anything can still have room a moment later.
/// One window is therefore exactly the right period — sleeping longer would
/// hold clients past the moment their slot existed, shorter would poll for no
/// new information.
const RETRY_INTERVAL: Duration = crate::governor::INTERVAL;
/// How often the `git.transfer.waited` row may be written, in milliseconds.
///
/// One second. See [`Admission::wait_row_due`]: the fact that clients are
/// waiting is worth a row, and one row per waiting client is worth none because
/// it is the only thing left on the screen.
const WAIT_ROW_INTERVAL_MS: u64 = 1000;
/// Whether an admission attempt that fails should move a refusal counter.
///
/// A wait loop asks the same door many times for one operation; only the last
/// answer is a refusal. See [`Admission::attempt`].
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
enum Count {
/// This ask is final: a refusal is a refusal.
Yes,
/// This ask is a retry inside a wait: a refusal here is a poll.
No,
}
impl Count {
fn bump(self, counter: &AtomicU64) {
if self == Count::Yes {
counter.fetch_add(1, Ordering::Relaxed);
}
}
}
/// The gate itself.
///
/// Cheap to clone by `Arc`; it holds one semaphore and one small map.
#[derive(Debug)]
pub struct Admission {
/// Both gates, or `None` when the operator turned the server-wide limit
/// off. That is a legitimate configuration (an appliance behind its own
/// front door), and it is spelled as an `Option` rather than as `usize::MAX`
/// so a log can say "unlimited" and mean it. Unlimited means unlimited: no
/// concurrency band and no memory budget, because a budget that binds on a
/// gate the operator switched off is a limit they did not ask for.
governor: Option<Governor>,
max_concurrent: Option<usize>,
max_per_principal: Option<usize>,
/// When `true`, `max_per_principal` binds unconditionally — the behaviour
/// before max-min fairness. `false` (the default) makes it the **floor** of
/// a share that only applies under contention.
hard_per_principal: bool,
/// The **anonymous** ceiling: the most reads
/// [`gunnar_core::ANONYMOUS_PRINCIPAL_ID`] may hold at once, or `None` to
/// let anonymous take the same max-min share as any keyed identity.
///
/// Kept as its own field rather than as an entry in some per-principal
/// override map, because a map would invite a second override and the
/// second override is a policy engine. There is exactly one caller that has
/// no identity, so there is exactly one number here.
anonymous_ceiling: Option<usize>,
/// Only ever holds principals with a transfer in flight; an entry is removed
/// when its count reaches zero, so an idle server carries an empty map
/// however many identities have ever connected.
///
/// Its **length** is therefore the live count of active identities, which is
/// what the max-min share divides by. Keeping the map free of zero entries
/// is no longer only hygiene: a leaked entry would shrink everybody's share.
per_principal: Mutex<HashMap<String, usize>>,
counters: Arc<Counters>,
/// Woken every time a guard is dropped, so a waiting admission learns of a
/// free slot the instant there is one rather than by polling.
///
/// `tokio::sync::Notify` and not a semaphore: the thing being waited for is
/// not one permit but *the three gates agreeing*, and only
/// [`Admission::try_admit`] knows whether they do. A waiter is woken, it
/// re-asks, and if the answer is still no it waits again — which is also
/// why a lost wakeup is impossible here: the `Notified` future is created
/// **before** the try, so a release that lands in between is still seen.
slot_freed: tokio::sync::Notify,
/// How long an operation may wait for a slot before it is refused. Zero
/// restores the instant refusal exactly.
wait: Duration,
/// How long a **keyed** (authenticated) caller may wait, when longer than
/// [`Admission::wait`]. `None`: the same wait as everybody. See
/// [`crate::governor::DEFAULT_KEYED_ADMISSION_WAIT`].
keyed_wait: Option<Duration>,
/// When this gate was built, so [`Admission::wait_row_due`] can hold a
/// deadline in one atomic instead of a lock.
epoch: std::time::Instant,
/// Milliseconds since [`Admission::epoch`] at which the next `waited` row
/// may be written.
next_wait_row_ms: AtomicU64,
/// **The pause** ([`Pause`]), `0` for none, else `1 + Pause as u8`. One
/// atomic read on every admission and nothing else on the hot path; set
/// by [`Self::pause`] from the runtime growth's drain, cleared by
/// [`Self::unpause`] when `/data` is back — or when the growth aborted.
pause: std::sync::atomic::AtomicU8,
/// **The per-namespace flow ledger, when the server wired one.**
///
/// `None` in every test in this crate and in any build with no control
/// store behind it, which is why it is an `Option` rather than something
/// this crate constructs: the gate must work on a bare `Admission::new`,
/// and a ledger it invented would count into a store nobody reads.
///
/// It is here rather than at the three doors that call [`Self::try_admit`]
/// — the SSH exec line, the smart-HTTP handler and the gRPC transfer gate —
/// because those three converge on [`Self::count_refusal`] and a refusal
/// counted at each of them would be the same event counted three ways, one
/// of which would go stale the first time a fourth door was added.
flow: Option<Arc<gunnar_core::FlowLedger>>,
/// **The admission policy** (`admit-share`): asked for a keyed transfer's
/// class. [`edda_admit::PassThrough`] until the server hands it the
/// configured one ([`Admission::with_policy`]), so a gate built by a test
/// that never heard of budgets decides exactly as before.
#[cfg(feature = "admit-share")]
policy: Policy,
}
/// `Arc<dyn Admit>` with a `Debug` that names the policy, so [`Admission`]
/// keeps its derive.
#[cfg(feature = "admit-share")]
#[derive(Clone)]
struct Policy(Arc<dyn edda_admit::Admit>);
#[cfg(feature = "admit-share")]
impl std::fmt::Debug for Policy {
fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
f.write_str(self.0.policy())
}
}
/// **The `edda_admit` kind of work an [`OpClass`] is.** Estimates learn per kind.
pub fn work_kind(class: OpClass) -> edda_admit::WorkKind {
match class {
OpClass::Clone => edda_admit::WorkKind::Clone,
OpClass::Fetch => edda_admit::WorkKind::Fetch,
OpClass::Push => edda_admit::WorkKind::Push,
OpClass::LsRemote | OpClass::Gc => edda_admit::WorkKind::Other,
}
}
impl Admission {
/// Build a gate from the two configured limits, with the per-principal one
/// as a max-min **share** (see the module docs), the server-wide one as the
/// ceiling of the governor's band, and the default memory budget.
pub fn new(max_concurrent: Option<usize>, max_per_principal: Option<usize>) -> Admission {
Admission::with_memory_budget(
max_concurrent,
max_per_principal,
DEFAULT_MEMORY_BUDGET_BYTES,
)
}
/// **Give the gate a flow ledger**, so a refusal is charged to the
/// namespace that met it and not only to the box's own counters.
///
/// Builder-style and optional for [`Admission::flow`]'s reason. A gate
/// without one refuses exactly as it always did and simply reports the
/// refusal nowhere but its own `AdmissionStats`, which is what every build
/// before the flow ledger existed did.
pub fn with_flow(mut self, flow: Arc<gunnar_core::FlowLedger>) -> Admission {
self.flow = Some(flow);
self
}
/// **The flow ledger this gate was given**, or `None`.
///
/// Handed out because this crate's other tenant-facing door — the exec line
/// that denies a principal who holds no grant on the repository it named —
/// needs to charge that tenant and has nowhere else in this crate to reach
/// a ledger. One handle, wired once by the server, rather than a second
/// field on `Shared` that a future constructor could forget to fill.
pub fn flow(&self) -> Option<&Arc<gunnar_core::FlowLedger>> {
self.flow.as_ref()
}
/// The same gate with the memory budget (gate 2) chosen explicitly.
///
/// The budget has to be an operator's number: by the OS-less law this code
/// cannot read the machine's RAM, so
/// [`DEFAULT_MEMORY_BUDGET_BYTES`](crate::governor::DEFAULT_MEMORY_BUDGET_BYTES)
/// is a constant with a measurement behind it rather than a fit to the box.
pub fn with_memory_budget(
max_concurrent: Option<usize>,
max_per_principal: Option<usize>,
memory_budget: u64,
) -> Admission {
Admission {
governor: max_concurrent
.map(|budget| Governor::new(LimitBand::for_budget(budget), memory_budget)),
max_concurrent,
max_per_principal,
hard_per_principal: false,
// On by default, because the anonymous door is open by default: a limit
// that has to be switched on is not protecting the configuration
// anybody actually runs.
anonymous_ceiling: Some(DEFAULT_MAX_ANONYMOUS_TRANSFERS),
per_principal: Mutex::new(HashMap::new()),
counters: Arc::new(Counters::default()),
slot_freed: tokio::sync::Notify::new(),
wait: DEFAULT_ADMISSION_WAIT,
keyed_wait: None,
epoch: std::time::Instant::now(),
next_wait_row_ms: AtomicU64::new(0),
pause: std::sync::atomic::AtomicU8::new(0),
flow: None,
#[cfg(feature = "admit-share")]
policy: Policy(Arc::new(edda_admit::PassThrough::new())),
}
}
/// **Give the gate its admission policy** (`admit-share`) — the one built
/// from `--admit-limit` ([`edda_admit::Settings::build`]; `none` is
/// PassThrough).
#[cfg(feature = "admit-share")]
pub fn with_policy(mut self, policy: Arc<dyn edda_admit::Admit>) -> Admission {
self.policy = Policy(policy);
self
}
/// **The admission policy in force**, for `Usage.AdmitPlan`.
///
/// Without `admit-share` the gate has none, and this is a fresh
/// [`edda_admit::PassThrough`] — every kind `Full`, no budget, nothing
/// learnt — which is the truth about that build.
pub fn admit_policy(&self) -> Arc<dyn edda_admit::Admit> {
#[cfg(feature = "admit-share")]
{
Arc::clone(&self.policy.0)
}
#[cfg(not(feature = "admit-share"))]
{
Arc::new(edda_admit::PassThrough::new())
}
}
/// **The usage `Usage.AdmitPlan` reports for `namespace`** — a READ for
/// the admin view, never on the admission path.
///
/// A policy with a budget answers from its own table (the same figure the
/// gate plans with). Without one — PassThrough, or a build without
/// `admit-share` — the flow ledger's window is summed directly, so the
/// view still shows what moved; a gate with no ledger counted nothing and
/// says zero.
pub fn admit_usage(
&self,
policy: &dyn edda_admit::Admit,
namespace: &str,
) -> edda_admit::Usage {
if policy.limit().is_some() {
return policy.usage(&edda_admit::Consumer::Namespace(namespace.to_owned()));
}
let window = edda_admit::DEFAULT_WINDOW_DAYS;
let bytes = match &self.flow {
Some(flow) if !namespace.is_empty() => {
flow.window_bytes(namespace, gunnar_core::now_ms(), window)
}
_ => 0,
};
edda_admit::Usage::new(bytes, window)
}
/// **The flow ledger as a seed** for a policy's table: a namespace's
/// bytes per UTC day over the window, read ONCE per namespace on first
/// sight. See [`gunnar_core::FlowLedger::moved`].
pub fn flow_seed(flow: Arc<gunnar_core::FlowLedger>) -> Arc<dyn edda_admit::Seed> {
Arc::new(
move |who: &edda_admit::Consumer, first_day: u64, today: u64| {
let edda_admit::Consumer::Namespace(namespace) = who else {
return Vec::new();
};
let days = u32::try_from(today.saturating_sub(first_day) + 1).unwrap_or(u32::MAX);
let now = gunnar_core::now_ms();
flow.moved(namespace, now, days)
.into_iter()
.filter_map(|(day, bytes)| u64::try_from(day).ok().map(|d| (d, bytes)))
.collect()
},
)
}
/// **The plan for a keyed transfer** (`admit-share`); `None` for the
/// anonymous principal, which has its own ceiling and no namespace budget.
#[cfg(feature = "admit-share")]
fn plan(&self, principal: &str, namespace: &str, class: OpClass) -> Option<edda_admit::Plan> {
if principal == gunnar_core::ANONYMOUS_PRINCIPAL_ID {
return None;
}
let policy = &*self.policy.0;
let who = edda_admit::Consumer::Namespace(namespace.to_owned());
// The policy's own table: lock-free loads for a namespace already
// seen, seeded ONCE from the flow ledger on first sight
// (`GunnarSsh::with_flow`). PassThrough answers without a table.
let usage = policy.usage(&who);
Some(policy.plan(&edda_admit::Work::new(work_kind(class), None), &who, &usage))
}
/// The same gate with `max_per_principal` as an **unconditional** ceiling.
///
/// The escape hatch for a host where an identity is a tenant and the cap is
/// what is being sold, rather than an appliance where the identities are all
/// the same team. Selected by
/// [`SshConfig::per_principal_limit_is_hard`](crate::SshConfig::per_principal_limit_is_hard).
pub fn hard_capped(
max_concurrent: Option<usize>,
max_per_principal: Option<usize>,
) -> Admission {
Admission {
hard_per_principal: true,
..Admission::new(max_concurrent, max_per_principal)
}
}
/// The same gate with a different memory budget (gate 2).
///
/// A builder rather than a fourth constructor argument, so the two
/// concurrency numbers keep the shape every existing caller passes them in.
/// Only meaningful before the gate has admitted anything: it replaces the
/// governor, and a governor holds live occupancy.
pub fn with_budget(self, memory_budget: u64) -> Admission {
Admission {
governor: self
.max_concurrent
.map(|budget| Governor::new(LimitBand::for_budget(budget), memory_budget)),
..self
}
}
/// A gate with no limits at all; every request is admitted.
///
/// **Including the anonymous ceiling.** "Unlimited" has to mean unlimited or
/// it is a trap: every test that reaches for this gate does so to take the
/// admission gate *out* of the picture, and one surviving ceiling would make
/// those tests fail for a reason that has nothing to do with what they are
/// about. An operator who wants only the anonymous ceiling builds a real
/// gate and passes `None` for the other two.
pub fn unlimited() -> Admission {
Admission {
anonymous_ceiling: None,
..Admission::new(None, None)
}
}
/// The same gate with the **anonymous ceiling** set explicitly;
/// `None` switches it off and lets anonymous take a keyed identity's share.
///
/// A builder and not a fourth constructor argument for the same reason
/// [`Admission::with_budget`] is one: the two concurrency numbers keep the
/// shape every existing caller already passes them in, and a gate built by
/// a test that has never heard of the anonymous door still gets the default ceiling rather
/// than silently getting none.
pub fn with_anonymous_ceiling(self, anonymous_ceiling: Option<usize>) -> Admission {
Admission {
anonymous_ceiling,
..self
}
}
/// The anonymous ceiling in force, or `None` when anonymous is held
/// only by the ordinary per-principal share.
pub fn max_anonymous(&self) -> Option<usize> {
self.anonymous_ceiling
}
/// **How many slots at the top of the box are keyed-only** right now
/// (D23b), or `0` on a gate with no server-wide limit to reserve out of.
///
/// Derived, never configured. It is whatever is left of the **governed**
/// limit above [`CONTENTION_PERCENT`], so it moves with the CoDel
/// controller — which is the entire point: the anonymous ceiling is an
/// absolute number and the box it is a share of is not, and a stranger
/// holding five of a shed nineteen is holding a quarter of the machine at
/// the moment the machine has least to give.
///
/// Reusing `CONTENTION_PERCENT` rather than inventing a fourth number is
/// the decision, not a shortcut. That constant is *defined* as the
/// occupancy at which gunnar starts asking keyed identities to share with
/// each other; the moment it does that is exactly the moment it should stop
/// asking them to share with strangers. Two thresholds would be two
/// opinions about when the box is busy, and the order between them would be
/// an accident of arithmetic rather than a decision.
///
/// At least `1` whenever there is a limit at all: a reserve that rounded to
/// zero on a small box would be an empty guarantee at exactly the size where
/// a single stranger is the whole problem.
pub fn keyed_reserve(&self) -> usize {
let Some(governor) = &self.governor else {
return 0;
};
let limit = governor.limit();
// The largest occupancy still BELOW contention is
// `ceil(limit * P / 100) - 1`; everything from there up is reserved.
// Integer throughout — a float threshold would put the boundary
// somewhere an operator cannot reproduce with a calculator.
limit - limit.saturating_mul(CONTENTION_PERCENT).div_ceil(100) + 1
}
/// Whether the per-principal limit binds unconditionally.
pub fn per_principal_is_hard(&self) -> bool {
self.hard_per_principal
}
/// The server-wide limit, if there is one.
pub fn max_concurrent(&self) -> Option<usize> {
self.max_concurrent
}
/// The per-principal limit, if there is one.
pub fn max_per_principal(&self) -> Option<usize> {
self.max_per_principal
}
/// The **governed** server-wide limit right now. `None` when unlimited.
pub fn governed_limit(&self) -> Option<usize> {
self.governor.as_ref().map(Governor::limit)
}
/// The memory budget (gate 2). `None` when unlimited.
pub fn memory_budget(&self) -> Option<u64> {
self.governor.as_ref().map(|g| g.memory().budget())
}
/// The governor, for a caller that wants the controller's own counters.
pub fn governor(&self) -> Option<&Governor> {
self.governor.as_ref()
}
/// Take a slot for `principal` running an operation of `class`, or say why
/// not. **Never blocks.**
///
/// This is the ONE door. Every user-level git operation — clone, fetch,
/// push, ls-remote, gc — passes through here and through nothing else, so
/// there is a single place where "the server cannot keep up" is decided and
/// a single place a counter has to be trusted.
///
/// Three gates, in this order, all of them **before any of the operation's
/// work is allocated**:
///
/// 1. **time** — the governed concurrency limit, moved by the per-class
/// CoDel controller;
/// 2. **space** — the memory reservation for `class`;
/// 3. **fairness** — the per-principal max-min share.
///
/// Time and space come first because they are facts about the *server*, and
/// a server that cannot serve anybody should not spend a lock working out
/// whose turn it was. The returned guard releases everything when it is
/// dropped — slot, reservation and principal count — so no error path, and
/// no panic, can skip it. That is how a leaked permit turns a working limit
/// into a server that refuses everything an hour after start-up.
///
/// The caller **must** call [`AdmissionGuard::entered_service`] when the
/// operation's work actually begins, or the controller gets no sojourn
/// sample from it and [`AdmissionStats::unmeasured_sojourn`] counts it.
/// `namespace` is the repository's namespace, for the flow ledger; `""`
/// where the door does not know one, which charges the refusal to the box
/// alone rather than inventing a tenant for it.
pub fn try_admit(
self: &Arc<Self>,
principal: &str,
namespace: &str,
class: OpClass,
) -> Result<AdmissionGuard, Refusal> {
self.attempt(principal, namespace, class, Count::Yes)
}
/// [`Admission::try_admit`], told whether this ask is the **final** one.
///
/// [`Admission::admit_within`] asks many times for one operation while it
/// waits, and a refusal counter that moved on every ask would say 21 where
/// one client was refused once. So the wait loop asks with [`Count::No`]
/// and counts once, at the end, when the refusal is final.
fn attempt(
self: &Arc<Self>,
principal: &str,
namespace: &str,
class: OpClass,
counting: Count,
) -> Result<AdmissionGuard, Refusal> {
let outcome = self.attempt_inner(principal, namespace, class, counting);
// **The per-namespace half of a FINAL refusal**, and the second of the
// two places it is charged from — see [`Self::charge_flow`] for why
// there are two and not one.
//
// RED when this was missing: `attempt` counts the box's own refusal
// counters INLINE (`counting.bump`, six sites) and never calls
// [`Self::count_refusal`], which only [`Self::admit_within`] reaches.
// So the SSH exec line charged the tenant and the two doors that call
// [`Self::try_admit`] — smart HTTP and the gRPC transfer gate — charged
// nobody. The compiler said so out loud (`unused variable: namespace`)
// and the claim that all three doors converged on one counting point
// was simply false.
if counting == Count::Yes && outcome.is_err() {
self.charge_flow(namespace);
}
outcome
}
/// **Charge one admission refusal to the namespace that met it.**
///
/// The one writer, called from the two places a refusal becomes FINAL:
/// [`Self::attempt`] for the doors that do not wait
/// ([`Self::try_admit`] — smart HTTP and the gRPC transfer gate), and
/// [`Self::count_refusal`] for the door that does
/// ([`Self::admit_within`] — the SSH exec line, which asks many times for
/// one operation and must charge exactly once, at the end).
///
/// Two call sites and not one because "final" means two different moments
/// on those two paths, and that is the same distinction [`Count`] already
/// exists to draw. What must not be two is the ARITHMETIC, and it is not.
fn charge_flow(&self, namespace: &str) {
if let Some(flow) = &self.flow {
flow.refuse(
namespace,
gunnar_core::now_ms(),
gunnar_core::Refusal::Admission,
);
}
}
fn attempt_inner(
self: &Arc<Self>,
principal: &str,
namespace: &str,
class: OpClass,
counting: Count,
) -> Result<AdmissionGuard, Refusal> {
// ── D23b: is this caller allowed the last slots? ───────────────────
//
// `true` only for the one principal that has no key behind it, so on the
// path that carries essentially all traffic this is a string compare
// against a `&'static str` and then nothing at all. Computed here rather
// than after the claim because it has to ride INSIDE the claim's
// compare-and-swap; see `Governor::claim_reserving` for the race that
// taught us so, and `Admission::keyed_reserve` for where the number
// comes from.
//
// Skipped when the ceiling is off (the operator opted out of anonymous read
// entirely, and `Admission::unlimited` must mean unlimited) and when it
// is 0 (the door is shut at every occupancy, and `AnonymousAtCapacity`
// is the truer sentence for that).
//
// ── the PAUSE, before any slot is looked at ─────────────────────
// A paused gate refuses everybody with the pause's words; nothing is
// claimed, so nothing has to be released, and the in-flight count
// keeps meaning "transfers that were admitted before the pause" —
// which is exactly what the drain waits on.
if let Some(pause) = self.paused() {
counting.bump(&self.counters.refused_paused);
return Err(Refusal::Paused { pause });
}
// ── admit-share: a keyed namespace over its budget takes the anonymous
// share (module docs). Without the feature this block does not exist
// and `reduced` is the constant `false`.
#[cfg(feature = "admit-share")]
let plan = self.plan(principal, namespace, class);
#[cfg(feature = "admit-share")]
let reduced = plan.as_ref().is_some_and(|p| {
matches!(
p.class,
edda_admit::Class::Reduced | edda_admit::Class::Floor
)
});
#[cfg(not(feature = "admit-share"))]
let (reduced, _) = (false, namespace);
let anonymous_ceiling = self
.anonymous_ceiling
.filter(|_| principal == gunnar_core::ANONYMOUS_PRINCIPAL_ID || reduced);
let reserve = match anonymous_ceiling {
Some(cap) if cap > 0 => self.keyed_reserve(),
_ => 0,
};
// The server-wide gates first: they are the resources that actually run
// out, and taking them first means the per-principal map is only ever
// touched by a request that could have run.
let counts = counting == Count::Yes;
let claim = match &self.governor {
Some(governor) => match governor.claim_reserving(class, reserve, counts) {
Ok(claim) => Some(claim),
Err(Shed::Reserved {
occupied,
limit,
reserve,
}) => {
#[cfg(feature = "admit-share")]
if reduced {
counting.bump(&self.counters.refused_reduced);
self.warn_reduced(plan.as_ref(), namespace);
return Err(Refusal::ReducedShare {
ceiling: anonymous_ceiling.unwrap_or(0),
busy: true,
});
}
counting.bump(&self.counters.refused_keyed_reserve);
return Err(Refusal::KeyedReserve {
occupied,
limit,
reserve,
});
}
Err(Shed::Concurrency { limit, .. }) => {
counting.bump(&self.counters.refused_at_capacity);
return Err(Refusal::AtCapacity { limit });
}
Err(Shed::LaneFull {
lane,
ceiling,
limit,
}) => {
counting.bump(&self.counters.refused_lane);
return Err(Refusal::LaneAtCapacity {
lane,
ceiling,
limit,
});
}
Err(Shed::Memory {
class,
wanted,
reserved,
budget,
}) => {
counting.bump(&self.counters.refused_memory);
return Err(Refusal::MemoryBudget {
class,
wanted,
reserved,
budget,
});
}
},
None => None,
};
// The absolute anonymous ceiling, which is a different question
// from the reserve above — that one asked *may this caller take the
// slots that are left*, and this one asks *how many may this caller ever
// hold at once*. Consulted here rather than before the claim because it
// needs the per-principal map, and the map is only ever touched by a
// request that could have run.
if self.max_per_principal.is_some() || anonymous_ceiling.is_some() {
let mut held = self.per_principal.lock().unwrap_or_else(|e| e.into_inner());
// Everything the decision needs, read once under the one lock the
// counts live under. `active` includes this candidate whether or
// not it already has an entry, because a share divided by the
// claimants that are already in is not this claimant's share.
let already_active = held.contains_key(principal);
let active = held.len() + usize::from(!already_active);
let count = held.entry(principal.to_string()).or_insert(0);
// The anonymous ceiling REPLACES the share for the anonymous
// principal; it does not combine with it. Not `min` of the two,
// because the two numbers answer different questions and a `min`
// would let the max-min rule's contention threshold leak in: on an
// idle server the share evaluates to the whole box, so `min` would
// read 5 — correct by accident — while under contention it could
// evaluate lower and the refusal would then be reported as a
// *share*, sending the operator to the wrong knob. One caller, one
// rule, one number in the log.
let (ceiling, anonymous) = match anonymous_ceiling {
Some(cap) => (cap, true),
// Unreachable with `None`: the `if` above is exactly the
// disjunction "there is a per-principal limit OR an anonymous
// ceiling", and this arm is the half where the second is off.
None => (
self.ceiling_for(self.max_per_principal.unwrap_or(usize::MAX), active),
false,
),
};
if *count >= ceiling {
// Do not leave a zero entry behind for a principal that was
// refused: `or_insert` may have just created one. This matters
// more than it used to — a stale entry inflates `active` and
// shrinks everyone's share.
if *count == 0 {
held.remove(principal);
}
drop(held);
// Hand the slot AND the memory reservation straight back. The
// guard is what normally does this, and there is no guard yet.
if let (Some(governor), Some(claim)) = (&self.governor, &claim) {
governor.release(claim, 0);
}
#[cfg(feature = "admit-share")]
if anonymous && reduced {
counting.bump(&self.counters.refused_reduced);
self.warn_reduced(plan.as_ref(), namespace);
return Err(Refusal::ReducedShare {
ceiling,
busy: false,
});
}
if anonymous {
counting.bump(&self.counters.refused_anonymous);
return Err(Refusal::AnonymousAtCapacity { limit: ceiling });
}
counting.bump(&self.counters.refused_per_principal);
return Err(Refusal::PrincipalAtCapacity {
limit: ceiling,
active_principals: active,
});
}
*count += 1;
}
self.counters.admitted.fetch_add(1, Ordering::Relaxed);
let in_flight = self.counters.in_flight.fetch_add(1, Ordering::Relaxed) + 1;
self.counters
.peak_in_flight
.fetch_max(in_flight, Ordering::Relaxed);
// admit-share: the policy's own slot shares are not configured here
// (this gate IS the share), so its `admit` cannot refuse a namespace;
// should a policy ever do so, the gate's decision above stands and the
// transfer simply teaches the estimate nothing.
#[cfg(feature = "admit-share")]
let admitted = plan.and_then(|plan| self.policy.0.admit(&plan).ok());
Ok(AdmissionGuard {
#[cfg(feature = "admit-share")]
admitted,
gate: Arc::clone(self),
// Some **exactly when** the block above incremented a count, so the
// guard's `Drop` decrements precisely what was incremented. The two
// conditions have to be the same expression: an anonymous
// admission counted under a gate with no per-principal limit and
// released by `self.max_per_principal.map(..)` would never be
// released, and the ceiling would fall by one per anonymous clone
// until the door was shut for good.
principal: (self.max_per_principal.is_some() || anonymous_ceiling.is_some())
.then(|| principal.to_string()),
claim,
marked: ServiceMark::default(),
wire: AtomicU64::new(0),
grown: AtomicU64::new(0),
})
}
/// The `warn` row for a reduced-share refusal, carrying the plan's sentence
/// — the one thing a `Copy` [`Refusal`] cannot.
#[cfg(feature = "admit-share")]
fn warn_reduced(&self, plan: Option<&edda_admit::Plan>, namespace: &str) {
tracing::warn!(
target: "gunnar::git",
event = "git.admission.reduced",
namespace,
why = plan.map_or("", |p| p.why.as_str()),
"a namespace over its transfer budget met a full reduced share"
);
}
/// The same gate with a different **admission wait**.
///
/// `Duration::ZERO` is the pre-2026-08-30 gate exactly: refuse on the spot.
/// See [`DEFAULT_ADMISSION_WAIT`] for why waiting is cheaper than refusing
/// once the client's reconnect is counted.
pub fn with_wait(self, wait: Duration) -> Admission {
Admission { wait, ..self }
}
/// How long this gate lets an operation wait for a slot.
pub fn wait(&self) -> Duration {
self.wait
}
/// The same gate with a **keyed wait**: how long an authenticated caller
/// may be queued for a slot ([`crate::governor::DEFAULT_KEYED_ADMISSION_WAIT`]).
/// `None` gives keyed callers the ordinary [`Admission::wait`].
pub fn with_keyed_wait(self, keyed_wait: Option<Duration>) -> Admission {
Admission { keyed_wait, ..self }
}
/// **How long `principal` may wait for a slot.** Anonymous: the ordinary
/// wait. A keyed identity: the keyed wait when it is longer — a refused
/// key reconnects (git-lfs's pure-SSH transfer cannot even do that: it
/// falls back to `git-lfs-authenticate` and fails the push), while the
/// parked future costs one session the key's own share already bounds. A
/// zero wait (`--admission-wait-ms 0`) is "refuse on the spot" for all.
pub fn wait_for(&self, principal: &str) -> Duration {
if self.wait.is_zero() || principal == gunnar_core::ANONYMOUS_PRINCIPAL_ID {
return self.wait;
}
self.keyed_wait.map_or(self.wait, |k| k.max(self.wait))
}
/// **Take a slot, waiting up to [`Admission::wait`] for one.**
///
/// The door every transport goes through. It is
/// [`Admission::try_admit`] in a bounded loop, and everything that decides
/// admission still lives there — this adds only the bound and the counters
/// that make the bound visible.
///
/// Returns the guard and **how long it waited**, which is the number the
/// caller logs and the number the CoDel controller is fed
/// ([`Governor::record_sojourn`](crate::governor::Governor::record_sojourn)):
/// the wait in *this* queue is the one signal the gate has authority over,
/// because shedding is exactly what shortens it.
///
/// A refusal after the wait is the same [`Refusal`] the instant path
/// returns, with the same sentence — a client must not be able to tell "the
/// server refused at once" from "the server tried for two seconds first"
/// by the shape of the failure, only by the log.
///
/// # Cancellation
///
/// Safe to drop at any await point: nothing is held across one. The slot is
/// taken by the *last* `try_admit`, and either it returns a guard (which
/// owns the release) or it took nothing at all.
pub async fn admit_within(
self: &Arc<Self>,
principal: &str,
namespace: &str,
class: OpClass,
) -> Result<(AdmissionGuard, Duration), Refusal> {
let started = std::time::Instant::now();
let wait = self.wait_for(principal);
let mut refusal = match self.attempt(principal, namespace, class, Count::No) {
Ok(guard) => {
// A wait of zero is a sample too, and it is the one that grows
// the limit back on a server that is coping.
self.record_wait(class, Duration::ZERO);
return Ok((guard, Duration::ZERO));
}
Err(refusal) => refusal,
};
// No wait configured — or the caller is anonymous and was refused for
// being anonymous, which are the refusals this door never parks.
//
// The wait exists because a refused *keyed* client reconnects, and one
// parked future is cheaper than the reconnect. For anonymous that trade
// inverts: the parked future is holding the TCP socket, the russh
// session and its key material for up to `self.wait`, and those are
// exactly the resources a clone fork-bomb is spending gunnar's box to
// acquire. Parking N anonymous readers to protect them from a reconnect
// hands an attacker N held sessions for free — the queue becomes the
// attack. It is also pointless on its own terms: the ceiling is small
// and hard, so unlike a governed limit it will not have grown by the
// time the waiter re-asks; only a transfer finishing can free it, and
// a stranger has no claim on being first in line for that.
//
// `KeyedReserve` (D23b) is parked no more than `AnonymousAtCapacity` is,
// for the same reason twice over. The socket-and-session argument is
// unchanged. And what a parked stranger would be waiting for here is the
// box falling back below `CONTENTION_PERCENT` — which is a KEYED caller
// finishing. Parking a stranger to wait for that is queueing it for a
// slot the whole rule exists to deny it, and every millisecond it spends
// in that queue it is holding the resources the rule was protecting.
//
// `Paused` is never parked either: the pause lasts minutes, not the
// milliseconds a wait covers, and a client parked into a growth is a
// client holding a session across an unmount for nothing.
if wait.is_zero()
|| matches!(
refusal,
Refusal::AnonymousAtCapacity { .. }
| Refusal::KeyedReserve { .. }
| Refusal::Paused { .. }
)
|| is_reduced(&refusal)
{
self.count_refusal(namespace, &refusal);
return Err(refusal);
}
// Parked, and COUNTED as parked: the count is the demand signal
// `Governor::lane_ceiling` reserves other lanes' floors from, and the
// guard pairs the decrement onto every exit including a cancelled
// future.
let _parked = Parked::new(self.governor.as_ref(), class.lane());
loop {
let remaining = match wait.checked_sub(started.elapsed()) {
Some(remaining) if !remaining.is_zero() => remaining,
_ => {
self.counters
.refused_after_wait
.fetch_add(1, Ordering::Relaxed);
// The refusal is final now, so it is counted now — once,
// not once per poll.
self.count_refusal(namespace, &refusal);
// A wait that ran out is the strongest evidence of a
// standing queue there is; file it before refusing.
self.record_wait(class, wait);
return Err(refusal);
}
};
// Registered BEFORE the retry, so a slot freed between the retry
// and the wait is not a lost wakeup.
let freed = self.slot_freed.notified();
match self.attempt(principal, namespace, class, Count::No) {
Ok(guard) => {
let waited = started.elapsed();
let ms = u64::try_from(waited.as_millis()).unwrap_or(u64::MAX);
self.counters
.admitted_after_wait
.fetch_add(1, Ordering::Relaxed);
self.counters
.waited_ms_total
.fetch_add(ms, Ordering::Relaxed);
self.counters.peak_wait_ms.fetch_max(ms, Ordering::Relaxed);
self.record_wait(class, waited);
return Ok((guard, waited));
}
// The same rule inside the loop, for the caller whose FIRST
// refusal was a full server and whose next one is the anonymous
// ceiling: a parked stranger must stop being parked the moment
// the reason becomes "you are a stranger". Without this arm the
// short-circuit above would be defeated by ordering alone.
Err(
again @ (Refusal::AnonymousAtCapacity { .. } | Refusal::KeyedReserve { .. }),
) => {
self.count_refusal(namespace, &again);
self.record_wait(class, started.elapsed());
return Err(again);
}
// The reduced share is never parked either (module docs).
Err(again) if is_reduced(&again) => {
self.count_refusal(namespace, &again);
self.record_wait(class, started.elapsed());
return Err(again);
}
Err(again) => refusal = again,
}
// A timeout as well as a notification, because the thing being
// waited for is not only a released slot: the CoDel limit grows on
// its own, and a server whose limit came back while nothing was
// finishing would otherwise keep everybody parked until it did.
let _ = tokio::time::timeout(remaining.min(RETRY_INTERVAL), freed).await;
}
}
/// Move the counter this refusal belongs to. Called exactly once per
/// refused operation, however many times the wait loop asked.
fn count_refusal(&self, namespace: &str, refusal: &Refusal) {
// The tenant's number and the box's number are taken off the one event,
// so they cannot disagree about how many times the door was shut.
self.charge_flow(namespace);
let counter = match refusal {
Refusal::AtCapacity { .. } => &self.counters.refused_at_capacity,
Refusal::LaneAtCapacity { .. } => &self.counters.refused_lane,
Refusal::PrincipalAtCapacity { .. } => &self.counters.refused_per_principal,
Refusal::AnonymousAtCapacity { .. } => &self.counters.refused_anonymous,
Refusal::KeyedReserve { .. } => &self.counters.refused_keyed_reserve,
Refusal::MemoryBudget { .. } => &self.counters.refused_memory,
Refusal::Paused { .. } => &self.counters.refused_paused,
#[cfg(feature = "admit-share")]
Refusal::ReducedShare { .. } => &self.counters.refused_reduced,
};
counter.fetch_add(1, Ordering::Relaxed);
}
/// **May a `waited` row be written now?** At most one per
/// [`WAIT_ROW_INTERVAL`].
///
/// A row per waiting transfer is a row per *client* under load: the
/// 256-client shape produced 5411 of them in one run, which on the
/// appliance's twenty-row console is the whole screen and nothing else.
/// The fact still has to be visible, so what is emitted is one row per
/// second carrying the AGGREGATE (`admitted_after_wait`, `peak_wait_ms`),
/// which is the shape an operator can actually read. The per-transfer
/// detail stays at `debug`.
///
/// One compare-and-swap on a deadline, so a hundred racing waiters produce
/// one row and not a hundred.
pub fn wait_row_due(&self) -> bool {
let now = self.epoch.elapsed().as_millis() as u64;
self.next_wait_row_ms
.fetch_update(Ordering::Relaxed, Ordering::Relaxed, |due| {
(now >= due).then_some(now + WAIT_ROW_INTERVAL_MS)
})
.is_ok()
}
/// File one admission wait against the controller.
fn record_wait(&self, class: OpClass, waited: Duration) {
if let Some(governor) = &self.governor {
governor.record_sojourn(class, waited);
}
}
/// How many transfers this principal may hold **right now**.
///
/// The whole max-min rule, in one place, so there is exactly one writer of
/// it and a test can call it directly instead of inferring it from a
/// refusal. Called with the per-principal lock held and the candidate's own
/// server-wide permit already taken.
///
/// `configured` is the operator's number and `active` counts the identities
/// holding a transfer, this candidate included.
fn ceiling_for(&self, configured: usize, active: usize) -> usize {
// A pinned cap, or no server-wide limit to take a share OF. In the
// second case `max_per_principal` is the only limit there is, so it has
// to bind unconditionally or it means nothing.
let (Some(governor), false) = (&self.governor, self.hard_per_principal) else {
return configured;
};
// The share divides the **governed** limit, not the configured ceiling.
// A share of capacity the governor has already decided the box does not
// have is not a share; dividing the configured 100 while the controller
// is holding 37 would hand every identity a ceiling above the whole
// server.
let server_wide = governor.limit();
// Occupancy read from the governor rather than from `in_flight`: our own
// slot is already taken, and so is the slot of every other request
// currently between `try_claim` and its `in_flight` increment. The
// counter would miss those and read the server as emptier than it is,
// which is the direction that lets a burst past the share.
let occupied = governor.occupied();
// Integer form of `occupied >= CONTENTION_PERCENT% of server_wide`, so
// there is no float and no rounding to argue about.
let contended =
occupied.saturating_mul(100) >= server_wide.saturating_mul(CONTENTION_PERCENT);
if !contended {
// Free slots exist; the server-wide gate is the only thing that
// should be able to refuse. This is the case the old code got
// wrong.
return server_wide;
}
// Max-min: an equal share of the server, but never less than the
// configured floor — an appliance that promised 32 keeps promising 32
// however many identities turn up, and the server-wide semaphore is
// what stops the floors from over-committing the box.
configured.max(server_wide / active.max(1))
}
/// **Pause the gate**: from here until [`Self::unpause`] every transfer
/// on every front door is refused with `pause.words()`, and nothing new
/// takes a slot. Idempotent; a second pause replaces the first's reason.
pub fn pause(&self, pause: Pause) {
self.pause.store(1 + pause as u8, Ordering::SeqCst);
tracing::warn!(
target: "gunnar::limits",
event = "admission.paused",
reason = pause.reason(),
in_flight = self.counters.in_flight.load(Ordering::Relaxed),
"the admission gate is PAUSED: every new git transfer is refused with `{}` until the \
gate is unpaused; transfers already admitted run to completion",
pause.words()
);
}
/// **Unpause the gate.** Idempotent; an unpaused gate stays unpaused.
pub fn unpause(&self) {
if self.pause.swap(0, Ordering::SeqCst) != 0 {
tracing::info!(
target: "gunnar::limits",
event = "admission.unpaused",
refused_while_paused = self.counters.refused_paused.load(Ordering::Relaxed),
"the admission gate is open again"
);
}
}
/// Count a refusal the session gave at the paused door BEFORE reaching
/// [`Self::admit_within`] (the door answers before the command is
/// resolved, so the gate never sees the request).
pub fn note_refused_paused(&self) {
self.counters.refused_paused.fetch_add(1, Ordering::Relaxed);
}
/// The pause in force, if any.
pub fn paused(&self) -> Option<Pause> {
match self.pause.load(Ordering::SeqCst) {
0 => None,
n if n == 1 + Pause::Growing as u8 => Some(Pause::Growing),
_ => None,
}
}
/// Transfers in flight right now — the number the drain waits to see
/// reach zero. A snapshot of one atomic.
pub fn in_flight(&self) -> u64 {
self.counters.in_flight.load(Ordering::Relaxed)
}
/// Everything the gate knows, as of now.
pub fn stats(&self) -> AdmissionStats {
AdmissionStats {
admitted: self.counters.admitted.load(Ordering::Relaxed),
refused_at_capacity: self.counters.refused_at_capacity.load(Ordering::Relaxed),
refused_per_principal: self.counters.refused_per_principal.load(Ordering::Relaxed),
refused_memory: self.counters.refused_memory.load(Ordering::Relaxed),
refused_anonymous: self.counters.refused_anonymous.load(Ordering::Relaxed),
refused_keyed_reserve: self.counters.refused_keyed_reserve.load(Ordering::Relaxed),
refused_paused: self.counters.refused_paused.load(Ordering::Relaxed),
#[cfg(feature = "admit-share")]
refused_reduced: self.counters.refused_reduced.load(Ordering::Relaxed),
in_flight: self.counters.in_flight.load(Ordering::Relaxed),
peak_in_flight: self.counters.peak_in_flight.load(Ordering::Relaxed),
limit: self.governor.as_ref().map_or(0, Governor::limit),
reserved_bytes: self.governor.as_ref().map_or(0, |g| g.memory().reserved()),
unmeasured_sojourn: self.counters.unmeasured_sojourn.load(Ordering::Relaxed),
admitted_after_wait: self.counters.admitted_after_wait.load(Ordering::Relaxed),
waited_ms_total: self.counters.waited_ms_total.load(Ordering::Relaxed),
peak_wait_ms: self.counters.peak_wait_ms.load(Ordering::Relaxed),
refused_after_wait: self.counters.refused_after_wait.load(Ordering::Relaxed),
refused_lane: self.counters.refused_lane.load(Ordering::Relaxed),
overdrawn: self.counters.overdrawn.load(Ordering::Relaxed),
grown_bytes: self.counters.grown_bytes.load(Ordering::Relaxed),
handoff_ms_total: self.counters.handoff_ms_total.load(Ordering::Relaxed),
peak_handoff_ms: self.counters.peak_handoff_ms.load(Ordering::Relaxed),
}
}
fn release(&self, principal: Option<&str>) {
if let Some(principal) = principal {
let mut held = self.per_principal.lock().unwrap_or_else(|e| e.into_inner());
if let Some(count) = held.get_mut(principal) {
*count = count.saturating_sub(1);
if *count == 0 {
held.remove(principal);
}
}
}
self.counters.in_flight.fetch_sub(1, Ordering::Relaxed);
}
}
/// Whether `refusal` is the reduced share — the one `admit_within` must not
/// park. The constant `false` without `admit-share`.
fn is_reduced(refusal: &Refusal) -> bool {
#[cfg(feature = "admit-share")]
{
matches!(refusal, Refusal::ReducedShare { .. })
}
#[cfg(not(feature = "admit-share"))]
{
let _ = refusal;
false
}
}
/// A held transfer slot **and** its memory reservation. Releasing both is
/// [`Drop`], not a method, so no error path — and no panic — can skip it.
///
/// That is the whole answer to "an operation that ERRORS must still release":
/// there is no release call to forget, because unwinding runs `Drop`.
#[derive(Debug)]
pub struct AdmissionGuard {
gate: Arc<Admission>,
/// `None` when there is no per-principal limit, so the map is never touched
/// and never grown by a configuration that does not use it.
principal: Option<String>,
/// `None` on an unlimited gate, where there is neither a slot nor a
/// reservation to give back.
claim: Option<Claim>,
marked: ServiceMark,
/// Bytes seen on this transfer's wire, in both directions.
wire: AtomicU64,
/// What has been charged to the ledger on top of [`Claim::bytes`]. The
/// guard owns this so [`Drop`] releases exactly what was taken.
grown: AtomicU64,
/// The policy's guard (`admit-share`), handed to `record` on drop with
/// [`AdmissionGuard::wire`] as the cost — the counter the memory gate
/// already keeps, so recording adds nothing per byte.
#[cfg(feature = "admit-share")]
admitted: Option<edda_admit::Guard>,
}
impl AdmissionGuard {
/// **The operation's work has actually started.** Files this item's sojourn
/// against its class's window.
///
/// Called by the caller and not by the gate, because only the caller knows
/// when the waiting stopped. In `crate::server` that is the first line of
/// the spawned transfer task: everything between admission and there is the
/// runtime's ready queue and the blocking-pool handoff, which is exactly the
/// invisible queue this whole module exists to make visible.
///
/// One-shot: a second call is ignored rather than filing a second, longer
/// sample for the same item.
pub fn entered_service(&self) {
let (Some(_governor), Some(claim)) = (&self.gate.governor, &self.claim) else {
return;
};
if self.marked.claim() {
let handoff = claim.enqueued.elapsed();
let ms = u64::try_from(handoff.as_millis()).unwrap_or(u64::MAX);
self.gate
.counters
.handoff_ms_total
.fetch_add(ms, Ordering::Relaxed);
self.gate
.counters
.peak_handoff_ms
.fetch_max(ms, Ordering::Relaxed);
}
}
/// The class this operation was admitted as.
pub fn class(&self) -> Option<OpClass> {
self.claim.as_ref().map(|claim| claim.class)
}
/// **Charge this transfer for `bytes` more on the wire.**
///
/// Called by the transport as bytes actually move, in either direction, and
/// this is what makes the memory gate cost-based rather than flat: the
/// admission reservation ([`OpClass::admission_bytes`]) is the entry cost,
/// and everything proportional to the transfer arrives here.
///
/// Charges [`WIRE_MEMORY_NUMERATOR`]/[`WIRE_MEMORY_DENOMINATOR`] of the
/// bytes, in steps of [`WIRE_RESERVE_STEP`], so a per-byte quantity never
/// becomes a per-byte decision. Returns `Some(Growth)` on the calls that
/// actually charged and `None` in between, so a caller can log the
/// overdraw without testing for it.
///
/// It cannot refuse — see [`Growth`]. What an overdraw does is stop *new*
/// admissions until this transfer finishes.
pub fn reserve_more(&self, bytes: u64) -> Option<Growth> {
let (Some(governor), Some(_)) = (&self.gate.governor, &self.claim) else {
return None;
};
let wire = self.wire.fetch_add(bytes, Ordering::Relaxed) + bytes;
let want = wire * WIRE_MEMORY_NUMERATOR / WIRE_MEMORY_DENOMINATOR / WIRE_RESERVE_STEP
* WIRE_RESERVE_STEP;
let held = self.grown.load(Ordering::Relaxed);
if want <= held {
return None;
}
let extra = want - held;
// `fetch_max`, not a store: two directions can be metering at once and
// the ledger must be charged once per step, not once per racer.
if self.grown.fetch_max(want, Ordering::Relaxed) >= want {
return None;
}
self.gate
.counters
.grown_bytes
.fetch_add(extra, Ordering::Relaxed);
let growth = governor.charge(want, extra);
if matches!(growth, Growth::Overdrawn { .. }) {
self.gate.counters.overdrawn.fetch_add(1, Ordering::Relaxed);
}
Some(growth)
}
}
impl Drop for AdmissionGuard {
fn drop(&mut self) {
// ONE record per transfer, here, off the stream: the wire total is the
// counter `reserve_more` already keeps for the memory gate, and its
// direction is the class's — a push's bytes came in, everything
// else's went out. Nothing was added to the per-byte path for it.
#[cfg(feature = "admit-share")]
if let Some(admitted) = self.admitted.take() {
let wire = self.wire.load(Ordering::Relaxed);
let push = matches!(self.claim.as_ref().map(|c| c.class), Some(OpClass::Push));
self.gate.policy.0.record(
admitted,
&edda_admit::Cost {
net_in: if push { wire } else { 0 },
net_out: if push { 0 } else { wire },
..edda_admit::Cost::default()
},
);
}
if let (Some(governor), Some(claim)) = (&self.gate.governor, &self.claim) {
governor.release(claim, self.grown.load(Ordering::Relaxed));
// A transfer that never said its work started gave the controller
// nothing. Counted rather than guessed at: filing the whole
// lifetime as a "sojourn" here would be worse than filing nothing,
// because service time is not wait time and every long clone would
// read as a standing queue.
if !self.marked.claimed() {
self.gate
.counters
.unmeasured_sojourn
.fetch_add(1, Ordering::Relaxed);
}
}
self.gate.release(self.principal.as_deref());
// The slot is free now, so anybody parked in `admit_within` should
// re-ask. `notify_waiters` wakes everyone parked *at this moment* and
// stores nothing for a future waiter, which is right: a waiter that
// arrives later runs `try_admit` itself before it ever parks.
self.gate.slot_freed.notify_waiters();
}
}
/// One operation's presence in [`Governor::waiting_in`], released on every
/// path — return, refusal, or a dropped future.
struct Parked<'a> {
governor: Option<&'a Governor>,
lane: Lane,
}
impl<'a> Parked<'a> {
fn new(governor: Option<&'a Governor>, lane: Lane) -> Parked<'a> {
if let Some(governor) = governor {
governor.entered_wait(lane);
}
Parked { governor, lane }
}
}
impl Drop for Parked<'_> {
fn drop(&mut self) {
if let Some(governor) = self.governor {
governor.left_wait(self.lane);
}
}
}
/// **The wire meter**: an `AsyncRead + AsyncWrite` that charges its
/// [`AdmissionGuard`] for every byte that crosses it.
///
/// This is the other half of cost-based admission. [`OpClass::admission_bytes`]
/// is what an operation reserves before anybody knows how big it is; this is
/// how the ledger finds out. It wraps the same duplex the transport was already
/// given, so there is no second copy of the bytes and no second place where a
/// transfer's size is known.
///
/// Both directions are metered through one counter: a push's cost is the pack
/// coming in and a clone's is the pack going out, and the ledger does not care
/// which way a byte was travelling when it made the server hold something.
///
/// Zero cost when it is not doing anything: [`AdmissionGuard::reserve_more`]
/// returns on an integer comparison until a whole [`WIRE_RESERVE_STEP`] has
/// gone by.
pub struct Metered<S> {
inner: S,
guard: Arc<AdmissionGuard>,
}
impl<S> Metered<S> {
/// Meter `inner` against `guard`.
pub fn new(inner: S, guard: Arc<AdmissionGuard>) -> Metered<S> {
Metered { inner, guard }
}
/// Charge for `bytes`, and say so once when the charge did not fit.
fn charge(&self, bytes: usize) {
let Some(growth) = self.guard.reserve_more(bytes as u64) else {
return;
};
if let Growth::Overdrawn {
held,
reserved,
budget,
} = growth
{
tracing::warn!(
target: "gunnar::git",
event = "git.transfer.overdrawn",
held_bytes = held,
reserved_bytes = reserved,
budget_bytes = budget,
"this transfer has now charged the memory budget past its limit; it finishes, \
and NEW transfers are refused by name until it does"
);
}
}
}
impl<S: tokio::io::AsyncRead + Unpin> tokio::io::AsyncRead for Metered<S> {
fn poll_read(
mut self: std::pin::Pin<&mut Self>,
cx: &mut std::task::Context<'_>,
buf: &mut tokio::io::ReadBuf<'_>,
) -> std::task::Poll<std::io::Result<()>> {
let before = buf.filled().len();
let polled = std::pin::Pin::new(&mut self.inner).poll_read(cx, buf);
if polled.is_ready() {
let read = buf.filled().len().saturating_sub(before);
if read > 0 {
self.charge(read);
}
}
polled
}
}
impl<S: tokio::io::AsyncWrite + Unpin> tokio::io::AsyncWrite for Metered<S> {
fn poll_write(
mut self: std::pin::Pin<&mut Self>,
cx: &mut std::task::Context<'_>,
buf: &[u8],
) -> std::task::Poll<std::io::Result<usize>> {
let polled = std::pin::Pin::new(&mut self.inner).poll_write(cx, buf);
if let std::task::Poll::Ready(Ok(written)) = &polled {
let written = *written;
if written > 0 {
self.charge(written);
}
}
polled
}
fn poll_flush(
mut self: std::pin::Pin<&mut Self>,
cx: &mut std::task::Context<'_>,
) -> std::task::Poll<std::io::Result<()>> {
std::pin::Pin::new(&mut self.inner).poll_flush(cx)
}
fn poll_shutdown(
mut self: std::pin::Pin<&mut Self>,
cx: &mut std::task::Context<'_>,
) -> std::task::Poll<std::io::Result<()>> {
std::pin::Pin::new(&mut self.inner).poll_shutdown(cx)
}
}
#[cfg(test)]
mod tests {
use super::*;
/// **A keyed caller is QUEUED past the short wait; a stranger is not.**
///
/// gunnar.rs, 2026-10-25: every lane pushes and fetches as one key, and
/// git-lfs cannot retry a refused pure-SSH `git-lfs-transfer` — it falls
/// back to `git-lfs-authenticate` and fails the whole push ("batch
/// request: exec request failed"). So an authenticated identity waits for
/// a slot up to the keyed wait (30 s on the appliance, under `exec_grace`'s
/// 60 s), while anonymous keeps the short wait and its never-park rules.
#[tokio::test]
async fn a_keyed_caller_waits_out_the_keyed_bound_and_a_stranger_does_not() {
let gate = Arc::new(
Admission::new(Some(1), None)
.with_wait(Duration::from_millis(50))
.with_keyed_wait(Some(Duration::from_secs(5))),
);
assert_eq!(gate.wait_for("admin"), Duration::from_secs(5));
assert_eq!(
gate.wait_for(gunnar_core::ANONYMOUS_PRINCIPAL_ID),
Duration::from_millis(50)
);
let holder = gate
.try_admit("other", "team", OpClass::Fetch)
.expect("the one slot");
// The slot frees after 400 ms: well past the short wait, inside the keyed one.
let freer = tokio::spawn(async move {
tokio::time::sleep(Duration::from_millis(400)).await;
drop(holder);
});
let started = std::time::Instant::now();
let (guard, waited) = gate
.admit_within("admin", "team", OpClass::Fetch)
.await
.expect("a keyed caller is queued for the slot, not refused after 50 ms");
assert!(
waited >= Duration::from_millis(300),
"it really waited: {waited:?}"
);
assert!(started.elapsed() < Duration::from_secs(5));
freer.await.unwrap();
// The slot is held again: a stranger is refused at the short bound.
let started = std::time::Instant::now();
gate.admit_within(gunnar_core::ANONYMOUS_PRINCIPAL_ID, "team", OpClass::Fetch)
.await
.expect_err("anonymous is never queued for the keyed bound");
assert!(
started.elapsed() < Duration::from_secs(1),
"{:?}",
started.elapsed()
);
drop(guard);
// `--admission-wait-ms 0` still means "refuse on the spot" for everyone.
let instant = Admission::new(Some(1), None)
.with_wait(Duration::ZERO)
.with_keyed_wait(Some(Duration::from_secs(30)));
assert_eq!(instant.wait_for("admin"), Duration::ZERO);
// No keyed bound configured: the keyed caller gets the ordinary wait.
let plain = Admission::new(Some(1), None).with_wait(Duration::from_millis(50));
assert_eq!(plain.wait_for("admin"), Duration::from_millis(50));
}
/// **The drain's door** (`DATA-SET-GROWTH-FLOW.md` §3 step 5, T1/T12):
/// a paused gate refuses every new transfer with the growing sentence,
/// takes no slot, counts the refusal under its own counter, never parks
/// the caller, and leaves what was already admitted in flight — which
/// is the number the drain waits on. Unpaused, it admits again.
#[tokio::test]
async fn a_paused_gate_refuses_with_the_growing_words_and_keeps_in_flight_honest() {
let gate = Arc::new(Admission::new(Some(4), None).with_wait(Duration::from_secs(5)));
let running = gate
.try_admit("one", "team", OpClass::Push)
.expect("admitted before the pause");
assert_eq!(gate.in_flight(), 1);
assert_eq!(gate.paused(), None);
gate.pause(Pause::Growing);
assert_eq!(gate.paused(), Some(Pause::Growing));
let refused = gate
.try_admit("two", "team", OpClass::Clone)
.expect_err("paused: refused");
assert_eq!(
refused,
Refusal::Paused {
pause: Pause::Growing
}
);
assert_eq!(refused.client_message(), GROWING_MESSAGE);
assert!(
refused.client_message().contains("growing"),
"the word is load-bearing"
);
assert!(refused.client_message().contains("try again"));
// ★ Every estate says it, not only gunnar.rs (the UpCloud mockup, 2026-10-09: test2's
// pushes were told "gunnar.rs is growing its storage"): no estate's domain is spelled.
assert!(
!refused.client_message().contains("gunnar.rs"),
"a test2 push was told gunnar.rs is growing: {}",
refused.client_message()
);
assert!(
refused
.client_message()
.starts_with("The server is growing its storage"),
"{}",
refused.client_message()
);
assert_eq!(refused.reason(), "paused-growing");
for other in [
"concurrent git transfers",
"per-principal limit",
"unauthenticated",
] {
assert!(
!refused.client_message().contains(other),
"shares no substring with {other}"
);
}
// The waiting door does not park a paused caller: it answers at once.
let started = std::time::Instant::now();
let parked = gate
.admit_within("three", "team", OpClass::Fetch)
.await
.expect_err("paused: refused, not parked");
assert!(matches!(parked, Refusal::Paused { .. }));
assert!(
started.elapsed() < Duration::from_secs(1),
"not parked for the wait"
);
// Counted under its own counter, in the total, and no slot was taken.
let stats = gate.stats();
assert_eq!(stats.refused_paused, 2);
assert_eq!(stats.refused(), 2);
assert_eq!(
stats.in_flight, 1,
"the pre-pause transfer is still what is in flight"
);
gate.pause(Pause::Growing);
assert_eq!(gate.paused(), Some(Pause::Growing), "idempotent");
drop(running);
assert_eq!(
gate.in_flight(),
0,
"the drain sees zero once the last one finishes"
);
gate.unpause();
gate.unpause();
assert_eq!(gate.paused(), None);
let _again = gate
.try_admit("two", "team", OpClass::Clone)
.expect("unpaused: admitted");
assert_eq!(gate.in_flight(), 1);
}
#[test]
fn the_server_wide_limit_refuses_and_counts() {
let gate = Arc::new(Admission::new(Some(2), None));
let a = gate
.try_admit("one", "team", OpClass::LsRemote)
.expect("the first is admitted");
let b = gate
.try_admit("two", "team", OpClass::LsRemote)
.expect("the second is admitted");
let refused = gate
.try_admit("three", "team", OpClass::LsRemote)
.expect_err("the third is refused");
assert_eq!(refused, Refusal::AtCapacity { limit: 2 });
let stats = gate.stats();
assert_eq!(stats.admitted, 2, "two got in");
assert_eq!(stats.refused_at_capacity, 1, "the refusal must be COUNTED");
assert_eq!(stats.in_flight, 2);
assert_eq!(stats.peak_in_flight, 2);
// …and the slot comes back, so this is a limit and not a budget.
drop(a);
assert_eq!(gate.stats().in_flight, 1);
let c = gate
.try_admit("three", "team", OpClass::LsRemote)
.expect("a freed slot is reusable");
assert_eq!(gate.stats().in_flight, 2);
assert_eq!(
gate.stats().peak_in_flight,
2,
"the peak is a high-water mark, not the current value"
);
drop((b, c));
assert_eq!(gate.stats().in_flight, 0);
}
/// **Direction A of the max-min bar.** A single principal must be able to
/// reach the *server-wide* limit on an otherwise idle server.
///
/// This is the bug, at the shipped ratio. Before the fix the per-principal
/// number was a wall, so this run stopped at 32 with the rest of the server
/// free, and the whole limit needed eight identities to reach.
///
/// # What the governor changed here, and what it did not
///
/// The number a lone principal can reach is now the **governed** limit —
/// the top of [`LimitBand::for_ceiling`] of the configured ceiling — and not
/// the configured ceiling itself. With the shipped
/// `DEFAULT_MAX_CONCURRENT_TRANSFERS` of 256 that is
/// [`MAX_LIMIT`](crate::governor::MAX_LIMIT), 100: 256 concurrent git
/// transfers is the sort of number the governor exists to stop a server
/// from believing about itself.
///
/// The property under test is unchanged and is still the one that matters:
/// **the per-principal number is a floor, not a wall**, so one identity on
/// an idle server reaches the whole of whatever the server-wide limit
/// currently is. It is asserted against `governed` rather than a literal so
/// it cannot drift away from the band.
#[test]
fn one_principal_alone_reaches_the_server_wide_limit() {
use crate::config::{
DEFAULT_MAX_CONCURRENT_TRANSFERS as CONFIGURED,
DEFAULT_MAX_TRANSFERS_PER_PRINCIPAL as FLOOR,
};
let gate = Arc::new(Admission::new(Some(CONFIGURED), Some(FLOOR)));
let governed = gate.governed_limit().expect("a governed limit");
// The top of the band is the CONFIGURED ceiling, not the design's
// default. This assertion read `MAX_LIMIT` while the band was capped at
// 100, which made a server told 256 serve 100 and called it correct —
// the test agreed with the bug because both had the same number in them.
assert_eq!(
governed, CONFIGURED,
"a fresh governor starts at the top of its band, and the top is what \
the operator configured"
);
assert!(
governed > FLOOR,
"a test where the governed limit is not above the per-principal floor cannot tell a \
floor from a wall"
);
let held: Vec<_> = (0..governed)
.map(|i| {
gate.try_admit("solo", "team", OpClass::LsRemote)
.unwrap_or_else(|refusal| {
panic!(
"transfer {i} of {governed} was refused on an EMPTY server holding \
one identity: {refusal}. The per-principal number is a floor, not a \
wall; a single-key client that stops at {FLOOR} is the old invisible ceiling, back"
)
})
})
.collect();
assert_eq!(held.len(), governed);
let stats = gate.stats();
assert_eq!(
stats.admitted, governed as u64,
"the server-wide limit in force must be reachable by ONE identity"
);
assert_eq!(
stats.refused_per_principal, 0,
"not one of these {governed} may be refused for fairness: there was nobody to be \
fair to"
);
assert_eq!(stats.peak_in_flight, governed as u64);
assert_eq!(stats.limit, governed, "the stats must report what refuses");
// …and the server-wide gate is still a gate. Without this the test
// above would also pass on an Admission that had lost both limits.
let full = gate
.try_admit("solo", "team", OpClass::LsRemote)
.expect_err("one past the governed limit");
assert_eq!(full, Refusal::AtCapacity { limit: governed });
drop(held);
}
/// **Direction B of the max-min bar.** Once the server IS contended, a
/// greedy principal is held to its share and a quiet one is still served.
///
/// `(4, 1)` so the contention threshold (75 % → 3 of 4) is reachable in a
/// handful of admissions, and so the share (`4 / 2 = 2`) is strictly above
/// the configured floor of 1 — a share equal to the floor could not tell
/// max-min fairness from the old wall.
#[test]
fn a_greedy_principal_is_held_to_its_share_once_the_server_is_contended() {
let gate = Arc::new(Admission::new(Some(4), Some(1)));
// The quiet one arrives first, so it is *active* and has a claim.
let _polite = gate
.try_admit("polite", "team", OpClass::LsRemote)
.expect("polite's first");
// Greedy takes its share: 4 slots / 2 identities = 2.
let _g1 = gate
.try_admit("greedy", "team", OpClass::LsRemote)
.expect("greedy's first");
let _g2 = gate
.try_admit("greedy", "team", OpClass::LsRemote)
.expect("greedy's second is still its share");
// Three of four are held, so the server is over the 75 % threshold and
// fairness is in force. Greedy has its two.
let refused = gate
.try_admit("greedy", "team", OpClass::LsRemote)
.expect_err("greedy's third is past its share of a contended server");
assert_eq!(
refused,
Refusal::PrincipalAtCapacity {
limit: 2,
active_principals: 2
},
"the refusal must name the SHARE in force (4/2), not the configured floor (1), or an \
operator reads a number that matches nothing they set"
);
assert!(
refused.client_message().contains("per-principal limit"),
"installtest's `refusals_in_log` tells the two refusals apart by this exact phrase; \
losing it silently zeroes the server-side witness: {}",
refused.client_message()
);
// The whole reason fairness exists: the fourth slot was NOT taken by
// greedy, and it is there for somebody else.
let _other = gate
.try_admit("polite", "team", OpClass::LsRemote)
.expect("the slot greedy was refused must be available to the quiet principal");
let stats = gate.stats();
assert_eq!(stats.refused_per_principal, 1);
assert_eq!(
stats.refused_at_capacity, 0,
"the server had a free slot; attributing this to capacity sends an operator to raise \
the wrong number"
);
assert_eq!(stats.in_flight, 4);
}
/// The share is *work-conserving*: two identities on a server that is
/// nowhere near full do not each get half of it.
///
/// Without the contention threshold this is what max-min fairness costs —
/// a busy principal capped at `server_wide / active` while most of the box
/// sits idle, which is the same "binds when there is nobody to be fair to"
/// mistake in a smaller costume.
#[test]
fn fairness_does_not_bind_while_there_are_free_slots() {
let gate = Arc::new(Admission::new(Some(100), Some(1)));
let _quiet = gate
.try_admit("quiet", "team", OpClass::LsRemote)
.expect("one transfer from a quiet identity");
// An equal share of 100 between 2 identities is 50. The server is only
// going to 74 % full, so the share must not be enforced at all.
let busy: Vec<_> = (0..73)
.map(|i| {
gate.try_admit("busy", "team", OpClass::LsRemote)
.unwrap_or_else(|r| panic!("transfer {i} refused at {}% full: {r}", i + 2))
})
.collect();
assert_eq!(
busy.len(),
73,
"74 of 100 held, so nothing may be refused yet"
);
assert_eq!(gate.stats().refused_per_principal, 0);
// The very next one crosses the threshold — 75 of 100 occupied — and
// the share engages at once: 100/2 = 50, and `busy` is already at 73.
let refused = gate
.try_admit("busy", "team", OpClass::LsRemote)
.expect_err(
"the transfer that makes the server 75% full is the first one weighed \
against the share",
);
assert_eq!(
refused,
Refusal::PrincipalAtCapacity {
limit: 50,
active_principals: 2
}
);
}
/// The pinned mode: an operator who wants the old wall still has it.
#[test]
fn a_hard_cap_binds_on_an_empty_server() {
let fair = Arc::new(Admission::new(Some(64), Some(2)));
let _a = fair
.try_admit("solo", "team", OpClass::LsRemote)
.expect("first");
let _b = fair
.try_admit("solo", "team", OpClass::LsRemote)
.expect("second");
fair.try_admit("solo", "team", OpClass::LsRemote)
.expect("a fair gate lets a lone principal past its floor on an idle server");
let hard = Arc::new(Admission::hard_capped(Some(64), Some(2)));
let _c = hard
.try_admit("solo", "team", OpClass::LsRemote)
.expect("first");
let _d = hard
.try_admit("solo", "team", OpClass::LsRemote)
.expect("second");
let refused = hard
.try_admit("solo", "team", OpClass::LsRemote)
.expect_err("a pinned cap binds whether or not anyone else is there");
assert_eq!(
refused,
Refusal::PrincipalAtCapacity {
limit: 2,
active_principals: 1
}
);
assert!(
!refused.client_message().contains("fair share"),
"a pinned cap is not a share and must not be explained as one: {}",
refused.client_message()
);
assert_eq!(hard.stats().refused_per_principal, 1);
assert_eq!(fair.stats().refused_per_principal, 0);
}
#[test]
fn a_refused_transfer_does_not_leak_a_server_wide_slot() {
// The subtle one: the server-wide permit is taken BEFORE the
// per-principal check, so a per-principal refusal must give it back. If
// it did not, a client hammering its own limit would drain the server.
//
// Driven through the *fair* path rather than a hard cap, because that
// is what ships: `polite` first so the server is contended and greedy's
// share (4/2 = 2) is what it hammers.
let gate = Arc::new(Admission::new(Some(4), Some(1)));
let _polite = gate
.try_admit("polite", "team", OpClass::LsRemote)
.expect("polite is active");
let _g1 = gate
.try_admit("greedy", "team", OpClass::LsRemote)
.expect("greedy's share is 2");
let _g2 = gate
.try_admit("greedy", "team", OpClass::LsRemote)
.expect("greedy's share is 2");
for _ in 0..100 {
gate.try_admit("greedy", "team", OpClass::LsRemote)
.expect_err("refused every time");
}
// Three slots are held; the fourth must still be free for anybody at all.
let _other = gate
.try_admit("somebody-else", "team", OpClass::LsRemote)
.expect("the fourth slot survived 100 refusals");
assert_eq!(gate.stats().in_flight, 4);
assert_eq!(gate.stats().refused_per_principal, 100);
}
/// A principal that finishes its work must give its share back to the
/// others, or the map's length — the divisor of every share — is a
/// high-water mark rather than a live count.
#[test]
fn a_departed_principal_stops_shrinking_everybody_elses_share() {
let gate = Arc::new(Admission::new(Some(4), Some(1)));
let leaving = gate
.try_admit("leaving", "team", OpClass::LsRemote)
.expect("two identities are active");
let _g1 = gate
.try_admit("stayer", "team", OpClass::LsRemote)
.expect("first");
let _g2 = gate
.try_admit("stayer", "team", OpClass::LsRemote)
.expect("share of 2 with 2 active");
gate.try_admit("stayer", "team", OpClass::LsRemote)
.expect_err("held to 2 while `leaving` is still there");
drop(leaving);
// Now one identity is active, so the share is the whole server again.
let _g3 = gate
.try_admit("stayer", "team", OpClass::LsRemote)
.expect("with the other identity gone the share must widen back out");
let _g4 = gate
.try_admit("stayer", "team", OpClass::LsRemote)
.expect("all four slots are reachable by the only identity left");
assert_eq!(gate.stats().in_flight, 4);
}
#[test]
fn the_map_does_not_grow_with_every_principal_ever_seen() {
let gate = Arc::new(Admission::new(Some(4), Some(1)));
for i in 0..500 {
let guard = gate
.try_admit(&format!("principal-{i}"), "team", OpClass::LsRemote)
.expect("admitted");
drop(guard);
}
// Read through the gate's own lock rather than a counter, because the
// thing being asserted is that the MAP is empty; a counter could agree
// while the map leaked.
let held = gate.per_principal.lock().expect("the per-principal map");
assert!(
held.is_empty(),
"the per-principal map kept {} entries for principals with nothing in flight",
held.len()
);
}
/// The three refusals must stay **tellable apart by their words**, because
/// that is how the only cross-process witness of them counts.
///
/// `installtest`'s `refusals_in_log` reads the server log and sorts
/// refusals by the substrings `"concurrent git transfers"` and
/// `"per-principal limit"` — the log records a refusal by its `Display`,
/// which is the same sentence the client is handed. A third reason that
/// collided with either phrase would be silently counted as the wrong one,
/// and an operator would raise the wrong number: the concurrency limit,
/// when the thing that ran out was memory.
///
/// # Seen red
///
/// By giving `Refusal::MemoryBudget` the natural wording, "the server is
/// already serving its limit of N concurrent git transfers within its
/// memory budget". Failure:
///
/// ```text
/// the memory refusal contains "concurrent git transfers", which
/// installtest counts as a CAPACITY refusal; an operator reading that
/// raises the concurrency limit and makes the memory problem worse
/// ```
#[test]
fn the_three_refusals_do_not_borrow_each_others_words() {
let capacity = Refusal::AtCapacity { limit: 37 }.client_message();
let principal = Refusal::PrincipalAtCapacity {
limit: 4,
active_principals: 3,
}
.client_message();
let memory = Refusal::MemoryBudget {
class: OpClass::Clone,
wanted: 2048 * 1024 * 1024,
reserved: 6144 * 1024 * 1024,
budget: 8192 * 1024 * 1024,
}
.client_message();
assert!(capacity.contains("concurrent git transfers"));
assert!(!capacity.contains("per-principal limit"));
assert!(principal.contains("per-principal limit"));
assert!(!principal.contains("concurrent git transfers"));
assert!(
!memory.contains("concurrent git transfers"),
"the memory refusal contains \"concurrent git transfers\", which installtest counts \
as a CAPACITY refusal; an operator reading that raises the concurrency limit and \
makes the memory problem worse: {memory}"
);
assert!(
!memory.contains("per-principal limit"),
"the memory refusal contains \"per-principal limit\", so installtest counts a \
server-wide memory exhaustion as one identity's fairness throttle: {memory}"
);
assert!(
memory.contains("memory budget"),
"the memory refusal must name the knob to turn: {memory}"
);
// All three are transient and every one of them must say so, or a CI
// log reads a busy server as a broken one.
for message in [&capacity, &principal, &memory] {
assert!(
message.contains("try again"),
"a transient refusal that does not say `try again` reads as permanent: {message}"
);
}
}
/// The work an operation would do, if it were admitted.
///
/// Deliberately something with an **observable cost**: it allocates, and it
/// records that it allocated. "No work was performed" asserted against a
/// bool nobody sets is a hollow guard; asserted against a
/// byte count that a real `Vec` really pushed into is not.
#[derive(Default)]
struct Work {
runs: AtomicU64,
bytes_allocated: AtomicU64,
}
impl Work {
/// One operation's buffer. A clone's real one is proportional to the
/// pack; this stands in for it at a size a test can afford.
fn perform(&self) -> Vec<u8> {
let buffer = vec![0u8; 1024 * 1024];
self.runs.fetch_add(1, Ordering::Relaxed);
self.bytes_allocated
.fetch_add(buffer.len() as u64, Ordering::Relaxed);
buffer
}
fn runs(&self) -> u64 {
self.runs.load(Ordering::Relaxed)
}
fn bytes(&self) -> u64 {
self.bytes_allocated.load(Ordering::Relaxed)
}
}
/// **A denial happens AT admission, and nothing the operation would have
/// done gets done.**
///
/// This is the entire value of the gate. A refusal that arrives after the
/// buffer has been built is not load shedding, it is a slower way of paying
/// the same cost — and for the clone class that cost is gigabytes
/// (2314 MB peak RSS, measured 2026-08-13).
///
/// Both gates are driven, because they refuse on different paths and each
/// could grow a bug the other's test would not see.
///
/// # Seen red
///
/// By moving `Work::perform` above the `try_admit` in the loop below, which
/// is precisely the mistake a real caller makes when it opens the pack
/// before asking whether it may. Failure:
///
/// ```text
/// 30 operations were refused, but the work ran 40 times and allocated
/// 41943040 bytes. A refusal after the allocation is not load shedding
/// ```
#[test]
fn a_denial_is_paid_for_by_nobody() {
// Gate 1: the concurrency limit. A pinned band of 10, so the refusals
// are the governor's and not the memory ledger's.
let gate = Arc::new(Admission::new(Some(10), None));
let work = Work::default();
let mut held = Vec::new();
let mut refused = 0;
for _ in 0..40 {
match gate.try_admit("solo", "team", OpClass::LsRemote) {
Ok(guard) => held.push((guard, work.perform())),
Err(_) => refused += 1,
}
}
assert_eq!(refused, 30, "10 of 40 fit under the limit");
assert_eq!(
work.runs(),
10,
"{refused} operations were refused, but the work ran {} times and allocated {} \
bytes. A refusal after the allocation is not load shedding",
work.runs(),
work.bytes()
);
assert_eq!(work.bytes(), 10 * 1024 * 1024);
// Gate 2: the memory budget. Four clones' worth, and eight arrivals.
let clone = OpClass::Clone.admission_bytes();
let gate = Arc::new(Admission::new(None, None).with_budget(clone * 4));
assert!(
gate.memory_budget().is_none(),
"an unlimited concurrency gate has no budget to attach one to; the memory gate must \
ride on the same governor or a configuration could get one without the other"
);
let gate = Arc::new(Admission::with_memory_budget(Some(100), None, clone * 4));
let work = Work::default();
let mut held = Vec::new();
let mut refused = Vec::new();
for _ in 0..8 {
match gate.try_admit("solo", "team", OpClass::Clone) {
Ok(guard) => held.push((guard, work.perform())),
Err(refusal) => refused.push(refusal),
}
}
assert_eq!(work.runs(), 4, "a 4-clone budget admits 4 clones");
assert_eq!(
work.bytes(),
4 * 1024 * 1024,
"the 4 refused clones allocated {} bytes between them, which is 4 more than a \
refusal at admission can possibly cost",
work.bytes() - 4 * 1024 * 1024
);
assert_eq!(refused.len(), 4);
assert!(
refused.iter().all(|r| matches!(
r,
Refusal::MemoryBudget {
class: OpClass::Clone,
..
}
)),
"the refusals must name MEMORY, or an operator raises the concurrency limit and \
makes it worse: {refused:?}"
);
let stats = gate.stats();
assert_eq!(stats.refused_memory, 4);
assert_eq!(
stats.refused_at_capacity, 0,
"the limit was 100 and 4 were held"
);
assert_eq!(stats.reserved_bytes, clone * 4);
}
/// **An operation that ERRORS — or panics — still releases both its slot
/// and its reservation.**
///
/// Leak this and the server wedges after N failures: a budget that only
/// comes back on the happy path is a budget that drains to zero the first
/// time a client hangs up mid-clone, which is the single most ordinary
/// thing a git client does.
///
/// # Seen red
///
/// By making `AdmissionGuard::drop` release only on the success path —
/// `if !std::thread::panicking() { … }` around the governor release, which
/// is exactly how a hand-written release-on-completion would behave.
/// Failure:
///
/// ```text
/// 6 operations ended in an error or a panic and 6442450944 bytes are still
/// reserved of 8589934592; the 7th would be refused, so the server has
/// wedged
/// ```
///
/// Three panics, 2 GiB each: three quarters of the whole budget gone to
/// operations that no longer exist.
#[test]
fn an_operation_that_fails_still_gives_its_reservation_back() {
let clone = OpClass::Clone.admission_bytes();
let gate = Arc::new(Admission::with_memory_budget(Some(100), None, clone * 4));
for _ in 0..3 {
// The error path: a guard that goes out of scope through `?`-style
// early return rather than through completion.
let outcome: Result<(), &str> = (|| {
let _guard = gate
.try_admit("solo", "team", OpClass::Clone)
.map_err(|_| "refused")?;
Err("the transport died mid-pack")
})();
assert!(outcome.is_err());
}
for _ in 0..3 {
// The panic path, which no explicit release call survives.
let panicked = std::panic::catch_unwind(std::panic::AssertUnwindSafe(|| {
let _guard = gate
.try_admit("solo", "team", OpClass::Clone)
.expect("admitted before the panic");
panic!("the pack walk blew up");
}));
assert!(panicked.is_err());
}
let stats = gate.stats();
assert_eq!(
stats.reserved_bytes,
0,
"6 operations ended in an error or a panic and {} bytes are still reserved of {}; \
the 7th would be refused, so the server has wedged",
stats.reserved_bytes,
gate.memory_budget().unwrap_or(0)
);
assert_eq!(stats.in_flight, 0);
assert_eq!(
gate.governor().expect("a governor").occupied(),
0,
"the concurrency slots leaked even though the reservations did not"
);
// The applied output: the budget is genuinely reusable, four times over.
let held: Vec<_> = (0..4)
.map(|i| {
gate.try_admit("solo", "team", OpClass::Clone)
.unwrap_or_else(|r| panic!("clone {i} after 6 failures was refused: {r}"))
})
.collect();
assert_eq!(held.len(), 4);
}
/// A transfer that never reports entering service is COUNTED, not guessed
/// at.
///
/// The controller only sees what [`AdmissionGuard::entered_service`] files.
/// A caller that forgets leaves it with no samples, and a CoDel controller
/// with no samples sits at the top of its band forever — a gate that looks
/// like it works and does not. This is the counter that makes that visible,
/// and `tests/admission.rs` is where the real SSH path is held to zero.
///
/// # Seen red
///
/// By making `AdmissionGuard::drop` increment the counter unconditionally
/// instead of only when the mark was never claimed. Failure:
///
/// ```text
/// assertion `left == right` failed: a transfer that DID report entering
/// service was counted as unmeasured: 2
/// left: 2
/// right: 1
/// ```
#[test]
fn a_transfer_that_never_enters_service_is_counted_as_unmeasured() {
let gate = Arc::new(Admission::new(Some(100), None));
drop(
gate.try_admit("solo", "team", OpClass::LsRemote)
.expect("admitted"),
);
assert_eq!(
gate.stats().unmeasured_sojourn,
1,
"a transfer that never said its work started filed no sojourn sample and must be \
counted; a controller starved of samples grows to its ceiling and stays there"
);
let guard = gate
.try_admit("solo", "team", OpClass::LsRemote)
.expect("admitted");
guard.entered_service();
drop(guard);
assert_eq!(
gate.stats().unmeasured_sojourn,
1,
"a transfer that DID report entering service was counted as unmeasured: {}",
gate.stats().unmeasured_sojourn
);
// …and the sample really reached the controller's window, rather than
// merely flipping a flag. One sample opens a window; nothing has closed
// one yet, so the limit has not moved and the class's window is live.
let governor = gate.governor().expect("a governor");
assert_eq!(governor.codel().shrinks() + governor.codel().grows(), 0);
let guard = gate
.try_admit("solo", "team", OpClass::LsRemote)
.expect("admitted");
guard.entered_service();
drop(guard);
assert_eq!(
gate.stats().unmeasured_sojourn,
1,
"the one-shot mark must not turn a second measured transfer into an unmeasured one"
);
}
#[test]
fn an_unlimited_gate_admits_everything_and_still_counts() {
let gate = Arc::new(Admission::unlimited());
let mut held = Vec::new();
for _ in 0..1000 {
held.push(
gate.try_admit("anyone", "team", OpClass::LsRemote)
.expect("no limit means no refusal"),
);
}
let stats = gate.stats();
assert_eq!(stats.admitted, 1000);
assert_eq!(stats.refused(), 0);
assert_eq!(stats.in_flight, 1000);
assert_eq!(stats.peak_in_flight, 1000);
}
}