Skip to main content

gunnar_ssh/
admission.rs

1//! **Admission control for git transfers**: an explicit, countable
2//! ceiling in place of the accidental one.
3//!
4//! # What the accidental ceiling was
5//!
6//! Every git transfer occupies one tokio **blocking-pool** thread for its whole
7//! duration (`gunnar-cli`'s `WireTransport::serve` bridges an async duplex into
8//! `gunnar-wire`'s blocking `BufRead`/`Write` through `SyncIoBridge`). Nothing
9//! bounded how many of those could exist at once, so the real limit was
10//! `tokio`'s default of 512 blocking threads, a number nobody chose. Past it,
11//! transfers queued **silently**: no log, no counter, no error. The failure an
12//! operator saw was "git is slow", with nothing anywhere to look at.
13//!
14//! # Why this refuses instead of queueing
15//!
16//! This gate **refuses** over the limit rather than queueing behind it, and the
17//! reasoning is not "refusing is simpler":
18//!
19//! * **A git client retries; a git client cannot see a queue.** `git` reports
20//!   whatever the server says and the user (or CI) runs it again. A refusal that
21//!   arrives in milliseconds with a sentence in it is strictly better
22//!   information than a socket that goes quiet for a minute.
23//! * **Queueing keeps the expensive things alive.** A queued transfer still
24//!   holds a TCP socket, a russh session, its channel and its key material. That
25//!   is the resource a fork-bomb of clones is trying to exhaust, and a queue
26//!   hands it to them for free.
27//! * **The two existing timeouts fight a queue.** `inactivity_timeout` (600 s)
28//!   and `exec_grace` (60 s) both measure a session that is not making progress.
29//!   A transfer parked in a queue looks exactly like one of those, so a long
30//!   wait ends in a disconnect anyway, after having occupied the slot for the
31//!   whole wait.
32//! * **A queue makes saturation invisible again.** The whole point of the ceiling is
33//!   that the ceiling must be *visible*. A refusal is an event with a number
34//!   attached; a wait is the thing that was already happening.
35//!
36//! So: [`Admission::try_admit`] never blocks and never waits. It either returns
37//! a guard or a [`Refusal`], and both outcomes move a counter.
38//!
39//! # Two limits, because one is not enough
40//!
41//! [`SshConfig::max_concurrent_transfers`](crate::SshConfig::max_concurrent_transfers)
42//! is the server-wide ceiling. On its own it is still exhaustible by a *single*
43//! misbehaving client, which would then deny service to everybody else, so
44//! [`SshConfig::max_transfers_per_principal`](crate::SshConfig::max_transfers_per_principal)
45//! caps what any one identity can hold at once. Identity is the principal id,
46//! not the peer address: gunnar authenticates keys, and a per-IP limit would
47//! punish everyone behind one CI runner's NAT while a botnet walked past it.
48//!
49//! # The per-principal limit is a SHARE, not a wall
50//!
51//! The second limit used to be an unconditional ceiling, and that was wrong in
52//! a way nobody could see from the code. **Measured on a 32-core desktop with
53//! the shipped defaults, 2026-08-03:**
54//!
55//! | load | served | refused |
56//! |---|---|---|
57//! | 1 principal, 64 clients  |  32 | 32 (per-principal) |
58//! | 32 principals, 320 clients | 256 | the rest |
59//!
60//! A client holding **one** key could never reach the advertised 256; that took
61//! eight identities. Every single-team appliance — which is what gunnar is —
62//! silently got one eighth of the capacity written on the box, and the gate that
63//! did it was binding on an **empty server**, where there was nobody to be fair
64//! to.
65//!
66//! The fairness mechanism itself is right: one identity must not be able to take
67//! a shared host away from everybody else. So the ceiling is now **max-min
68//! fair** — it applies only under contention, and what it applies is a share:
69//!
70//! ```text
71//! contended  = occupied (this candidate included) ≥ CONTENTION_PERCENT % of the server-wide limit
72//! ceiling(p) = if !contended { the server-wide limit }
73//!              else          { max(configured, server_wide / active principals) }
74//! admit      = server-wide gate has room  AND  p holds < ceiling(p)
75//! ```
76//!
77//! * **Idle server, one identity** → not contended, so the only ceiling is the
78//!   server-wide one. 64 clients on one key now get 64 slots, and 256 clients
79//!   get 256. That is the half the old gate got wrong.
80//! * **Contended server, many identities** → each gets `server_wide / active`,
81//!   with the configured number as a **floor** nobody drops below however many
82//!   identities show up. A greedy principal is held to its share exactly as
83//!   before.
84//! * **Work-conserving.** Enforcing the share only above the contention
85//!   threshold is what stops "two identities, one of them idle" from costing the
86//!   busy one half the server for no benefit. Below the threshold there are
87//!   free slots, so refusing anything would refuse work the box could do.
88//!
89//! `active principals` is the number of identities holding at least one
90//! transfer *right now* (the candidate included), which is the only demand
91//! signal an admission gate has. It is exact, not sampled: it is the length of
92//! the same map the counts live in, read under the same lock.
93//!
94//! # The ANONYMOUS ceiling: the one caller the share above cannot hold
95//!
96//! Everything above is keyed on a principal id, and that is deliberate: gunnar
97//! authenticates keys, so the fair share divides the server between *identities*
98//! and a greedy one is held to its slice. Anonymous read then opened a door with no key
99//! behind it. Every unauthenticated reader of a public repository arrives as the
100//! single principal [`gunnar_core::ANONYMOUS_PRINCIPAL_ID`] — the id is refused
101//! at registration ([`gunnar_core::Principal::new`]) so nothing else can ever
102//! claim it — which *looks* like the whole internet already sharing one row of
103//! the map above, and therefore already bounded.
104//!
105//! It is not, and the reason is the paragraph before it: **the share is
106//! work-conserving on purpose.** Below [`CONTENTION_PERCENT`] of the governed
107//! limit the share does not bind at all and every principal's ceiling is the
108//! whole server. That rule was measured and is right for a keyed identity —
109//! refusing a lone team its own idle appliance is the exact mistake the 2026-08-03
110//! table above records. Applied to a stranger it says: *an idle box may be taken
111//! entirely by anonymous traffic.* An operator arriving into that does not even
112//! get a clean refusal; they get [`Admission::admit_within`]'s wait, because
113//! nothing evicts a transfer already in flight.
114//!
115//! So anonymous gets a **second, different kind of ceiling**, and the ways it
116//! differs from the per-principal share are the whole design:
117//!
118//! * **Hard, not a share.** It binds on an empty server. There is no
119//!   contention threshold, because the thing being rationed is not fairness
120//!   between identities — it is the blast radius of a caller gunnar cannot name,
121//!   bill, revoke or ring up. Work-conserving is the right instinct for a tenant
122//!   and the wrong one for a stranger.
123//! * **Small in absolute terms**, not a fraction. Five
124//!   ([`DEFAULT_MAX_ANONYMOUS_TRANSFERS`]) against a keyed floor of 32 and a
125//!   server-wide 256. Anonymous can therefore never be more than ~2 % of the
126//!   box, which is what makes "an operator is never held up by an anonymous
127//!   flood" a *structural* statement rather than a hope about timing.
128//! * **Refused instantly, never parked.** [`Admission::admit_within`]'s wait
129//!   exists because a refused keyed client reconnects, and one parked future is
130//!   cheaper than that reconnect. For anonymous the arithmetic inverts: parking
131//!   holds the TCP socket, the russh session and its key material, which is
132//!   precisely the resource a clone fork-bomb is trying to exhaust. So an
133//!   `AnonymousAtCapacity` refusal is final at the first ask.
134//! * **Counted separately** ([`AdmissionStats::refused_anonymous`]). Folded into
135//!   `refused_per_principal` it would be indistinguishable from a CI runner
136//!   hitting its share, and the two demand opposite operator actions: one says
137//!   *raise the limit*, the other says *you are being scraped*.
138//!
139//! ## The keyed reserve: the other half of the same door (D23b)
140//!
141//! The ceiling above is **absolute**, and the box it is a share of is **not**.
142//! [`Governor`]'s CoDel controller sheds the governed limit down its band
143//! whenever the machine is queueing, and the shed was measured reaching a
144//! limit of 19. Five out of 19 is a quarter of the machine, held by a caller
145//! that cannot be named — and held at exactly the moment there is least to go
146//! round. The `~2 %` in the bullet above is a statement about an *unshed* box.
147//! The same arithmetic undoes an operator who raised the ceiling on purpose for
148//! a browse replica: the number they raised bears no relation to the number the
149//! controller is enforcing an hour later.
150//!
151//! So the top of the box is **keyed-only**. An anonymous read is admitted only
152//! while occupancy is *below* [`CONTENTION_PERCENT`] of the governed limit;
153//! from there up it is refused with [`Refusal::KeyedReserve`], instantly, and
154//! counted as [`AdmissionStats::refused_keyed_reserve`]. What that reserves is
155//! [`Admission::keyed_reserve`] slots — 65 of the shipped 256, 2 of a shed 4 —
156//! and a keyed caller can therefore never be refused *by a stranger's
157//! acquisition*.
158//!
159//! Three things about it are decisions rather than defaults:
160//!
161//! * **The threshold is `CONTENTION_PERCENT`, not a fourth number.** That
162//!   constant is *defined* as the occupancy at which gunnar starts asking keyed
163//!   identities to share with each other. The moment it does that is exactly the
164//!   moment it should stop asking them to share with strangers. A second
165//!   threshold would be a second opinion about when the box is busy, and the
166//!   order between the two would be an accident of arithmetic.
167//! * **It refuses an acquisition; it does not evict.** Taking a slot back from
168//!   a transfer already running throws away every byte of pack already built,
169//!   hands a legitimate visitor a torn stream that `git` reports as a network
170//!   error (indistinguishable from a broken repository), and does not even
171//!   reliably give the slot to the keyed caller that triggered it — the
172//!   `slot_freed` wakeup goes to whoever is parked. It would make a
173//!   stranger cost *more* than serving it, because the evicted client comes back
174//!   and redoes the work. Refusing before the first byte costs one
175//!   compare-and-swap.
176//! * **It is enforced inside the claim's compare-and-swap**
177//!   ([`Governor::claim_reserving`](crate::governor::Governor::claim_reserving)),
178//!   not checked around it. Both orderings of "read occupancy, then claim" were
179//!   measured and both leak — the failures are quoted in
180//!   `tests/keyed_reserve.rs`. A limit checked outside the atomic that enforces
181//!   it is not a limit.
182//!
183//! ### What that does and does not promise
184//!
185//! Stated exactly, because the exact statement is weaker than the slogan and the
186//! difference is where somebody would otherwise be surprised. A stranger
187//! admitted onto a *quiet* box is still holding its slot if the box fills a
188//! moment later, and this gate will not take it back. So "a keyed caller is
189//! never refused while a stranger holds anything" is **not** true, and the only
190//! way to make it true is the eviction ruled out above.
191//!
192//! What is true is a **ratchet**. From the moment occupancy reaches
193//! `CONTENTION_PERCENT`, no stranger is admitted again, so anonymous occupancy
194//! from there on only ever falls: every anonymous transfer that finishes hands
195//! its slot to the keyed side and cannot take it back. A flood therefore cannot
196//! hold a busy box at its anonymous maximum — it decays to zero at the rate
197//! transfers complete, and `SshConfig::inactivity_timeout` bounds that rate from
198//! above. The residue is bounded too, at `min(anonymous ceiling, limit −
199//! reserve)`, which on the shipped appliance is 5 of 256. That is the number a
200//! keyed caller could in the worst case be queued behind, and it is the number
201//! the ceiling was chosen to make small.
202//!
203//! What anonymous keeps is therefore the *whole* allowance on a box that is not
204//! busy, which is the ordinary state of an appliance and the only state a public
205//! repository is interesting in. If keyed demand is permanently above
206//! `CONTENTION_PERCENT` the public door is permanently shut — and that is
207//! accepted, deliberately: a box that is three-quarters full of paid-for work
208//! has nothing to spend on a stranger, the refusal is instant and names its own
209//! rule so the operator can see it happening, and the two answers (a bigger
210//! limit, or a browse replica) are both things an operator does rather than
211//! things gunnar should guess. It is not a livelock: nothing is parked, nothing
212//! is retried on gunnar's side, and a stranger costs a string compare.
213//!
214//! ## Why there is no priority WAKE order
215//!
216//! The obvious third mechanism — when a slot frees, wake keyed waiters before
217//! anonymous ones — is not implemented because it would be a no-op.
218//! [`Admission::admit_within`]'s queue **has no anonymous waiters in it**: every
219//! refusal that names anonymity short-circuits the wait, so the only callers
220//! parked on the gate's own `slot_freed` notify are keyed ones already. Adding
221//! a priority queue would mean adding a data structure and a lock in order to
222//! sort a queue with one class in it.
223//!
224//! ## Why not queue the overflow instead
225//!
226//! Same answer as the module's opening section, only more so. A bounded queue
227//! of anonymous readers is a bounded pile of held sockets, and an unbounded one
228//! is a slower outage with worse diagnostics. Refusing in milliseconds with a
229//! sentence in it is the strictly better outcome for a `git` that will retry
230//! anyway, and it is the only outcome that keeps the cost of an anonymous
231//! request proportional to what gunnar is willing to spend on it.
232//!
233//! ## Fairness *among* anonymous readers
234//!
235//! There is none, and there cannot be: telling two strangers apart requires an
236//! identity, and not having one is the definition of this caller. What bounds a
237//! single greedy anonymous client is therefore not a share but the **time** any
238//! one permit can be held — which is already bounded by
239//! `SshConfig::inactivity_timeout` and by the transfer finishing. The ceiling's
240//! job is to cap the total, and the existing timeouts recycle it.
241//!
242//! An operator who wants the old behaviour — a hard per-identity cap that binds
243//! whether or not anyone else is there, e.g. a multi-tenant host that bills by
244//! identity — sets
245//! [`SshConfig::per_principal_limit_is_hard`](crate::SshConfig::per_principal_limit_is_hard).
246//! The configured numbers keep their meaning either way.
247//!
248//! # Saturation is observable, by construction
249//!
250//! [`AdmissionStats`] is a snapshot of five monotonic-or-live counters, reachable
251//! from [`BoundServer::admission`](crate::BoundServer::admission). Every refusal
252//! is also logged at `warn` with the limit that was hit and what was in flight.
253//! `tests/admission.rs` drives a server past its cap and asserts the counter
254//! moved, not merely that nothing crashed.
255//!
256//! # The server-wide limit is a GOVERNOR now, not a constant
257//!
258//! `max_concurrent_transfers` used to be the number of permits in a semaphore,
259//! and a fixed number is wrong in both directions at once: too high and the box
260//! is already thrashing before anything is refused, too low and it refuses work
261//! it could have done. It is also unknowable in advance — the right number
262//! depends on the repositories, the disks and the other tenants.
263//!
264//! So the constant is **replaced** — not joined by a second mechanism — with
265//! [`crate::governor::Governor`], and the operator's number becomes the band's
266//! ceiling rather than the limit itself. Two gates, in this order, both of them
267//! *at admission* and before anything has been allocated:
268//!
269//! 1. **time** — a CoDel controller that measures how long operations *waited*,
270//!    per class, and moves one shared limit;
271//! 2. **space** — a memory reservation the server owns, rather than a
272//!    measurement of the machine it is taken from.
273//!
274//! `crate::governor` sets out why sojourn rather than queue length, why the
275//! minimum rather than the average, why the target is per class, and what the
276//! memory estimate does *not* yet bound. Everything below — the per-principal
277//! max-min share, the counters, the refusal messages — is unchanged and now
278//! divides the *governed* limit instead of the configured one, which is the
279//! right divisor: a share of capacity the server does not have is not a share.
280//!
281//! # The usage share (`admit-share`): a class from `edda_admit`, a share from here
282//!
283//! With the `admit-share` feature the gate asks
284//! [`edda_admit::Admit::plan`] for every **keyed** transfer, with the
285//! namespace's rolling-window bytes out of the flow ledger
286//! ([`gunnar_core::FlowLedger::window_bytes`]) as the usage. The plan's class
287//! maps onto a share this gate already has:
288//!
289//! * `Full` → the per-principal max-min share, exactly as above;
290//! * `Reduced` / `Floor` → **the anonymous ceiling**: a hard cap that binds on
291//!   an idle box, plus the keyed reserve. An over-budget namespace is still
292//!   served — at a stranger's width — and refused FAST, with
293//!   [`Refusal::ReducedShare`], never parked in [`Admission::admit_within`]'s
294//!   wait: the same reasoning that refuses anonymous instead of queueing it
295//!   applies to anybody held to the anonymous share.
296//!
297//! `record` is fed when the guard drops, with the transfer's wire bytes in each
298//! direction, so the estimate `Usage.AdmitPlan` shows learns per kind. The
299//! policy's own slot shares are not configured here — this gate is the share.
300//!
301//! **Without the feature there is no policy field, no call and no branch on
302//! the path**: the admission decision is today's, and `tests/admission.rs`,
303//! `anonymous_ceiling.rs` and `keyed_reserve.rs` run unchanged to prove it.
304//!
305//! # What this is NOT
306//!
307//! It is not intra-request parallelism, and it is deliberately the *opposite*
308//! decision. gatling has no shared pool, so fanning a clone out inside the
309//! request at forge concurrency spawns tens of thousands of OS threads, and
310//! the measured deflate curve is already
311//! *slower* at 64 threads than at 32. Throughput at the load target comes from
312//! serving many requests at once, each costing about one core, which is exactly
313//! what this gate meters.
314
315use std::collections::HashMap;
316use std::sync::atomic::{AtomicU64, Ordering};
317use std::sync::{Arc, Mutex};
318
319use std::time::Duration;
320
321use crate::governor::{
322    Claim, Governor, Growth, Lane, LimitBand, OpClass, ServiceMark, Shed, DEFAULT_ADMISSION_WAIT,
323    DEFAULT_MEMORY_BUDGET_BYTES, WIRE_MEMORY_DENOMINATOR, WIRE_MEMORY_NUMERATOR, WIRE_RESERVE_STEP,
324};
325
326/// Why a transfer was not admitted.
327///
328/// Every variant carries the number that was hit, because a refusal an operator
329/// cannot tie to a number is a mystery rather than a signal.
330#[derive(Debug, Clone, Copy, PartialEq, Eq)]
331pub enum Refusal {
332    /// The server-wide concurrent-transfer limit is full.
333    AtCapacity {
334        /// The server-wide limit **in force at the moment of the refusal**.
335        ///
336        /// The *governed* limit, not the configured one: the configured number
337        /// is the ceiling of
338        /// [`LimitBand`](crate::governor::LimitBand), and what actually refuses
339        /// is wherever the CoDel controller has moved to inside it. Reported
340        /// rather than assumed, for the same reason the per-principal share is:
341        /// an operator who read one number out of a config file and a different
342        /// one out of a log would conclude the gate was broken.
343        limit: usize,
344    },
345    /// This principal already holds its share of the server-wide limit.
346    PrincipalAtCapacity {
347        /// The per-principal limit **in force at the moment of the refusal**.
348        ///
349        /// Under [`SshConfig::per_principal_limit_is_hard`] that is the
350        /// configured number. Otherwise it is the max-min share —
351        /// `max(configured, server_wide / active)` — which is why it is
352        /// reported rather than assumed: an operator who read the configured
353        /// number out of a file and a different number out of a log would
354        /// conclude the gate was broken.
355        ///
356        /// [`SshConfig::per_principal_limit_is_hard`]: crate::SshConfig::per_principal_limit_is_hard
357        limit: usize,
358        /// How many identities held a transfer when this one was refused.
359        ///
360        /// `1` means the refusal came from a hard configured cap: a share is
361        /// never smaller than the whole server when there is only one claimant.
362        /// Without this an operator cannot tell "you are being throttled
363        /// because the box is shared" from "you are being throttled because
364        /// somebody pinned a number".
365        active_principals: usize,
366    },
367    /// This operation's **lane** is full while the rest of the server is held
368    /// by the other lanes' floors (gate 1b, [`Lane::floor`]).
369    ///
370    /// Its own variant because the operator action is different: "the box is
371    /// full" says raise the budget, and this one says a *write* storm is being
372    /// held to five eighths of the box on purpose so that reads still land.
373    LaneAtCapacity {
374        /// The lane that was full.
375        lane: Lane,
376        /// What that lane may hold right now.
377        ceiling: usize,
378        /// The governed server-wide limit it is a share of.
379        limit: usize,
380    },
381    /// The **anonymous** caller is already holding its whole ceiling.
382    ///
383    /// Its own variant, and not [`Refusal::PrincipalAtCapacity`] with the
384    /// anonymous id in it, because the operator action is different in kind. A
385    /// per-principal refusal names a client that can be found, telephoned or
386    /// given a bigger share; this one says *strangers are at their cap*, which
387    /// is either normal (a popular public repository) or the first sign of a
388    /// scrape, and in neither case is the answer "raise the concurrency limit".
389    /// It also carries no `active_principals`, because there is exactly one
390    /// anonymous principal and dividing anything by it would be theatre.
391    AnonymousAtCapacity {
392        /// The anonymous ceiling in force, i.e.
393        /// [`Admission::max_anonymous`]. Unlike every other limit reported
394        /// here this one is *not* governed or shared — it is the operator's
395        /// number, unmodified — which is exactly why it is worth printing:
396        /// an operator who sees a refusal against 5 on a box with 200 free
397        /// slots must be able to see immediately that a different knob did it.
398        limit: usize,
399    },
400    /// The stranger asked for a slot inside the part of the box that only a
401    /// **keyed** identity may hold (D23b).
402    ///
403    /// Its own variant, and specifically **not**
404    /// [`Refusal::AnonymousAtCapacity`], because the two say opposite things
405    /// about the machine and are fixed by opposite knobs. `AnonymousAtCapacity`
406    /// means *strangers are at their cap* — it can fire on an idle box, and the
407    /// only number that changes it is `--max-anonymous-transfers`. This one
408    /// means *the box is too busy to spend any of itself on a stranger*: the
409    /// anonymous ceiling is nowhere near reached, and lowering it would do
410    /// nothing at all. The knob here is the server-wide limit, or the size of
411    /// the box, or a browse replica.
412    ///
413    /// It also carries no principal and no share, because it is not a statement
414    /// about this caller at all: it is a statement about occupancy, and the only
415    /// thing about the caller that entered into it was that it has no key.
416    KeyedReserve {
417        /// Slots occupied as the gate's own compare-and-swap observed them —
418        /// the reserve rides inside that atomic
419        /// ([`Governor::claim_reserving`](crate::governor::Governor::claim_reserving)),
420        /// so this is the number that actually made the decision and not a
421        /// second reading taken nearby. Reported so an operator can see the
422        /// decision rather than reconstruct it: `occupied + reserve >= limit`
423        /// is the whole rule, visible in one line of log.
424        occupied: usize,
425        /// The **governed** server-wide limit those slots are out of, i.e.
426        /// wherever the CoDel controller has moved to inside its band. The
427        /// configured number would be actively misleading here: the whole
428        /// reason this gate exists is that the governed limit moves and the
429        /// anonymous ceiling does not.
430        limit: usize,
431        /// How many slots at the top of that limit are keyed-only, i.e.
432        /// [`Admission::keyed_reserve`]. Printed because it is derived
433        /// ([`CONTENTION_PERCENT`]) and therefore appears in no config file an
434        /// operator could look it up in.
435        reserve: usize,
436    },
437    /// The server's **memory budget** is already spoken for (gate 2).
438    ///
439    /// A third reason, kept separate from [`Refusal::AtCapacity`] on purpose: an
440    /// operator who reads "at capacity" raises the concurrency limit, and doing
441    /// that here makes the problem worse. The numbers travel so the log says
442    /// which knob to turn.
443    MemoryBudget {
444        /// The class whose estimate did not fit.
445        class: OpClass,
446        /// What [`OpClass::estimated_bytes`] asked for.
447        wanted: u64,
448        /// What was reserved by operations already in flight.
449        reserved: u64,
450        /// The configured budget.
451        budget: u64,
452    },
453    /// **The gate is PAUSED** — every transfer is refused, not for want of a
454    /// slot but because the server asked for a quiet moment: the runtime
455    /// data-set growth drains the box before it unmounts `/data`
456    /// (`DATA-SET-GROWTH-FLOW.md` §3 step 5, T1/T12). Carries no number
457    /// because no number was hit; carries the reason, whose words the client
458    /// reads. Set and cleared by [`Admission::pause`] / [`Admission::unpause`]
459    /// — one gate, so all three git fronts pause together.
460    Paused {
461        /// Why, and in which words.
462        pause: Pause,
463    },
464    /// **This namespace is over its usage budget** (`admit-share`), so the
465    /// transfer was held to the anonymous share and that share was full.
466    ///
467    /// Its own variant, and neither [`Refusal::AnonymousAtCapacity`] nor
468    /// [`Refusal::KeyedReserve`], because the caller HAS a key and the operator
469    /// action is different again: the tenant's budget (`--admit-limit`), or the
470    /// tenant waiting for the window to move. The plan's sentence ("100 GB/30d
471    /// used 112 GB; this clone ~1.2 GB; Reduced") is in the `warn` row beside it.
472    #[cfg(feature = "admit-share")]
473    ReducedShare {
474        /// The anonymous ceiling the namespace is held to.
475        ceiling: usize,
476        /// `true` when the box was past `CONTENTION_PERCENT` and the keyed
477        /// reserve refused it; `false` when its own ceiling was full.
478        busy: bool,
479    },
480}
481
482/// **Why the admission gate is paused** — the reason, and the sentence the
483/// client is handed for it. An enum rather than a `String` so [`Refusal`]
484/// stays `Copy` and the words are fixed in one place, where a test can hold
485/// them.
486#[derive(Debug, Clone, Copy, PartialEq, Eq)]
487pub enum Pause {
488    /// The data set is being grown at runtime: drained, unmounted, grown,
489    /// mounted again — minutes, and the box stays up throughout.
490    Growing,
491}
492
493impl Pause {
494    /// The client's sentence. **The word `growing` is load-bearing** the way
495    /// `starting` is in [`crate::STARTING_MESSAGE`]: it is what tells a
496    /// person, a CI script and the bench that this is a planned pause with an
497    /// end, not a broken repository and not a full box.
498    pub const fn words(self) -> &'static str {
499        match self {
500            Pause::Growing => GROWING_MESSAGE,
501        }
502    }
503    /// The log's `reason` field.
504    pub const fn reason(self) -> &'static str {
505        match self {
506            Pause::Growing => "paused-growing",
507        }
508    }
509}
510
511/// What a client is told while the data set grows. The contract's words
512/// (`DATA-SET-GROWTH-FLOW.md` §0, ruling 1; the drain design note in
513/// `.nornir/data-set-grow-drain.md`), verbatim.
514///
515/// It names no estate: the same build serves gunnar.rs, test2 and every sold estate, and a
516/// test2 push told "gunnar.rs is growing" was told about somebody else's server (2026-10-09).
517pub const GROWING_MESSAGE: &str = "The server is growing its storage to make room for new \
518                                   members. Pushes are paused — please try again in a few \
519                                   minutes.";
520
521impl Refusal {
522    /// What the client is told, in its own protocol.
523    ///
524    /// Says *try again* explicitly. "Server busy" with no instruction reads as a
525    /// permanent failure to a human and to a CI log alike, and this one is
526    /// transient by definition.
527    pub fn client_message(&self) -> String {
528        match self {
529            Refusal::AtCapacity { limit } => format!(
530                "the server is already serving its limit of {limit} concurrent git transfers; \
531                 try again in a moment"
532            ),
533            // Both arms must keep the words "per-principal limit", and neither
534            // may contain "concurrent git transfers": `installtest`'s
535            // `refusals_in_log` tells the two refusals apart by exactly those
536            // substrings, because the log records the refusal by its `Display`
537            // — the same sentence the client is handed. Two witnesses of one
538            // event only agree if they are reading the same string.
539            Refusal::PrincipalAtCapacity {
540                limit,
541                active_principals,
542            } if *active_principals > 1 => format!(
543                "you already have {limit} git transfers in flight, which is your fair share of \
544                 this server while {active_principals} identities are using it; that is the \
545                 per-principal limit in force right now, and it rises again as the server \
546                 empties; try again in a moment"
547            ),
548            Refusal::PrincipalAtCapacity { limit, .. } => format!(
549                "you already have {limit} git transfers in flight on this server, which is the \
550                 per-principal limit; try again in a moment"
551            ),
552            // Shares neither "concurrent git transfers" nor "per-principal
553            // limit": `installtest`'s `refusals_in_log` tells refusals apart by
554            // exactly those, and a fourth reason that collided with either
555            // would be counted as the wrong one.
556            Refusal::LaneAtCapacity {
557                lane,
558                ceiling,
559                limit,
560            } => format!(
561                "this server is holding {lane} operations to {ceiling} of its {limit} slots so \
562                 that the other kinds of request still land; try again in a moment"
563            ),
564            // Shares neither "concurrent git transfers" nor "per-principal
565            // limit", for the reason given twice above. It says *unauthenticated*
566            // in the first clause so a human reading it in a `git` output knows
567            // instantly that the fix is a key, not patience: this is the one
568            // refusal on the box that a caller can make go away by logging in.
569            Refusal::AnonymousAtCapacity { limit } => format!(
570                "this server is already serving {limit} unauthenticated readers, which is all it \
571                 offers without a key; try again in a moment, or push a key and clone as \
572                 yourself"
573            ),
574            // Shares neither "concurrent git transfers" nor "per-principal
575            // limit", for the reason given twice above — and it must not be the
576            // same sentence as `AnonymousAtCapacity` either, because the two
577            // are fixed by different knobs. It says *with a key* so the human
578            // holding the `git` prompt learns the same thing the operator does:
579            // this refusal is about not having one.
580            Refusal::KeyedReserve {
581                occupied,
582                limit,
583                reserve,
584            } => format!(
585                "this server is {occupied} of {limit} slots busy and keeps its last {reserve} for \
586                 callers with a key; try again in a moment, or push a key and clone as yourself"
587            ),
588            // Deliberately shares neither of the two substrings above:
589            // `installtest`'s `refusals_in_log` tells refusals apart by
590            // "concurrent git transfers" and "per-principal limit", and a third
591            // reason that collided with either would be counted as the wrong
592            // one. It says *memory* so the operator raises the budget rather
593            // than the concurrency limit.
594            Refusal::MemoryBudget {
595                class,
596                wanted,
597                reserved,
598                budget,
599            } => format!(
600                "this server holds a memory budget of {} MiB for git operations in flight, {} MiB \
601                 of it is reserved, and a {class} needs {} MiB more than is left; try again in a \
602                 moment",
603                budget / (1024 * 1024),
604                reserved / (1024 * 1024),
605                wanted / (1024 * 1024),
606            ),
607            // Its own words entirely (see `Pause::words`): none of the three
608            // substrings the other refusals are told apart by, and a
609            // sentence that says WHY and for how long.
610            Refusal::Paused { pause } => pause.words().to_owned(),
611            // Shares neither "concurrent git transfers" nor "per-principal
612            // limit" (`installtest`'s `refusals_in_log`), and says *budget* so
613            // the human at the prompt learns it is about their usage, not the box.
614            #[cfg(feature = "admit-share")]
615            Refusal::ReducedShare { ceiling, busy } => {
616                format!(
617                "this namespace is over its transfer budget for the window, so it is served at \
618                 the reduced share of {ceiling} git transfers at once{}; try again in a moment, \
619                 or ask for a larger budget",
620                if *busy { ", and none while the server is this busy" } else { "" }
621            )
622            }
623        }
624    }
625
626    /// The **machine-readable** reason, for a log's `reason` field.
627    ///
628    /// The sentence above is for the person holding the `git` prompt; this is
629    /// for the operator filtering a log, and they need different things. Until
630    /// the anonymous ceiling existed every refusal here was fairly described by
631    /// one string — the box was out of *something* server-wide — and both
632    /// fronts hard-coded `admission-gate-full` accordingly. That string is now
633    /// actively misleading on the one refusal that can fire with the box nearly
634    /// empty: an operator who greps `admission-gate-full`, sees a hit, and
635    /// looks at 59 free slots concludes the gate is broken, which is exactly
636    /// the "the limit exists but produces no signal an operator can act on"
637    /// failure the unbounded listener was about, wearing a different hat.
638    ///
639    /// Only the fronts that can actually *see* an anonymous caller use this.
640    /// The gRPC control plane keeps its literal, because
641    /// [`gunnar_grpc::auth`] refuses the anonymous principal outright: a
642    /// `reason` there that could report anonymity would be describing a state
643    /// that crate makes unreachable.
644    pub const fn reason(&self) -> &'static str {
645        match self {
646            Refusal::AtCapacity { .. } => "admission-gate-full",
647            Refusal::LaneAtCapacity { .. } => "admission-lane-full",
648            Refusal::PrincipalAtCapacity { .. } => "admission-principal-share",
649            Refusal::AnonymousAtCapacity { .. } => "admission-anonymous-ceiling",
650            Refusal::KeyedReserve { .. } => "admission-keyed-reserve",
651            Refusal::MemoryBudget { .. } => "admission-memory-budget",
652            Refusal::Paused { pause } => pause.reason(),
653            #[cfg(feature = "admit-share")]
654            Refusal::ReducedShare { .. } => "admission-reduced-share",
655        }
656    }
657}
658
659impl std::fmt::Display for Refusal {
660    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
661        f.write_str(&self.client_message())
662    }
663}
664
665/// A snapshot of the gate's counters.
666///
667/// Taken atomically field by field rather than under a lock: these are a
668/// **signal**, and a signal that costs a mutex on the request path would be
669/// paid for by the thing it is measuring.
670#[derive(Debug, Clone, Copy, PartialEq, Eq, Default)]
671pub struct AdmissionStats {
672    /// Transfers let through since the server started.
673    pub admitted: u64,
674    /// Transfers refused because the server-wide limit was full.
675    pub refused_at_capacity: u64,
676    /// Transfers refused because one principal held its whole share.
677    pub refused_per_principal: u64,
678    /// Transfers refused because the **memory budget** was spoken for (gate 2).
679    pub refused_memory: u64,
680    /// Reads refused because the **anonymous ceiling** was full.
681    ///
682    /// Deliberately not folded into
683    /// [`refused_per_principal`](Self::refused_per_principal): that counter
684    /// moving means a client gunnar knows about is being throttled, and this one
685    /// moving means strangers are at their cap. A dashboard that added them
686    /// together would show one number that can be entirely benign (a popular
687    /// public repository, working exactly as designed) or entirely alarming (a
688    /// keyed CI fleet starved of its share) with no way to tell which.
689    pub refused_anonymous: u64,
690    /// Reads refused because the slot they wanted was inside the **keyed
691    /// reserve** (D23b).
692    ///
693    /// Deliberately not folded into
694    /// [`refused_anonymous`](Self::refused_anonymous), which is the other half
695    /// of the same door and means the opposite thing. `refused_anonymous`
696    /// moving on an otherwise quiet box is *normal* — a popular public
697    /// repository, the ceiling doing its job — and the operator response is
698    /// none. This counter moving means the box was over
699    /// [`CONTENTION_PERCENT`] full of real work at the moment a stranger
700    /// asked, which is a **capacity** signal: it is the public door being shut
701    /// by load, and it will keep being shut until either the load falls or the
702    /// box grows. Summed together the two would show one number that means
703    /// "everything is fine" and "your public repositories are effectively
704    /// offline" with no way to tell which.
705    pub refused_keyed_reserve: u64,
706    /// Transfers refused at the **reduced share** (`admit-share`): a namespace
707    /// over its usage budget, held to the anonymous width.
708    #[cfg(feature = "admit-share")]
709    pub refused_reduced: u64,
710    /// Transfers refused because the gate was **paused** ([`Pause`]): the
711    /// data set was being grown. Its own counter, because this one moving is
712    /// neither load nor strangers — it is a planned window, and an operator
713    /// reading it wants to know how many clients met the window, not to
714    /// raise any limit.
715    pub refused_paused: u64,
716    /// Transfers in flight right now.
717    pub in_flight: u64,
718    /// The most that were ever in flight at once. This is the number that says
719    /// how close to the ceiling a server has actually been; `in_flight` alone
720    /// never shows it, because a sampler always misses the peak.
721    pub peak_in_flight: u64,
722    /// The **governed** server-wide limit right now, i.e. where the CoDel
723    /// controller has moved to inside its band. `0` when there is no limit.
724    ///
725    /// Without this the shed is invisible: an operator seeing refusals against a
726    /// configured 100 has no way to know the governor has taken it to 37, which
727    /// is the same "the limit exists but produces no signal" the unbounded listener was
728    /// written about.
729    pub limit: usize,
730    /// Bytes reserved by operations in flight, against
731    /// [`memory_budget`](Admission::memory_budget).
732    pub reserved_bytes: u64,
733    /// Operations that were admitted only after **waiting** for a slot
734    /// ([`DEFAULT_ADMISSION_WAIT`]).
735    ///
736    /// This is the queue made countable. It is not a failure count: every one
737    /// of these was served, and every one of them would have been a refusal
738    /// before 2026-08-30.
739    pub admitted_after_wait: u64,
740    /// Milliseconds spent waiting, summed over every operation that waited.
741    pub waited_ms_total: u64,
742    /// The longest single admission wait, in milliseconds. The number that says
743    /// whether the bound is doing anything.
744    pub peak_wait_ms: u64,
745    /// Waits that ran out and became refusals.
746    pub refused_after_wait: u64,
747    /// Refusals because the operation's **lane** was at its ceiling while the
748    /// other lanes' floors held the rest.
749    pub refused_lane: u64,
750    /// Transfers whose reservation grew past the memory budget while they ran.
751    /// Nonzero means one transfer is now keeping new ones out — which is the
752    /// budget working, not failing.
753    pub overdrawn: u64,
754    /// Bytes charged by running transfers over and above their admission
755    /// reservations.
756    pub grown_bytes: u64,
757    /// Milliseconds between admission and the transfer's work actually
758    /// starting, summed over every transfer that reported it.
759    ///
760    /// **Observed, not controlled.** Until 2026-08-30 this gap *was* the CoDel
761    /// controller's input, and that was the defect: it is the tokio ready queue
762    /// plus the blocking-pool handoff, which is set by how many clients are
763    /// connected and doing crypto, not by how many transfers are admitted. A
764    /// refused client reconnects, so shedding made this number worse and the
765    /// controller shed until it hit its floor — the `limit of 19`
766    /// measured. It is kept as a *counter* because it is a real and useful
767    /// symptom of a box that is out of CPU; it is no longer a control input.
768    pub handoff_ms_total: u64,
769    /// The longest single admission-to-work gap, in milliseconds.
770    pub peak_handoff_ms: u64,
771    /// Transfers that finished **without ever reporting that their work
772    /// started**, so they contributed no sojourn sample.
773    ///
774    /// A blindness counter, not a health one. The CoDel controller only sees
775    /// what [`AdmissionGuard::entered_service`] files; a caller that forgets to
776    /// call it leaves the controller with no samples at all, and a controller
777    /// with no samples grows to the top of its band and stays there — a gate
778    /// that looks like it is working and is not. This number must be 0 on any
779    /// path that is actually governed.
780    pub unmeasured_sojourn: u64,
781}
782
783impl AdmissionStats {
784    /// Every refusal, however it was refused.
785    pub fn refused(&self) -> u64 {
786        self.refused_at_capacity
787            + self.refused_lane
788            + self.refused_per_principal
789            + self.refused_memory
790            + self.refused_anonymous
791            + self.refused_keyed_reserve
792            + self.refused_paused
793            + self.refused_reduced()
794    }
795
796    #[cfg(feature = "admit-share")]
797    fn refused_reduced(&self) -> u64 {
798        self.refused_reduced
799    }
800
801    #[cfg(not(feature = "admit-share"))]
802    fn refused_reduced(&self) -> u64 {
803        0
804    }
805
806    /// The mean admission wait over the operations that waited, in
807    /// milliseconds. `0` when nothing has ever waited.
808    pub fn mean_wait_ms(&self) -> u64 {
809        self.waited_ms_total
810            .checked_div(self.admitted_after_wait.max(1))
811            .unwrap_or(0)
812    }
813}
814
815#[derive(Debug, Default)]
816struct Counters {
817    admitted: AtomicU64,
818    refused_at_capacity: AtomicU64,
819    refused_lane: AtomicU64,
820    refused_per_principal: AtomicU64,
821    refused_memory: AtomicU64,
822    /// Reads refused by the anonymous ceiling.
823    refused_anonymous: AtomicU64,
824    /// Reads refused by the keyed reserve (D23b).
825    refused_keyed_reserve: AtomicU64,
826    /// Transfers refused while the gate was paused ([`Pause`]).
827    refused_paused: AtomicU64,
828    /// Transfers refused at the reduced share (`admit-share`).
829    #[cfg(feature = "admit-share")]
830    refused_reduced: AtomicU64,
831    in_flight: AtomicU64,
832    peak_in_flight: AtomicU64,
833    unmeasured_sojourn: AtomicU64,
834    /// Admissions that had to wait for a slot at all.
835    admitted_after_wait: AtomicU64,
836    /// Milliseconds spent waiting, summed over every operation that waited.
837    waited_ms_total: AtomicU64,
838    /// The longest single wait, in milliseconds.
839    peak_wait_ms: AtomicU64,
840    /// Waits that ran out and became refusals.
841    refused_after_wait: AtomicU64,
842    /// Transfers whose growing reservation went past the budget
843    /// ([`Growth::Overdrawn`]).
844    overdrawn: AtomicU64,
845    /// Bytes charged to the ledger by running transfers, over and above their
846    /// admission reservations.
847    grown_bytes: AtomicU64,
848    /// Milliseconds between admission and the transfer's work starting, summed.
849    handoff_ms_total: AtomicU64,
850    /// The longest single admission-to-work gap, in milliseconds.
851    peak_handoff_ms: AtomicU64,
852}
853
854/// How full the server-wide gate must be, in percent, before the per-principal
855/// share is enforced at all.
856///
857/// **Chosen from the measurement, not picked round.** With the shipped defaults
858/// (256 server-wide, 32 per principal) 75 % is 192 in flight, which leaves 64
859/// free slots — exactly **twice** the configured per-principal floor — as
860/// headroom in which a newly arriving identity is never asked about fairness.
861/// Two floors rather than one, so a second identity can arrive *and* be
862/// throttled to its floor without the first having to give anything back.
863///
864/// Below the threshold there are free slots by definition, and refusing a
865/// transfer there refuses work the box could actually do: that is the exact
866/// mistake the old unconditional cap made, where 32 clients on one key were
867/// refused against 224 idle slots. 192 in flight is also already 6× the core
868/// count of the 32-core box the defaults were sized for
869/// (Measurement 3 — 63 % efficiency at 32
870/// threads), so the machine is deep into queueing before fairness engages, and
871/// fairness is only interesting once queueing has started.
872///
873/// It is deliberately **not** configurable. A third number to get wrong is how
874/// the first two ended up unexamined for four sessions; the two an operator
875/// sizes are the limit and the floor.
876pub const CONTENTION_PERCENT: usize = 75;
877
878/// **How many git reads the server will serve without a key at all.**
879///
880/// Five. Chosen small and chosen *absolute*, and both halves are the decision:
881///
882/// * **Small.** Against the shipped 256 server-wide this is under 2 % of the
883///   box. That number is not a guess at demand, it is a statement about
884///   *liability*: it is the most an operator can lose to a caller who cannot be
885///   named, revoked, billed or telephoned. Everything above five anonymous
886///   readers is served by them coming back a moment later, which git clients do
887///   by themselves; nothing above five is worth an operator's console being
888///   slow.
889/// * **Absolute, not a fraction of the limit.** A percentage would grow the
890///   stranger's allowance every time an operator bought a bigger box, which is
891///   exactly backwards — a bigger appliance is bought to serve *the team*
892///   better, and the untrusted share should stay where it was. It also keeps
893///   this ceiling meaningful when the CoDel controller has shed the governed
894///   limit right down: a fraction of a shrinking limit shrinks with it, and the
895///   moment the box is in trouble is the moment this cap most needs to hold
896///   still.
897///
898/// It is far below [`DEFAULT_MAX_TRANSFERS_PER_PRINCIPAL`] (32), which is the
899/// number *one keyed identity is guaranteed*, and that ordering is asserted by
900/// `tests/anonymous_ceiling.rs` rather than left to two constants drifting
901/// apart. It is well above 1, because a public repository that can only ever be
902/// cloned by one person at a time is not public in any useful sense — and
903/// because a ceiling of 1 turns every ordinary CI runner pair into a refusal.
904///
905/// Raise it on a box whose *purpose* is public serving (a browse replica, see
906/// [`crate::control::AnonymousPolicy`]); lower it to 0 — or close the door with
907/// [`SshConfig::without_anonymous_access`](crate::SshConfig::without_anonymous_access)
908/// — on one whose purpose is not.
909///
910/// [`DEFAULT_MAX_TRANSFERS_PER_PRINCIPAL`]: crate::config::DEFAULT_MAX_TRANSFERS_PER_PRINCIPAL
911pub const DEFAULT_MAX_ANONYMOUS_TRANSFERS: usize = 5;
912
913/// How long a parked admission sleeps before re-asking, when nothing wakes it.
914///
915/// A waiter is normally woken by a guard being dropped
916/// ([`Admission::slot_freed`]). This bound exists for the other way the answer
917/// changes: the CoDel limit grows by one every window
918/// ([`crate::governor::INTERVAL`], 100 ms) whose delay was under target, so a
919/// server that is not finishing anything can still have room a moment later.
920/// One window is therefore exactly the right period — sleeping longer would
921/// hold clients past the moment their slot existed, shorter would poll for no
922/// new information.
923const RETRY_INTERVAL: Duration = crate::governor::INTERVAL;
924
925/// How often the `git.transfer.waited` row may be written, in milliseconds.
926///
927/// One second. See [`Admission::wait_row_due`]: the fact that clients are
928/// waiting is worth a row, and one row per waiting client is worth none because
929/// it is the only thing left on the screen.
930const WAIT_ROW_INTERVAL_MS: u64 = 1000;
931
932/// Whether an admission attempt that fails should move a refusal counter.
933///
934/// A wait loop asks the same door many times for one operation; only the last
935/// answer is a refusal. See [`Admission::attempt`].
936#[derive(Debug, Clone, Copy, PartialEq, Eq)]
937enum Count {
938    /// This ask is final: a refusal is a refusal.
939    Yes,
940    /// This ask is a retry inside a wait: a refusal here is a poll.
941    No,
942}
943
944impl Count {
945    fn bump(self, counter: &AtomicU64) {
946        if self == Count::Yes {
947            counter.fetch_add(1, Ordering::Relaxed);
948        }
949    }
950}
951
952/// The gate itself.
953///
954/// Cheap to clone by `Arc`; it holds one semaphore and one small map.
955#[derive(Debug)]
956pub struct Admission {
957    /// Both gates, or `None` when the operator turned the server-wide limit
958    /// off. That is a legitimate configuration (an appliance behind its own
959    /// front door), and it is spelled as an `Option` rather than as `usize::MAX`
960    /// so a log can say "unlimited" and mean it. Unlimited means unlimited: no
961    /// concurrency band and no memory budget, because a budget that binds on a
962    /// gate the operator switched off is a limit they did not ask for.
963    governor: Option<Governor>,
964    max_concurrent: Option<usize>,
965    max_per_principal: Option<usize>,
966    /// When `true`, `max_per_principal` binds unconditionally — the behaviour
967    /// before max-min fairness. `false` (the default) makes it the **floor** of
968    /// a share that only applies under contention.
969    hard_per_principal: bool,
970    /// The **anonymous** ceiling: the most reads
971    /// [`gunnar_core::ANONYMOUS_PRINCIPAL_ID`] may hold at once, or `None` to
972    /// let anonymous take the same max-min share as any keyed identity.
973    ///
974    /// Kept as its own field rather than as an entry in some per-principal
975    /// override map, because a map would invite a second override and the
976    /// second override is a policy engine. There is exactly one caller that has
977    /// no identity, so there is exactly one number here.
978    anonymous_ceiling: Option<usize>,
979    /// Only ever holds principals with a transfer in flight; an entry is removed
980    /// when its count reaches zero, so an idle server carries an empty map
981    /// however many identities have ever connected.
982    ///
983    /// Its **length** is therefore the live count of active identities, which is
984    /// what the max-min share divides by. Keeping the map free of zero entries
985    /// is no longer only hygiene: a leaked entry would shrink everybody's share.
986    per_principal: Mutex<HashMap<String, usize>>,
987    counters: Arc<Counters>,
988    /// Woken every time a guard is dropped, so a waiting admission learns of a
989    /// free slot the instant there is one rather than by polling.
990    ///
991    /// `tokio::sync::Notify` and not a semaphore: the thing being waited for is
992    /// not one permit but *the three gates agreeing*, and only
993    /// [`Admission::try_admit`] knows whether they do. A waiter is woken, it
994    /// re-asks, and if the answer is still no it waits again — which is also
995    /// why a lost wakeup is impossible here: the `Notified` future is created
996    /// **before** the try, so a release that lands in between is still seen.
997    slot_freed: tokio::sync::Notify,
998    /// How long an operation may wait for a slot before it is refused. Zero
999    /// restores the instant refusal exactly.
1000    wait: Duration,
1001    /// How long a **keyed** (authenticated) caller may wait, when longer than
1002    /// [`Admission::wait`]. `None`: the same wait as everybody. See
1003    /// [`crate::governor::DEFAULT_KEYED_ADMISSION_WAIT`].
1004    keyed_wait: Option<Duration>,
1005    /// When this gate was built, so [`Admission::wait_row_due`] can hold a
1006    /// deadline in one atomic instead of a lock.
1007    epoch: std::time::Instant,
1008    /// Milliseconds since [`Admission::epoch`] at which the next `waited` row
1009    /// may be written.
1010    next_wait_row_ms: AtomicU64,
1011    /// **The pause** ([`Pause`]), `0` for none, else `1 + Pause as u8`. One
1012    /// atomic read on every admission and nothing else on the hot path; set
1013    /// by [`Self::pause`] from the runtime growth's drain, cleared by
1014    /// [`Self::unpause`] when `/data` is back — or when the growth aborted.
1015    pause: std::sync::atomic::AtomicU8,
1016    /// **The per-namespace flow ledger, when the server wired one.**
1017    ///
1018    /// `None` in every test in this crate and in any build with no control
1019    /// store behind it, which is why it is an `Option` rather than something
1020    /// this crate constructs: the gate must work on a bare `Admission::new`,
1021    /// and a ledger it invented would count into a store nobody reads.
1022    ///
1023    /// It is here rather than at the three doors that call [`Self::try_admit`]
1024    /// — the SSH exec line, the smart-HTTP handler and the gRPC transfer gate —
1025    /// because those three converge on [`Self::count_refusal`] and a refusal
1026    /// counted at each of them would be the same event counted three ways, one
1027    /// of which would go stale the first time a fourth door was added.
1028    flow: Option<Arc<gunnar_core::FlowLedger>>,
1029    /// **The admission policy** (`admit-share`): asked for a keyed transfer's
1030    /// class. [`edda_admit::PassThrough`] until the server hands it the
1031    /// configured one ([`Admission::with_policy`]), so a gate built by a test
1032    /// that never heard of budgets decides exactly as before.
1033    #[cfg(feature = "admit-share")]
1034    policy: Policy,
1035}
1036
1037/// `Arc<dyn Admit>` with a `Debug` that names the policy, so [`Admission`]
1038/// keeps its derive.
1039#[cfg(feature = "admit-share")]
1040#[derive(Clone)]
1041struct Policy(Arc<dyn edda_admit::Admit>);
1042
1043#[cfg(feature = "admit-share")]
1044impl std::fmt::Debug for Policy {
1045    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
1046        f.write_str(self.0.policy())
1047    }
1048}
1049
1050/// **The `edda_admit` kind of work an [`OpClass`] is.** Estimates learn per kind.
1051pub fn work_kind(class: OpClass) -> edda_admit::WorkKind {
1052    match class {
1053        OpClass::Clone => edda_admit::WorkKind::Clone,
1054        OpClass::Fetch => edda_admit::WorkKind::Fetch,
1055        OpClass::Push => edda_admit::WorkKind::Push,
1056        OpClass::LsRemote | OpClass::Gc => edda_admit::WorkKind::Other,
1057    }
1058}
1059
1060impl Admission {
1061    /// Build a gate from the two configured limits, with the per-principal one
1062    /// as a max-min **share** (see the module docs), the server-wide one as the
1063    /// ceiling of the governor's band, and the default memory budget.
1064    pub fn new(max_concurrent: Option<usize>, max_per_principal: Option<usize>) -> Admission {
1065        Admission::with_memory_budget(
1066            max_concurrent,
1067            max_per_principal,
1068            DEFAULT_MEMORY_BUDGET_BYTES,
1069        )
1070    }
1071
1072    /// **Give the gate a flow ledger**, so a refusal is charged to the
1073    /// namespace that met it and not only to the box's own counters.
1074    ///
1075    /// Builder-style and optional for [`Admission::flow`]'s reason. A gate
1076    /// without one refuses exactly as it always did and simply reports the
1077    /// refusal nowhere but its own `AdmissionStats`, which is what every build
1078    /// before the flow ledger existed did.
1079    pub fn with_flow(mut self, flow: Arc<gunnar_core::FlowLedger>) -> Admission {
1080        self.flow = Some(flow);
1081        self
1082    }
1083
1084    /// **The flow ledger this gate was given**, or `None`.
1085    ///
1086    /// Handed out because this crate's other tenant-facing door — the exec line
1087    /// that denies a principal who holds no grant on the repository it named —
1088    /// needs to charge that tenant and has nowhere else in this crate to reach
1089    /// a ledger. One handle, wired once by the server, rather than a second
1090    /// field on `Shared` that a future constructor could forget to fill.
1091    pub fn flow(&self) -> Option<&Arc<gunnar_core::FlowLedger>> {
1092        self.flow.as_ref()
1093    }
1094
1095    /// The same gate with the memory budget (gate 2) chosen explicitly.
1096    ///
1097    /// The budget has to be an operator's number: by the OS-less law this code
1098    /// cannot read the machine's RAM, so
1099    /// [`DEFAULT_MEMORY_BUDGET_BYTES`](crate::governor::DEFAULT_MEMORY_BUDGET_BYTES)
1100    /// is a constant with a measurement behind it rather than a fit to the box.
1101    pub fn with_memory_budget(
1102        max_concurrent: Option<usize>,
1103        max_per_principal: Option<usize>,
1104        memory_budget: u64,
1105    ) -> Admission {
1106        Admission {
1107            governor: max_concurrent
1108                .map(|budget| Governor::new(LimitBand::for_budget(budget), memory_budget)),
1109            max_concurrent,
1110            max_per_principal,
1111            hard_per_principal: false,
1112            // On by default, because the anonymous door is open by default: a limit
1113            // that has to be switched on is not protecting the configuration
1114            // anybody actually runs.
1115            anonymous_ceiling: Some(DEFAULT_MAX_ANONYMOUS_TRANSFERS),
1116            per_principal: Mutex::new(HashMap::new()),
1117            counters: Arc::new(Counters::default()),
1118            slot_freed: tokio::sync::Notify::new(),
1119            wait: DEFAULT_ADMISSION_WAIT,
1120            keyed_wait: None,
1121            epoch: std::time::Instant::now(),
1122            next_wait_row_ms: AtomicU64::new(0),
1123            pause: std::sync::atomic::AtomicU8::new(0),
1124            flow: None,
1125            #[cfg(feature = "admit-share")]
1126            policy: Policy(Arc::new(edda_admit::PassThrough::new())),
1127        }
1128    }
1129
1130    /// **Give the gate its admission policy** (`admit-share`) — the one built
1131    /// from `--admit-limit` ([`edda_admit::Settings::build`]; `none` is
1132    /// PassThrough).
1133    #[cfg(feature = "admit-share")]
1134    pub fn with_policy(mut self, policy: Arc<dyn edda_admit::Admit>) -> Admission {
1135        self.policy = Policy(policy);
1136        self
1137    }
1138
1139    /// **The admission policy in force**, for `Usage.AdmitPlan`.
1140    ///
1141    /// Without `admit-share` the gate has none, and this is a fresh
1142    /// [`edda_admit::PassThrough`] — every kind `Full`, no budget, nothing
1143    /// learnt — which is the truth about that build.
1144    pub fn admit_policy(&self) -> Arc<dyn edda_admit::Admit> {
1145        #[cfg(feature = "admit-share")]
1146        {
1147            Arc::clone(&self.policy.0)
1148        }
1149        #[cfg(not(feature = "admit-share"))]
1150        {
1151            Arc::new(edda_admit::PassThrough::new())
1152        }
1153    }
1154
1155    /// **The usage `Usage.AdmitPlan` reports for `namespace`** — a READ for
1156    /// the admin view, never on the admission path.
1157    ///
1158    /// A policy with a budget answers from its own table (the same figure the
1159    /// gate plans with). Without one — PassThrough, or a build without
1160    /// `admit-share` — the flow ledger's window is summed directly, so the
1161    /// view still shows what moved; a gate with no ledger counted nothing and
1162    /// says zero.
1163    pub fn admit_usage(
1164        &self,
1165        policy: &dyn edda_admit::Admit,
1166        namespace: &str,
1167    ) -> edda_admit::Usage {
1168        if policy.limit().is_some() {
1169            return policy.usage(&edda_admit::Consumer::Namespace(namespace.to_owned()));
1170        }
1171        let window = edda_admit::DEFAULT_WINDOW_DAYS;
1172        let bytes = match &self.flow {
1173            Some(flow) if !namespace.is_empty() => {
1174                flow.window_bytes(namespace, gunnar_core::now_ms(), window)
1175            }
1176            _ => 0,
1177        };
1178        edda_admit::Usage::new(bytes, window)
1179    }
1180
1181    /// **The flow ledger as a seed** for a policy's table: a namespace's
1182    /// bytes per UTC day over the window, read ONCE per namespace on first
1183    /// sight. See [`gunnar_core::FlowLedger::moved`].
1184    pub fn flow_seed(flow: Arc<gunnar_core::FlowLedger>) -> Arc<dyn edda_admit::Seed> {
1185        Arc::new(
1186            move |who: &edda_admit::Consumer, first_day: u64, today: u64| {
1187                let edda_admit::Consumer::Namespace(namespace) = who else {
1188                    return Vec::new();
1189                };
1190                let days = u32::try_from(today.saturating_sub(first_day) + 1).unwrap_or(u32::MAX);
1191                let now = gunnar_core::now_ms();
1192                flow.moved(namespace, now, days)
1193                    .into_iter()
1194                    .filter_map(|(day, bytes)| u64::try_from(day).ok().map(|d| (d, bytes)))
1195                    .collect()
1196            },
1197        )
1198    }
1199
1200    /// **The plan for a keyed transfer** (`admit-share`); `None` for the
1201    /// anonymous principal, which has its own ceiling and no namespace budget.
1202    #[cfg(feature = "admit-share")]
1203    fn plan(&self, principal: &str, namespace: &str, class: OpClass) -> Option<edda_admit::Plan> {
1204        if principal == gunnar_core::ANONYMOUS_PRINCIPAL_ID {
1205            return None;
1206        }
1207        let policy = &*self.policy.0;
1208        let who = edda_admit::Consumer::Namespace(namespace.to_owned());
1209        // The policy's own table: lock-free loads for a namespace already
1210        // seen, seeded ONCE from the flow ledger on first sight
1211        // (`GunnarSsh::with_flow`). PassThrough answers without a table.
1212        let usage = policy.usage(&who);
1213        Some(policy.plan(&edda_admit::Work::new(work_kind(class), None), &who, &usage))
1214    }
1215
1216    /// The same gate with `max_per_principal` as an **unconditional** ceiling.
1217    ///
1218    /// The escape hatch for a host where an identity is a tenant and the cap is
1219    /// what is being sold, rather than an appliance where the identities are all
1220    /// the same team. Selected by
1221    /// [`SshConfig::per_principal_limit_is_hard`](crate::SshConfig::per_principal_limit_is_hard).
1222    pub fn hard_capped(
1223        max_concurrent: Option<usize>,
1224        max_per_principal: Option<usize>,
1225    ) -> Admission {
1226        Admission {
1227            hard_per_principal: true,
1228            ..Admission::new(max_concurrent, max_per_principal)
1229        }
1230    }
1231
1232    /// The same gate with a different memory budget (gate 2).
1233    ///
1234    /// A builder rather than a fourth constructor argument, so the two
1235    /// concurrency numbers keep the shape every existing caller passes them in.
1236    /// Only meaningful before the gate has admitted anything: it replaces the
1237    /// governor, and a governor holds live occupancy.
1238    pub fn with_budget(self, memory_budget: u64) -> Admission {
1239        Admission {
1240            governor: self
1241                .max_concurrent
1242                .map(|budget| Governor::new(LimitBand::for_budget(budget), memory_budget)),
1243            ..self
1244        }
1245    }
1246
1247    /// A gate with no limits at all; every request is admitted.
1248    ///
1249    /// **Including the anonymous ceiling.** "Unlimited" has to mean unlimited or
1250    /// it is a trap: every test that reaches for this gate does so to take the
1251    /// admission gate *out* of the picture, and one surviving ceiling would make
1252    /// those tests fail for a reason that has nothing to do with what they are
1253    /// about. An operator who wants only the anonymous ceiling builds a real
1254    /// gate and passes `None` for the other two.
1255    pub fn unlimited() -> Admission {
1256        Admission {
1257            anonymous_ceiling: None,
1258            ..Admission::new(None, None)
1259        }
1260    }
1261
1262    /// The same gate with the **anonymous ceiling** set explicitly;
1263    /// `None` switches it off and lets anonymous take a keyed identity's share.
1264    ///
1265    /// A builder and not a fourth constructor argument for the same reason
1266    /// [`Admission::with_budget`] is one: the two concurrency numbers keep the
1267    /// shape every existing caller already passes them in, and a gate built by
1268    /// a test that has never heard of the anonymous door still gets the default ceiling rather
1269    /// than silently getting none.
1270    pub fn with_anonymous_ceiling(self, anonymous_ceiling: Option<usize>) -> Admission {
1271        Admission {
1272            anonymous_ceiling,
1273            ..self
1274        }
1275    }
1276
1277    /// The anonymous ceiling in force, or `None` when anonymous is held
1278    /// only by the ordinary per-principal share.
1279    pub fn max_anonymous(&self) -> Option<usize> {
1280        self.anonymous_ceiling
1281    }
1282
1283    /// **How many slots at the top of the box are keyed-only** right now
1284    /// (D23b), or `0` on a gate with no server-wide limit to reserve out of.
1285    ///
1286    /// Derived, never configured. It is whatever is left of the **governed**
1287    /// limit above [`CONTENTION_PERCENT`], so it moves with the CoDel
1288    /// controller — which is the entire point: the anonymous ceiling is an
1289    /// absolute number and the box it is a share of is not, and a stranger
1290    /// holding five of a shed nineteen is holding a quarter of the machine at
1291    /// the moment the machine has least to give.
1292    ///
1293    /// Reusing `CONTENTION_PERCENT` rather than inventing a fourth number is
1294    /// the decision, not a shortcut. That constant is *defined* as the
1295    /// occupancy at which gunnar starts asking keyed identities to share with
1296    /// each other; the moment it does that is exactly the moment it should stop
1297    /// asking them to share with strangers. Two thresholds would be two
1298    /// opinions about when the box is busy, and the order between them would be
1299    /// an accident of arithmetic rather than a decision.
1300    ///
1301    /// At least `1` whenever there is a limit at all: a reserve that rounded to
1302    /// zero on a small box would be an empty guarantee at exactly the size where
1303    /// a single stranger is the whole problem.
1304    pub fn keyed_reserve(&self) -> usize {
1305        let Some(governor) = &self.governor else {
1306            return 0;
1307        };
1308        let limit = governor.limit();
1309        // The largest occupancy still BELOW contention is
1310        // `ceil(limit * P / 100) - 1`; everything from there up is reserved.
1311        // Integer throughout — a float threshold would put the boundary
1312        // somewhere an operator cannot reproduce with a calculator.
1313        limit - limit.saturating_mul(CONTENTION_PERCENT).div_ceil(100) + 1
1314    }
1315
1316    /// Whether the per-principal limit binds unconditionally.
1317    pub fn per_principal_is_hard(&self) -> bool {
1318        self.hard_per_principal
1319    }
1320
1321    /// The server-wide limit, if there is one.
1322    pub fn max_concurrent(&self) -> Option<usize> {
1323        self.max_concurrent
1324    }
1325
1326    /// The per-principal limit, if there is one.
1327    pub fn max_per_principal(&self) -> Option<usize> {
1328        self.max_per_principal
1329    }
1330
1331    /// The **governed** server-wide limit right now. `None` when unlimited.
1332    pub fn governed_limit(&self) -> Option<usize> {
1333        self.governor.as_ref().map(Governor::limit)
1334    }
1335
1336    /// The memory budget (gate 2). `None` when unlimited.
1337    pub fn memory_budget(&self) -> Option<u64> {
1338        self.governor.as_ref().map(|g| g.memory().budget())
1339    }
1340
1341    /// The governor, for a caller that wants the controller's own counters.
1342    pub fn governor(&self) -> Option<&Governor> {
1343        self.governor.as_ref()
1344    }
1345
1346    /// Take a slot for `principal` running an operation of `class`, or say why
1347    /// not. **Never blocks.**
1348    ///
1349    /// This is the ONE door. Every user-level git operation — clone, fetch,
1350    /// push, ls-remote, gc — passes through here and through nothing else, so
1351    /// there is a single place where "the server cannot keep up" is decided and
1352    /// a single place a counter has to be trusted.
1353    ///
1354    /// Three gates, in this order, all of them **before any of the operation's
1355    /// work is allocated**:
1356    ///
1357    /// 1. **time** — the governed concurrency limit, moved by the per-class
1358    ///    CoDel controller;
1359    /// 2. **space** — the memory reservation for `class`;
1360    /// 3. **fairness** — the per-principal max-min share.
1361    ///
1362    /// Time and space come first because they are facts about the *server*, and
1363    /// a server that cannot serve anybody should not spend a lock working out
1364    /// whose turn it was. The returned guard releases everything when it is
1365    /// dropped — slot, reservation and principal count — so no error path, and
1366    /// no panic, can skip it. That is how a leaked permit turns a working limit
1367    /// into a server that refuses everything an hour after start-up.
1368    ///
1369    /// The caller **must** call [`AdmissionGuard::entered_service`] when the
1370    /// operation's work actually begins, or the controller gets no sojourn
1371    /// sample from it and [`AdmissionStats::unmeasured_sojourn`] counts it.
1372    /// `namespace` is the repository's namespace, for the flow ledger; `""`
1373    /// where the door does not know one, which charges the refusal to the box
1374    /// alone rather than inventing a tenant for it.
1375    pub fn try_admit(
1376        self: &Arc<Self>,
1377        principal: &str,
1378        namespace: &str,
1379        class: OpClass,
1380    ) -> Result<AdmissionGuard, Refusal> {
1381        self.attempt(principal, namespace, class, Count::Yes)
1382    }
1383
1384    /// [`Admission::try_admit`], told whether this ask is the **final** one.
1385    ///
1386    /// [`Admission::admit_within`] asks many times for one operation while it
1387    /// waits, and a refusal counter that moved on every ask would say 21 where
1388    /// one client was refused once. So the wait loop asks with [`Count::No`]
1389    /// and counts once, at the end, when the refusal is final.
1390    fn attempt(
1391        self: &Arc<Self>,
1392        principal: &str,
1393        namespace: &str,
1394        class: OpClass,
1395        counting: Count,
1396    ) -> Result<AdmissionGuard, Refusal> {
1397        let outcome = self.attempt_inner(principal, namespace, class, counting);
1398        // **The per-namespace half of a FINAL refusal**, and the second of the
1399        // two places it is charged from — see [`Self::charge_flow`] for why
1400        // there are two and not one.
1401        //
1402        // RED when this was missing: `attempt` counts the box's own refusal
1403        // counters INLINE (`counting.bump`, six sites) and never calls
1404        // [`Self::count_refusal`], which only [`Self::admit_within`] reaches.
1405        // So the SSH exec line charged the tenant and the two doors that call
1406        // [`Self::try_admit`] — smart HTTP and the gRPC transfer gate — charged
1407        // nobody. The compiler said so out loud (`unused variable: namespace`)
1408        // and the claim that all three doors converged on one counting point
1409        // was simply false.
1410        if counting == Count::Yes && outcome.is_err() {
1411            self.charge_flow(namespace);
1412        }
1413        outcome
1414    }
1415
1416    /// **Charge one admission refusal to the namespace that met it.**
1417    ///
1418    /// The one writer, called from the two places a refusal becomes FINAL:
1419    /// [`Self::attempt`] for the doors that do not wait
1420    /// ([`Self::try_admit`] — smart HTTP and the gRPC transfer gate), and
1421    /// [`Self::count_refusal`] for the door that does
1422    /// ([`Self::admit_within`] — the SSH exec line, which asks many times for
1423    /// one operation and must charge exactly once, at the end).
1424    ///
1425    /// Two call sites and not one because "final" means two different moments
1426    /// on those two paths, and that is the same distinction [`Count`] already
1427    /// exists to draw. What must not be two is the ARITHMETIC, and it is not.
1428    fn charge_flow(&self, namespace: &str) {
1429        if let Some(flow) = &self.flow {
1430            flow.refuse(
1431                namespace,
1432                gunnar_core::now_ms(),
1433                gunnar_core::Refusal::Admission,
1434            );
1435        }
1436    }
1437
1438    fn attempt_inner(
1439        self: &Arc<Self>,
1440        principal: &str,
1441        namespace: &str,
1442        class: OpClass,
1443        counting: Count,
1444    ) -> Result<AdmissionGuard, Refusal> {
1445        // ── D23b: is this caller allowed the last slots? ───────────────────
1446        //
1447        // `true` only for the one principal that has no key behind it, so on the
1448        // path that carries essentially all traffic this is a string compare
1449        // against a `&'static str` and then nothing at all. Computed here rather
1450        // than after the claim because it has to ride INSIDE the claim's
1451        // compare-and-swap; see `Governor::claim_reserving` for the race that
1452        // taught us so, and `Admission::keyed_reserve` for where the number
1453        // comes from.
1454        //
1455        // Skipped when the ceiling is off (the operator opted out of anonymous read
1456        // entirely, and `Admission::unlimited` must mean unlimited) and when it
1457        // is 0 (the door is shut at every occupancy, and `AnonymousAtCapacity`
1458        // is the truer sentence for that).
1459        //
1460        // ── the PAUSE, before any slot is looked at ─────────────────────
1461        // A paused gate refuses everybody with the pause's words; nothing is
1462        // claimed, so nothing has to be released, and the in-flight count
1463        // keeps meaning "transfers that were admitted before the pause" —
1464        // which is exactly what the drain waits on.
1465        if let Some(pause) = self.paused() {
1466            counting.bump(&self.counters.refused_paused);
1467            return Err(Refusal::Paused { pause });
1468        }
1469        // ── admit-share: a keyed namespace over its budget takes the anonymous
1470        // share (module docs). Without the feature this block does not exist
1471        // and `reduced` is the constant `false`.
1472        #[cfg(feature = "admit-share")]
1473        let plan = self.plan(principal, namespace, class);
1474        #[cfg(feature = "admit-share")]
1475        let reduced = plan.as_ref().is_some_and(|p| {
1476            matches!(
1477                p.class,
1478                edda_admit::Class::Reduced | edda_admit::Class::Floor
1479            )
1480        });
1481        #[cfg(not(feature = "admit-share"))]
1482        let (reduced, _) = (false, namespace);
1483        let anonymous_ceiling = self
1484            .anonymous_ceiling
1485            .filter(|_| principal == gunnar_core::ANONYMOUS_PRINCIPAL_ID || reduced);
1486        let reserve = match anonymous_ceiling {
1487            Some(cap) if cap > 0 => self.keyed_reserve(),
1488            _ => 0,
1489        };
1490
1491        // The server-wide gates first: they are the resources that actually run
1492        // out, and taking them first means the per-principal map is only ever
1493        // touched by a request that could have run.
1494        let counts = counting == Count::Yes;
1495        let claim = match &self.governor {
1496            Some(governor) => match governor.claim_reserving(class, reserve, counts) {
1497                Ok(claim) => Some(claim),
1498                Err(Shed::Reserved {
1499                    occupied,
1500                    limit,
1501                    reserve,
1502                }) => {
1503                    #[cfg(feature = "admit-share")]
1504                    if reduced {
1505                        counting.bump(&self.counters.refused_reduced);
1506                        self.warn_reduced(plan.as_ref(), namespace);
1507                        return Err(Refusal::ReducedShare {
1508                            ceiling: anonymous_ceiling.unwrap_or(0),
1509                            busy: true,
1510                        });
1511                    }
1512                    counting.bump(&self.counters.refused_keyed_reserve);
1513                    return Err(Refusal::KeyedReserve {
1514                        occupied,
1515                        limit,
1516                        reserve,
1517                    });
1518                }
1519                Err(Shed::Concurrency { limit, .. }) => {
1520                    counting.bump(&self.counters.refused_at_capacity);
1521                    return Err(Refusal::AtCapacity { limit });
1522                }
1523                Err(Shed::LaneFull {
1524                    lane,
1525                    ceiling,
1526                    limit,
1527                }) => {
1528                    counting.bump(&self.counters.refused_lane);
1529                    return Err(Refusal::LaneAtCapacity {
1530                        lane,
1531                        ceiling,
1532                        limit,
1533                    });
1534                }
1535                Err(Shed::Memory {
1536                    class,
1537                    wanted,
1538                    reserved,
1539                    budget,
1540                }) => {
1541                    counting.bump(&self.counters.refused_memory);
1542                    return Err(Refusal::MemoryBudget {
1543                        class,
1544                        wanted,
1545                        reserved,
1546                        budget,
1547                    });
1548                }
1549            },
1550            None => None,
1551        };
1552
1553        // The absolute anonymous ceiling, which is a different question
1554        // from the reserve above — that one asked *may this caller take the
1555        // slots that are left*, and this one asks *how many may this caller ever
1556        // hold at once*. Consulted here rather than before the claim because it
1557        // needs the per-principal map, and the map is only ever touched by a
1558        // request that could have run.
1559        if self.max_per_principal.is_some() || anonymous_ceiling.is_some() {
1560            let mut held = self.per_principal.lock().unwrap_or_else(|e| e.into_inner());
1561
1562            // Everything the decision needs, read once under the one lock the
1563            // counts live under. `active` includes this candidate whether or
1564            // not it already has an entry, because a share divided by the
1565            // claimants that are already in is not this claimant's share.
1566            let already_active = held.contains_key(principal);
1567            let active = held.len() + usize::from(!already_active);
1568            let count = held.entry(principal.to_string()).or_insert(0);
1569
1570            // The anonymous ceiling REPLACES the share for the anonymous
1571            // principal; it does not combine with it. Not `min` of the two,
1572            // because the two numbers answer different questions and a `min`
1573            // would let the max-min rule's contention threshold leak in: on an
1574            // idle server the share evaluates to the whole box, so `min` would
1575            // read 5 — correct by accident — while under contention it could
1576            // evaluate lower and the refusal would then be reported as a
1577            // *share*, sending the operator to the wrong knob. One caller, one
1578            // rule, one number in the log.
1579            let (ceiling, anonymous) = match anonymous_ceiling {
1580                Some(cap) => (cap, true),
1581                // Unreachable with `None`: the `if` above is exactly the
1582                // disjunction "there is a per-principal limit OR an anonymous
1583                // ceiling", and this arm is the half where the second is off.
1584                None => (
1585                    self.ceiling_for(self.max_per_principal.unwrap_or(usize::MAX), active),
1586                    false,
1587                ),
1588            };
1589            if *count >= ceiling {
1590                // Do not leave a zero entry behind for a principal that was
1591                // refused: `or_insert` may have just created one. This matters
1592                // more than it used to — a stale entry inflates `active` and
1593                // shrinks everyone's share.
1594                if *count == 0 {
1595                    held.remove(principal);
1596                }
1597                drop(held);
1598                // Hand the slot AND the memory reservation straight back. The
1599                // guard is what normally does this, and there is no guard yet.
1600                if let (Some(governor), Some(claim)) = (&self.governor, &claim) {
1601                    governor.release(claim, 0);
1602                }
1603                #[cfg(feature = "admit-share")]
1604                if anonymous && reduced {
1605                    counting.bump(&self.counters.refused_reduced);
1606                    self.warn_reduced(plan.as_ref(), namespace);
1607                    return Err(Refusal::ReducedShare {
1608                        ceiling,
1609                        busy: false,
1610                    });
1611                }
1612                if anonymous {
1613                    counting.bump(&self.counters.refused_anonymous);
1614                    return Err(Refusal::AnonymousAtCapacity { limit: ceiling });
1615                }
1616                counting.bump(&self.counters.refused_per_principal);
1617                return Err(Refusal::PrincipalAtCapacity {
1618                    limit: ceiling,
1619                    active_principals: active,
1620                });
1621            }
1622            *count += 1;
1623        }
1624
1625        self.counters.admitted.fetch_add(1, Ordering::Relaxed);
1626        let in_flight = self.counters.in_flight.fetch_add(1, Ordering::Relaxed) + 1;
1627        self.counters
1628            .peak_in_flight
1629            .fetch_max(in_flight, Ordering::Relaxed);
1630
1631        // admit-share: the policy's own slot shares are not configured here
1632        // (this gate IS the share), so its `admit` cannot refuse a namespace;
1633        // should a policy ever do so, the gate's decision above stands and the
1634        // transfer simply teaches the estimate nothing.
1635        #[cfg(feature = "admit-share")]
1636        let admitted = plan.and_then(|plan| self.policy.0.admit(&plan).ok());
1637        Ok(AdmissionGuard {
1638            #[cfg(feature = "admit-share")]
1639            admitted,
1640            gate: Arc::clone(self),
1641            // Some **exactly when** the block above incremented a count, so the
1642            // guard's `Drop` decrements precisely what was incremented. The two
1643            // conditions have to be the same expression: an anonymous
1644            // admission counted under a gate with no per-principal limit and
1645            // released by `self.max_per_principal.map(..)` would never be
1646            // released, and the ceiling would fall by one per anonymous clone
1647            // until the door was shut for good.
1648            principal: (self.max_per_principal.is_some() || anonymous_ceiling.is_some())
1649                .then(|| principal.to_string()),
1650            claim,
1651            marked: ServiceMark::default(),
1652            wire: AtomicU64::new(0),
1653            grown: AtomicU64::new(0),
1654        })
1655    }
1656
1657    /// The `warn` row for a reduced-share refusal, carrying the plan's sentence
1658    /// — the one thing a `Copy` [`Refusal`] cannot.
1659    #[cfg(feature = "admit-share")]
1660    fn warn_reduced(&self, plan: Option<&edda_admit::Plan>, namespace: &str) {
1661        tracing::warn!(
1662            target: "gunnar::git",
1663            event = "git.admission.reduced",
1664            namespace,
1665            why = plan.map_or("", |p| p.why.as_str()),
1666            "a namespace over its transfer budget met a full reduced share"
1667        );
1668    }
1669
1670    /// The same gate with a different **admission wait**.
1671    ///
1672    /// `Duration::ZERO` is the pre-2026-08-30 gate exactly: refuse on the spot.
1673    /// See [`DEFAULT_ADMISSION_WAIT`] for why waiting is cheaper than refusing
1674    /// once the client's reconnect is counted.
1675    pub fn with_wait(self, wait: Duration) -> Admission {
1676        Admission { wait, ..self }
1677    }
1678
1679    /// How long this gate lets an operation wait for a slot.
1680    pub fn wait(&self) -> Duration {
1681        self.wait
1682    }
1683
1684    /// The same gate with a **keyed wait**: how long an authenticated caller
1685    /// may be queued for a slot ([`crate::governor::DEFAULT_KEYED_ADMISSION_WAIT`]).
1686    /// `None` gives keyed callers the ordinary [`Admission::wait`].
1687    pub fn with_keyed_wait(self, keyed_wait: Option<Duration>) -> Admission {
1688        Admission { keyed_wait, ..self }
1689    }
1690
1691    /// **How long `principal` may wait for a slot.** Anonymous: the ordinary
1692    /// wait. A keyed identity: the keyed wait when it is longer — a refused
1693    /// key reconnects (git-lfs's pure-SSH transfer cannot even do that: it
1694    /// falls back to `git-lfs-authenticate` and fails the push), while the
1695    /// parked future costs one session the key's own share already bounds. A
1696    /// zero wait (`--admission-wait-ms 0`) is "refuse on the spot" for all.
1697    pub fn wait_for(&self, principal: &str) -> Duration {
1698        if self.wait.is_zero() || principal == gunnar_core::ANONYMOUS_PRINCIPAL_ID {
1699            return self.wait;
1700        }
1701        self.keyed_wait.map_or(self.wait, |k| k.max(self.wait))
1702    }
1703
1704    /// **Take a slot, waiting up to [`Admission::wait`] for one.**
1705    ///
1706    /// The door every transport goes through. It is
1707    /// [`Admission::try_admit`] in a bounded loop, and everything that decides
1708    /// admission still lives there — this adds only the bound and the counters
1709    /// that make the bound visible.
1710    ///
1711    /// Returns the guard and **how long it waited**, which is the number the
1712    /// caller logs and the number the CoDel controller is fed
1713    /// ([`Governor::record_sojourn`](crate::governor::Governor::record_sojourn)):
1714    /// the wait in *this* queue is the one signal the gate has authority over,
1715    /// because shedding is exactly what shortens it.
1716    ///
1717    /// A refusal after the wait is the same [`Refusal`] the instant path
1718    /// returns, with the same sentence — a client must not be able to tell "the
1719    /// server refused at once" from "the server tried for two seconds first"
1720    /// by the shape of the failure, only by the log.
1721    ///
1722    /// # Cancellation
1723    ///
1724    /// Safe to drop at any await point: nothing is held across one. The slot is
1725    /// taken by the *last* `try_admit`, and either it returns a guard (which
1726    /// owns the release) or it took nothing at all.
1727    pub async fn admit_within(
1728        self: &Arc<Self>,
1729        principal: &str,
1730        namespace: &str,
1731        class: OpClass,
1732    ) -> Result<(AdmissionGuard, Duration), Refusal> {
1733        let started = std::time::Instant::now();
1734        let wait = self.wait_for(principal);
1735        let mut refusal = match self.attempt(principal, namespace, class, Count::No) {
1736            Ok(guard) => {
1737                // A wait of zero is a sample too, and it is the one that grows
1738                // the limit back on a server that is coping.
1739                self.record_wait(class, Duration::ZERO);
1740                return Ok((guard, Duration::ZERO));
1741            }
1742            Err(refusal) => refusal,
1743        };
1744        // No wait configured — or the caller is anonymous and was refused for
1745        // being anonymous, which are the refusals this door never parks.
1746        //
1747        // The wait exists because a refused *keyed* client reconnects, and one
1748        // parked future is cheaper than the reconnect. For anonymous that trade
1749        // inverts: the parked future is holding the TCP socket, the russh
1750        // session and its key material for up to `self.wait`, and those are
1751        // exactly the resources a clone fork-bomb is spending gunnar's box to
1752        // acquire. Parking N anonymous readers to protect them from a reconnect
1753        // hands an attacker N held sessions for free — the queue becomes the
1754        // attack. It is also pointless on its own terms: the ceiling is small
1755        // and hard, so unlike a governed limit it will not have grown by the
1756        // time the waiter re-asks; only a transfer finishing can free it, and
1757        // a stranger has no claim on being first in line for that.
1758        //
1759        // `KeyedReserve` (D23b) is parked no more than `AnonymousAtCapacity` is,
1760        // for the same reason twice over. The socket-and-session argument is
1761        // unchanged. And what a parked stranger would be waiting for here is the
1762        // box falling back below `CONTENTION_PERCENT` — which is a KEYED caller
1763        // finishing. Parking a stranger to wait for that is queueing it for a
1764        // slot the whole rule exists to deny it, and every millisecond it spends
1765        // in that queue it is holding the resources the rule was protecting.
1766        //
1767        // `Paused` is never parked either: the pause lasts minutes, not the
1768        // milliseconds a wait covers, and a client parked into a growth is a
1769        // client holding a session across an unmount for nothing.
1770        if wait.is_zero()
1771            || matches!(
1772                refusal,
1773                Refusal::AnonymousAtCapacity { .. }
1774                    | Refusal::KeyedReserve { .. }
1775                    | Refusal::Paused { .. }
1776            )
1777            || is_reduced(&refusal)
1778        {
1779            self.count_refusal(namespace, &refusal);
1780            return Err(refusal);
1781        }
1782        // Parked, and COUNTED as parked: the count is the demand signal
1783        // `Governor::lane_ceiling` reserves other lanes' floors from, and the
1784        // guard pairs the decrement onto every exit including a cancelled
1785        // future.
1786        let _parked = Parked::new(self.governor.as_ref(), class.lane());
1787        loop {
1788            let remaining = match wait.checked_sub(started.elapsed()) {
1789                Some(remaining) if !remaining.is_zero() => remaining,
1790                _ => {
1791                    self.counters
1792                        .refused_after_wait
1793                        .fetch_add(1, Ordering::Relaxed);
1794                    // The refusal is final now, so it is counted now — once,
1795                    // not once per poll.
1796                    self.count_refusal(namespace, &refusal);
1797                    // A wait that ran out is the strongest evidence of a
1798                    // standing queue there is; file it before refusing.
1799                    self.record_wait(class, wait);
1800                    return Err(refusal);
1801                }
1802            };
1803            // Registered BEFORE the retry, so a slot freed between the retry
1804            // and the wait is not a lost wakeup.
1805            let freed = self.slot_freed.notified();
1806            match self.attempt(principal, namespace, class, Count::No) {
1807                Ok(guard) => {
1808                    let waited = started.elapsed();
1809                    let ms = u64::try_from(waited.as_millis()).unwrap_or(u64::MAX);
1810                    self.counters
1811                        .admitted_after_wait
1812                        .fetch_add(1, Ordering::Relaxed);
1813                    self.counters
1814                        .waited_ms_total
1815                        .fetch_add(ms, Ordering::Relaxed);
1816                    self.counters.peak_wait_ms.fetch_max(ms, Ordering::Relaxed);
1817                    self.record_wait(class, waited);
1818                    return Ok((guard, waited));
1819                }
1820                // The same rule inside the loop, for the caller whose FIRST
1821                // refusal was a full server and whose next one is the anonymous
1822                // ceiling: a parked stranger must stop being parked the moment
1823                // the reason becomes "you are a stranger". Without this arm the
1824                // short-circuit above would be defeated by ordering alone.
1825                Err(
1826                    again @ (Refusal::AnonymousAtCapacity { .. } | Refusal::KeyedReserve { .. }),
1827                ) => {
1828                    self.count_refusal(namespace, &again);
1829                    self.record_wait(class, started.elapsed());
1830                    return Err(again);
1831                }
1832                // The reduced share is never parked either (module docs).
1833                Err(again) if is_reduced(&again) => {
1834                    self.count_refusal(namespace, &again);
1835                    self.record_wait(class, started.elapsed());
1836                    return Err(again);
1837                }
1838                Err(again) => refusal = again,
1839            }
1840            // A timeout as well as a notification, because the thing being
1841            // waited for is not only a released slot: the CoDel limit grows on
1842            // its own, and a server whose limit came back while nothing was
1843            // finishing would otherwise keep everybody parked until it did.
1844            let _ = tokio::time::timeout(remaining.min(RETRY_INTERVAL), freed).await;
1845        }
1846    }
1847
1848    /// Move the counter this refusal belongs to. Called exactly once per
1849    /// refused operation, however many times the wait loop asked.
1850    fn count_refusal(&self, namespace: &str, refusal: &Refusal) {
1851        // The tenant's number and the box's number are taken off the one event,
1852        // so they cannot disagree about how many times the door was shut.
1853        self.charge_flow(namespace);
1854        let counter = match refusal {
1855            Refusal::AtCapacity { .. } => &self.counters.refused_at_capacity,
1856            Refusal::LaneAtCapacity { .. } => &self.counters.refused_lane,
1857            Refusal::PrincipalAtCapacity { .. } => &self.counters.refused_per_principal,
1858            Refusal::AnonymousAtCapacity { .. } => &self.counters.refused_anonymous,
1859            Refusal::KeyedReserve { .. } => &self.counters.refused_keyed_reserve,
1860            Refusal::MemoryBudget { .. } => &self.counters.refused_memory,
1861            Refusal::Paused { .. } => &self.counters.refused_paused,
1862            #[cfg(feature = "admit-share")]
1863            Refusal::ReducedShare { .. } => &self.counters.refused_reduced,
1864        };
1865        counter.fetch_add(1, Ordering::Relaxed);
1866    }
1867
1868    /// **May a `waited` row be written now?** At most one per
1869    /// [`WAIT_ROW_INTERVAL`].
1870    ///
1871    /// A row per waiting transfer is a row per *client* under load: the
1872    /// 256-client shape produced 5411 of them in one run, which on the
1873    /// appliance's twenty-row console is the whole screen and nothing else.
1874    /// The fact still has to be visible, so what is emitted is one row per
1875    /// second carrying the AGGREGATE (`admitted_after_wait`, `peak_wait_ms`),
1876    /// which is the shape an operator can actually read. The per-transfer
1877    /// detail stays at `debug`.
1878    ///
1879    /// One compare-and-swap on a deadline, so a hundred racing waiters produce
1880    /// one row and not a hundred.
1881    pub fn wait_row_due(&self) -> bool {
1882        let now = self.epoch.elapsed().as_millis() as u64;
1883        self.next_wait_row_ms
1884            .fetch_update(Ordering::Relaxed, Ordering::Relaxed, |due| {
1885                (now >= due).then_some(now + WAIT_ROW_INTERVAL_MS)
1886            })
1887            .is_ok()
1888    }
1889
1890    /// File one admission wait against the controller.
1891    fn record_wait(&self, class: OpClass, waited: Duration) {
1892        if let Some(governor) = &self.governor {
1893            governor.record_sojourn(class, waited);
1894        }
1895    }
1896
1897    /// How many transfers this principal may hold **right now**.
1898    ///
1899    /// The whole max-min rule, in one place, so there is exactly one writer of
1900    /// it and a test can call it directly instead of inferring it from a
1901    /// refusal. Called with the per-principal lock held and the candidate's own
1902    /// server-wide permit already taken.
1903    ///
1904    /// `configured` is the operator's number and `active` counts the identities
1905    /// holding a transfer, this candidate included.
1906    fn ceiling_for(&self, configured: usize, active: usize) -> usize {
1907        // A pinned cap, or no server-wide limit to take a share OF. In the
1908        // second case `max_per_principal` is the only limit there is, so it has
1909        // to bind unconditionally or it means nothing.
1910        let (Some(governor), false) = (&self.governor, self.hard_per_principal) else {
1911            return configured;
1912        };
1913
1914        // The share divides the **governed** limit, not the configured ceiling.
1915        // A share of capacity the governor has already decided the box does not
1916        // have is not a share; dividing the configured 100 while the controller
1917        // is holding 37 would hand every identity a ceiling above the whole
1918        // server.
1919        let server_wide = governor.limit();
1920
1921        // Occupancy read from the governor rather than from `in_flight`: our own
1922        // slot is already taken, and so is the slot of every other request
1923        // currently between `try_claim` and its `in_flight` increment. The
1924        // counter would miss those and read the server as emptier than it is,
1925        // which is the direction that lets a burst past the share.
1926        let occupied = governor.occupied();
1927
1928        // Integer form of `occupied >= CONTENTION_PERCENT% of server_wide`, so
1929        // there is no float and no rounding to argue about.
1930        let contended =
1931            occupied.saturating_mul(100) >= server_wide.saturating_mul(CONTENTION_PERCENT);
1932        if !contended {
1933            // Free slots exist; the server-wide gate is the only thing that
1934            // should be able to refuse. This is the case the old code got
1935            // wrong.
1936            return server_wide;
1937        }
1938
1939        // Max-min: an equal share of the server, but never less than the
1940        // configured floor — an appliance that promised 32 keeps promising 32
1941        // however many identities turn up, and the server-wide semaphore is
1942        // what stops the floors from over-committing the box.
1943        configured.max(server_wide / active.max(1))
1944    }
1945
1946    /// **Pause the gate**: from here until [`Self::unpause`] every transfer
1947    /// on every front door is refused with `pause.words()`, and nothing new
1948    /// takes a slot. Idempotent; a second pause replaces the first's reason.
1949    pub fn pause(&self, pause: Pause) {
1950        self.pause.store(1 + pause as u8, Ordering::SeqCst);
1951        tracing::warn!(
1952            target: "gunnar::limits",
1953            event = "admission.paused",
1954            reason = pause.reason(),
1955            in_flight = self.counters.in_flight.load(Ordering::Relaxed),
1956            "the admission gate is PAUSED: every new git transfer is refused with `{}` until the \
1957             gate is unpaused; transfers already admitted run to completion",
1958            pause.words()
1959        );
1960    }
1961
1962    /// **Unpause the gate.** Idempotent; an unpaused gate stays unpaused.
1963    pub fn unpause(&self) {
1964        if self.pause.swap(0, Ordering::SeqCst) != 0 {
1965            tracing::info!(
1966                target: "gunnar::limits",
1967                event = "admission.unpaused",
1968                refused_while_paused = self.counters.refused_paused.load(Ordering::Relaxed),
1969                "the admission gate is open again"
1970            );
1971        }
1972    }
1973
1974    /// Count a refusal the session gave at the paused door BEFORE reaching
1975    /// [`Self::admit_within`] (the door answers before the command is
1976    /// resolved, so the gate never sees the request).
1977    pub fn note_refused_paused(&self) {
1978        self.counters.refused_paused.fetch_add(1, Ordering::Relaxed);
1979    }
1980
1981    /// The pause in force, if any.
1982    pub fn paused(&self) -> Option<Pause> {
1983        match self.pause.load(Ordering::SeqCst) {
1984            0 => None,
1985            n if n == 1 + Pause::Growing as u8 => Some(Pause::Growing),
1986            _ => None,
1987        }
1988    }
1989
1990    /// Transfers in flight right now — the number the drain waits to see
1991    /// reach zero. A snapshot of one atomic.
1992    pub fn in_flight(&self) -> u64 {
1993        self.counters.in_flight.load(Ordering::Relaxed)
1994    }
1995
1996    /// Everything the gate knows, as of now.
1997    pub fn stats(&self) -> AdmissionStats {
1998        AdmissionStats {
1999            admitted: self.counters.admitted.load(Ordering::Relaxed),
2000            refused_at_capacity: self.counters.refused_at_capacity.load(Ordering::Relaxed),
2001            refused_per_principal: self.counters.refused_per_principal.load(Ordering::Relaxed),
2002            refused_memory: self.counters.refused_memory.load(Ordering::Relaxed),
2003            refused_anonymous: self.counters.refused_anonymous.load(Ordering::Relaxed),
2004            refused_keyed_reserve: self.counters.refused_keyed_reserve.load(Ordering::Relaxed),
2005            refused_paused: self.counters.refused_paused.load(Ordering::Relaxed),
2006            #[cfg(feature = "admit-share")]
2007            refused_reduced: self.counters.refused_reduced.load(Ordering::Relaxed),
2008            in_flight: self.counters.in_flight.load(Ordering::Relaxed),
2009            peak_in_flight: self.counters.peak_in_flight.load(Ordering::Relaxed),
2010            limit: self.governor.as_ref().map_or(0, Governor::limit),
2011            reserved_bytes: self.governor.as_ref().map_or(0, |g| g.memory().reserved()),
2012            unmeasured_sojourn: self.counters.unmeasured_sojourn.load(Ordering::Relaxed),
2013            admitted_after_wait: self.counters.admitted_after_wait.load(Ordering::Relaxed),
2014            waited_ms_total: self.counters.waited_ms_total.load(Ordering::Relaxed),
2015            peak_wait_ms: self.counters.peak_wait_ms.load(Ordering::Relaxed),
2016            refused_after_wait: self.counters.refused_after_wait.load(Ordering::Relaxed),
2017            refused_lane: self.counters.refused_lane.load(Ordering::Relaxed),
2018            overdrawn: self.counters.overdrawn.load(Ordering::Relaxed),
2019            grown_bytes: self.counters.grown_bytes.load(Ordering::Relaxed),
2020            handoff_ms_total: self.counters.handoff_ms_total.load(Ordering::Relaxed),
2021            peak_handoff_ms: self.counters.peak_handoff_ms.load(Ordering::Relaxed),
2022        }
2023    }
2024
2025    fn release(&self, principal: Option<&str>) {
2026        if let Some(principal) = principal {
2027            let mut held = self.per_principal.lock().unwrap_or_else(|e| e.into_inner());
2028            if let Some(count) = held.get_mut(principal) {
2029                *count = count.saturating_sub(1);
2030                if *count == 0 {
2031                    held.remove(principal);
2032                }
2033            }
2034        }
2035        self.counters.in_flight.fetch_sub(1, Ordering::Relaxed);
2036    }
2037}
2038
2039/// Whether `refusal` is the reduced share — the one `admit_within` must not
2040/// park. The constant `false` without `admit-share`.
2041fn is_reduced(refusal: &Refusal) -> bool {
2042    #[cfg(feature = "admit-share")]
2043    {
2044        matches!(refusal, Refusal::ReducedShare { .. })
2045    }
2046    #[cfg(not(feature = "admit-share"))]
2047    {
2048        let _ = refusal;
2049        false
2050    }
2051}
2052
2053/// A held transfer slot **and** its memory reservation. Releasing both is
2054/// [`Drop`], not a method, so no error path — and no panic — can skip it.
2055///
2056/// That is the whole answer to "an operation that ERRORS must still release":
2057/// there is no release call to forget, because unwinding runs `Drop`.
2058#[derive(Debug)]
2059pub struct AdmissionGuard {
2060    gate: Arc<Admission>,
2061    /// `None` when there is no per-principal limit, so the map is never touched
2062    /// and never grown by a configuration that does not use it.
2063    principal: Option<String>,
2064    /// `None` on an unlimited gate, where there is neither a slot nor a
2065    /// reservation to give back.
2066    claim: Option<Claim>,
2067    marked: ServiceMark,
2068    /// Bytes seen on this transfer's wire, in both directions.
2069    wire: AtomicU64,
2070    /// What has been charged to the ledger on top of [`Claim::bytes`]. The
2071    /// guard owns this so [`Drop`] releases exactly what was taken.
2072    grown: AtomicU64,
2073    /// The policy's guard (`admit-share`), handed to `record` on drop with
2074    /// [`AdmissionGuard::wire`] as the cost — the counter the memory gate
2075    /// already keeps, so recording adds nothing per byte.
2076    #[cfg(feature = "admit-share")]
2077    admitted: Option<edda_admit::Guard>,
2078}
2079
2080impl AdmissionGuard {
2081    /// **The operation's work has actually started.** Files this item's sojourn
2082    /// against its class's window.
2083    ///
2084    /// Called by the caller and not by the gate, because only the caller knows
2085    /// when the waiting stopped. In `crate::server` that is the first line of
2086    /// the spawned transfer task: everything between admission and there is the
2087    /// runtime's ready queue and the blocking-pool handoff, which is exactly the
2088    /// invisible queue this whole module exists to make visible.
2089    ///
2090    /// One-shot: a second call is ignored rather than filing a second, longer
2091    /// sample for the same item.
2092    pub fn entered_service(&self) {
2093        let (Some(_governor), Some(claim)) = (&self.gate.governor, &self.claim) else {
2094            return;
2095        };
2096        if self.marked.claim() {
2097            let handoff = claim.enqueued.elapsed();
2098            let ms = u64::try_from(handoff.as_millis()).unwrap_or(u64::MAX);
2099            self.gate
2100                .counters
2101                .handoff_ms_total
2102                .fetch_add(ms, Ordering::Relaxed);
2103            self.gate
2104                .counters
2105                .peak_handoff_ms
2106                .fetch_max(ms, Ordering::Relaxed);
2107        }
2108    }
2109
2110    /// The class this operation was admitted as.
2111    pub fn class(&self) -> Option<OpClass> {
2112        self.claim.as_ref().map(|claim| claim.class)
2113    }
2114
2115    /// **Charge this transfer for `bytes` more on the wire.**
2116    ///
2117    /// Called by the transport as bytes actually move, in either direction, and
2118    /// this is what makes the memory gate cost-based rather than flat: the
2119    /// admission reservation ([`OpClass::admission_bytes`]) is the entry cost,
2120    /// and everything proportional to the transfer arrives here.
2121    ///
2122    /// Charges [`WIRE_MEMORY_NUMERATOR`]/[`WIRE_MEMORY_DENOMINATOR`] of the
2123    /// bytes, in steps of [`WIRE_RESERVE_STEP`], so a per-byte quantity never
2124    /// becomes a per-byte decision. Returns `Some(Growth)` on the calls that
2125    /// actually charged and `None` in between, so a caller can log the
2126    /// overdraw without testing for it.
2127    ///
2128    /// It cannot refuse — see [`Growth`]. What an overdraw does is stop *new*
2129    /// admissions until this transfer finishes.
2130    pub fn reserve_more(&self, bytes: u64) -> Option<Growth> {
2131        let (Some(governor), Some(_)) = (&self.gate.governor, &self.claim) else {
2132            return None;
2133        };
2134        let wire = self.wire.fetch_add(bytes, Ordering::Relaxed) + bytes;
2135        let want = wire * WIRE_MEMORY_NUMERATOR / WIRE_MEMORY_DENOMINATOR / WIRE_RESERVE_STEP
2136            * WIRE_RESERVE_STEP;
2137        let held = self.grown.load(Ordering::Relaxed);
2138        if want <= held {
2139            return None;
2140        }
2141        let extra = want - held;
2142        // `fetch_max`, not a store: two directions can be metering at once and
2143        // the ledger must be charged once per step, not once per racer.
2144        if self.grown.fetch_max(want, Ordering::Relaxed) >= want {
2145            return None;
2146        }
2147        self.gate
2148            .counters
2149            .grown_bytes
2150            .fetch_add(extra, Ordering::Relaxed);
2151        let growth = governor.charge(want, extra);
2152        if matches!(growth, Growth::Overdrawn { .. }) {
2153            self.gate.counters.overdrawn.fetch_add(1, Ordering::Relaxed);
2154        }
2155        Some(growth)
2156    }
2157}
2158
2159impl Drop for AdmissionGuard {
2160    fn drop(&mut self) {
2161        // ONE record per transfer, here, off the stream: the wire total is the
2162        // counter `reserve_more` already keeps for the memory gate, and its
2163        // direction is the class's — a push's bytes came in, everything
2164        // else's went out. Nothing was added to the per-byte path for it.
2165        #[cfg(feature = "admit-share")]
2166        if let Some(admitted) = self.admitted.take() {
2167            let wire = self.wire.load(Ordering::Relaxed);
2168            let push = matches!(self.claim.as_ref().map(|c| c.class), Some(OpClass::Push));
2169            self.gate.policy.0.record(
2170                admitted,
2171                &edda_admit::Cost {
2172                    net_in: if push { wire } else { 0 },
2173                    net_out: if push { 0 } else { wire },
2174                    ..edda_admit::Cost::default()
2175                },
2176            );
2177        }
2178        if let (Some(governor), Some(claim)) = (&self.gate.governor, &self.claim) {
2179            governor.release(claim, self.grown.load(Ordering::Relaxed));
2180            // A transfer that never said its work started gave the controller
2181            // nothing. Counted rather than guessed at: filing the whole
2182            // lifetime as a "sojourn" here would be worse than filing nothing,
2183            // because service time is not wait time and every long clone would
2184            // read as a standing queue.
2185            if !self.marked.claimed() {
2186                self.gate
2187                    .counters
2188                    .unmeasured_sojourn
2189                    .fetch_add(1, Ordering::Relaxed);
2190            }
2191        }
2192        self.gate.release(self.principal.as_deref());
2193        // The slot is free now, so anybody parked in `admit_within` should
2194        // re-ask. `notify_waiters` wakes everyone parked *at this moment* and
2195        // stores nothing for a future waiter, which is right: a waiter that
2196        // arrives later runs `try_admit` itself before it ever parks.
2197        self.gate.slot_freed.notify_waiters();
2198    }
2199}
2200
2201/// One operation's presence in [`Governor::waiting_in`], released on every
2202/// path — return, refusal, or a dropped future.
2203struct Parked<'a> {
2204    governor: Option<&'a Governor>,
2205    lane: Lane,
2206}
2207
2208impl<'a> Parked<'a> {
2209    fn new(governor: Option<&'a Governor>, lane: Lane) -> Parked<'a> {
2210        if let Some(governor) = governor {
2211            governor.entered_wait(lane);
2212        }
2213        Parked { governor, lane }
2214    }
2215}
2216
2217impl Drop for Parked<'_> {
2218    fn drop(&mut self) {
2219        if let Some(governor) = self.governor {
2220            governor.left_wait(self.lane);
2221        }
2222    }
2223}
2224
2225/// **The wire meter**: an `AsyncRead + AsyncWrite` that charges its
2226/// [`AdmissionGuard`] for every byte that crosses it.
2227///
2228/// This is the other half of cost-based admission. [`OpClass::admission_bytes`]
2229/// is what an operation reserves before anybody knows how big it is; this is
2230/// how the ledger finds out. It wraps the same duplex the transport was already
2231/// given, so there is no second copy of the bytes and no second place where a
2232/// transfer's size is known.
2233///
2234/// Both directions are metered through one counter: a push's cost is the pack
2235/// coming in and a clone's is the pack going out, and the ledger does not care
2236/// which way a byte was travelling when it made the server hold something.
2237///
2238/// Zero cost when it is not doing anything: [`AdmissionGuard::reserve_more`]
2239/// returns on an integer comparison until a whole [`WIRE_RESERVE_STEP`] has
2240/// gone by.
2241pub struct Metered<S> {
2242    inner: S,
2243    guard: Arc<AdmissionGuard>,
2244}
2245
2246impl<S> Metered<S> {
2247    /// Meter `inner` against `guard`.
2248    pub fn new(inner: S, guard: Arc<AdmissionGuard>) -> Metered<S> {
2249        Metered { inner, guard }
2250    }
2251
2252    /// Charge for `bytes`, and say so once when the charge did not fit.
2253    fn charge(&self, bytes: usize) {
2254        let Some(growth) = self.guard.reserve_more(bytes as u64) else {
2255            return;
2256        };
2257        if let Growth::Overdrawn {
2258            held,
2259            reserved,
2260            budget,
2261        } = growth
2262        {
2263            tracing::warn!(
2264                target: "gunnar::git",
2265                event = "git.transfer.overdrawn",
2266                held_bytes = held,
2267                reserved_bytes = reserved,
2268                budget_bytes = budget,
2269                "this transfer has now charged the memory budget past its limit; it finishes, \
2270                 and NEW transfers are refused by name until it does"
2271            );
2272        }
2273    }
2274}
2275
2276impl<S: tokio::io::AsyncRead + Unpin> tokio::io::AsyncRead for Metered<S> {
2277    fn poll_read(
2278        mut self: std::pin::Pin<&mut Self>,
2279        cx: &mut std::task::Context<'_>,
2280        buf: &mut tokio::io::ReadBuf<'_>,
2281    ) -> std::task::Poll<std::io::Result<()>> {
2282        let before = buf.filled().len();
2283        let polled = std::pin::Pin::new(&mut self.inner).poll_read(cx, buf);
2284        if polled.is_ready() {
2285            let read = buf.filled().len().saturating_sub(before);
2286            if read > 0 {
2287                self.charge(read);
2288            }
2289        }
2290        polled
2291    }
2292}
2293
2294impl<S: tokio::io::AsyncWrite + Unpin> tokio::io::AsyncWrite for Metered<S> {
2295    fn poll_write(
2296        mut self: std::pin::Pin<&mut Self>,
2297        cx: &mut std::task::Context<'_>,
2298        buf: &[u8],
2299    ) -> std::task::Poll<std::io::Result<usize>> {
2300        let polled = std::pin::Pin::new(&mut self.inner).poll_write(cx, buf);
2301        if let std::task::Poll::Ready(Ok(written)) = &polled {
2302            let written = *written;
2303            if written > 0 {
2304                self.charge(written);
2305            }
2306        }
2307        polled
2308    }
2309
2310    fn poll_flush(
2311        mut self: std::pin::Pin<&mut Self>,
2312        cx: &mut std::task::Context<'_>,
2313    ) -> std::task::Poll<std::io::Result<()>> {
2314        std::pin::Pin::new(&mut self.inner).poll_flush(cx)
2315    }
2316
2317    fn poll_shutdown(
2318        mut self: std::pin::Pin<&mut Self>,
2319        cx: &mut std::task::Context<'_>,
2320    ) -> std::task::Poll<std::io::Result<()>> {
2321        std::pin::Pin::new(&mut self.inner).poll_shutdown(cx)
2322    }
2323}
2324
2325#[cfg(test)]
2326mod tests {
2327    use super::*;
2328
2329    /// **A keyed caller is QUEUED past the short wait; a stranger is not.**
2330    ///
2331    /// gunnar.rs, 2026-10-25: every lane pushes and fetches as one key, and
2332    /// git-lfs cannot retry a refused pure-SSH `git-lfs-transfer` — it falls
2333    /// back to `git-lfs-authenticate` and fails the whole push ("batch
2334    /// request: exec request failed"). So an authenticated identity waits for
2335    /// a slot up to the keyed wait (30 s on the appliance, under `exec_grace`'s
2336    /// 60 s), while anonymous keeps the short wait and its never-park rules.
2337    #[tokio::test]
2338    async fn a_keyed_caller_waits_out_the_keyed_bound_and_a_stranger_does_not() {
2339        let gate = Arc::new(
2340            Admission::new(Some(1), None)
2341                .with_wait(Duration::from_millis(50))
2342                .with_keyed_wait(Some(Duration::from_secs(5))),
2343        );
2344        assert_eq!(gate.wait_for("admin"), Duration::from_secs(5));
2345        assert_eq!(
2346            gate.wait_for(gunnar_core::ANONYMOUS_PRINCIPAL_ID),
2347            Duration::from_millis(50)
2348        );
2349        let holder = gate
2350            .try_admit("other", "team", OpClass::Fetch)
2351            .expect("the one slot");
2352
2353        // The slot frees after 400 ms: well past the short wait, inside the keyed one.
2354        let freer = tokio::spawn(async move {
2355            tokio::time::sleep(Duration::from_millis(400)).await;
2356            drop(holder);
2357        });
2358        let started = std::time::Instant::now();
2359        let (guard, waited) = gate
2360            .admit_within("admin", "team", OpClass::Fetch)
2361            .await
2362            .expect("a keyed caller is queued for the slot, not refused after 50 ms");
2363        assert!(
2364            waited >= Duration::from_millis(300),
2365            "it really waited: {waited:?}"
2366        );
2367        assert!(started.elapsed() < Duration::from_secs(5));
2368        freer.await.unwrap();
2369
2370        // The slot is held again: a stranger is refused at the short bound.
2371        let started = std::time::Instant::now();
2372        gate.admit_within(gunnar_core::ANONYMOUS_PRINCIPAL_ID, "team", OpClass::Fetch)
2373            .await
2374            .expect_err("anonymous is never queued for the keyed bound");
2375        assert!(
2376            started.elapsed() < Duration::from_secs(1),
2377            "{:?}",
2378            started.elapsed()
2379        );
2380        drop(guard);
2381
2382        // `--admission-wait-ms 0` still means "refuse on the spot" for everyone.
2383        let instant = Admission::new(Some(1), None)
2384            .with_wait(Duration::ZERO)
2385            .with_keyed_wait(Some(Duration::from_secs(30)));
2386        assert_eq!(instant.wait_for("admin"), Duration::ZERO);
2387        // No keyed bound configured: the keyed caller gets the ordinary wait.
2388        let plain = Admission::new(Some(1), None).with_wait(Duration::from_millis(50));
2389        assert_eq!(plain.wait_for("admin"), Duration::from_millis(50));
2390    }
2391
2392    /// **The drain's door** (`DATA-SET-GROWTH-FLOW.md` §3 step 5, T1/T12):
2393    /// a paused gate refuses every new transfer with the growing sentence,
2394    /// takes no slot, counts the refusal under its own counter, never parks
2395    /// the caller, and leaves what was already admitted in flight — which
2396    /// is the number the drain waits on. Unpaused, it admits again.
2397    #[tokio::test]
2398    async fn a_paused_gate_refuses_with_the_growing_words_and_keeps_in_flight_honest() {
2399        let gate = Arc::new(Admission::new(Some(4), None).with_wait(Duration::from_secs(5)));
2400        let running = gate
2401            .try_admit("one", "team", OpClass::Push)
2402            .expect("admitted before the pause");
2403        assert_eq!(gate.in_flight(), 1);
2404        assert_eq!(gate.paused(), None);
2405
2406        gate.pause(Pause::Growing);
2407        assert_eq!(gate.paused(), Some(Pause::Growing));
2408        let refused = gate
2409            .try_admit("two", "team", OpClass::Clone)
2410            .expect_err("paused: refused");
2411        assert_eq!(
2412            refused,
2413            Refusal::Paused {
2414                pause: Pause::Growing
2415            }
2416        );
2417        assert_eq!(refused.client_message(), GROWING_MESSAGE);
2418        assert!(
2419            refused.client_message().contains("growing"),
2420            "the word is load-bearing"
2421        );
2422        assert!(refused.client_message().contains("try again"));
2423        // ★ Every estate says it, not only gunnar.rs (the UpCloud mockup, 2026-10-09: test2's
2424        // pushes were told "gunnar.rs is growing its storage"): no estate's domain is spelled.
2425        assert!(
2426            !refused.client_message().contains("gunnar.rs"),
2427            "a test2 push was told gunnar.rs is growing: {}",
2428            refused.client_message()
2429        );
2430        assert!(
2431            refused
2432                .client_message()
2433                .starts_with("The server is growing its storage"),
2434            "{}",
2435            refused.client_message()
2436        );
2437        assert_eq!(refused.reason(), "paused-growing");
2438        for other in [
2439            "concurrent git transfers",
2440            "per-principal limit",
2441            "unauthenticated",
2442        ] {
2443            assert!(
2444                !refused.client_message().contains(other),
2445                "shares no substring with {other}"
2446            );
2447        }
2448        // The waiting door does not park a paused caller: it answers at once.
2449        let started = std::time::Instant::now();
2450        let parked = gate
2451            .admit_within("three", "team", OpClass::Fetch)
2452            .await
2453            .expect_err("paused: refused, not parked");
2454        assert!(matches!(parked, Refusal::Paused { .. }));
2455        assert!(
2456            started.elapsed() < Duration::from_secs(1),
2457            "not parked for the wait"
2458        );
2459        // Counted under its own counter, in the total, and no slot was taken.
2460        let stats = gate.stats();
2461        assert_eq!(stats.refused_paused, 2);
2462        assert_eq!(stats.refused(), 2);
2463        assert_eq!(
2464            stats.in_flight, 1,
2465            "the pre-pause transfer is still what is in flight"
2466        );
2467        gate.pause(Pause::Growing);
2468        assert_eq!(gate.paused(), Some(Pause::Growing), "idempotent");
2469
2470        drop(running);
2471        assert_eq!(
2472            gate.in_flight(),
2473            0,
2474            "the drain sees zero once the last one finishes"
2475        );
2476        gate.unpause();
2477        gate.unpause();
2478        assert_eq!(gate.paused(), None);
2479        let _again = gate
2480            .try_admit("two", "team", OpClass::Clone)
2481            .expect("unpaused: admitted");
2482        assert_eq!(gate.in_flight(), 1);
2483    }
2484
2485    #[test]
2486    fn the_server_wide_limit_refuses_and_counts() {
2487        let gate = Arc::new(Admission::new(Some(2), None));
2488        let a = gate
2489            .try_admit("one", "team", OpClass::LsRemote)
2490            .expect("the first is admitted");
2491        let b = gate
2492            .try_admit("two", "team", OpClass::LsRemote)
2493            .expect("the second is admitted");
2494        let refused = gate
2495            .try_admit("three", "team", OpClass::LsRemote)
2496            .expect_err("the third is refused");
2497        assert_eq!(refused, Refusal::AtCapacity { limit: 2 });
2498
2499        let stats = gate.stats();
2500        assert_eq!(stats.admitted, 2, "two got in");
2501        assert_eq!(stats.refused_at_capacity, 1, "the refusal must be COUNTED");
2502        assert_eq!(stats.in_flight, 2);
2503        assert_eq!(stats.peak_in_flight, 2);
2504
2505        // …and the slot comes back, so this is a limit and not a budget.
2506        drop(a);
2507        assert_eq!(gate.stats().in_flight, 1);
2508        let c = gate
2509            .try_admit("three", "team", OpClass::LsRemote)
2510            .expect("a freed slot is reusable");
2511        assert_eq!(gate.stats().in_flight, 2);
2512        assert_eq!(
2513            gate.stats().peak_in_flight,
2514            2,
2515            "the peak is a high-water mark, not the current value"
2516        );
2517        drop((b, c));
2518        assert_eq!(gate.stats().in_flight, 0);
2519    }
2520
2521    /// **Direction A of the max-min bar.** A single principal must be able to
2522    /// reach the *server-wide* limit on an otherwise idle server.
2523    ///
2524    /// This is the bug, at the shipped ratio. Before the fix the per-principal
2525    /// number was a wall, so this run stopped at 32 with the rest of the server
2526    /// free, and the whole limit needed eight identities to reach.
2527    ///
2528    /// # What the governor changed here, and what it did not
2529    ///
2530    /// The number a lone principal can reach is now the **governed** limit —
2531    /// the top of [`LimitBand::for_ceiling`] of the configured ceiling — and not
2532    /// the configured ceiling itself. With the shipped
2533    /// `DEFAULT_MAX_CONCURRENT_TRANSFERS` of 256 that is
2534    /// [`MAX_LIMIT`](crate::governor::MAX_LIMIT), 100: 256 concurrent git
2535    /// transfers is the sort of number the governor exists to stop a server
2536    /// from believing about itself.
2537    ///
2538    /// The property under test is unchanged and is still the one that matters:
2539    /// **the per-principal number is a floor, not a wall**, so one identity on
2540    /// an idle server reaches the whole of whatever the server-wide limit
2541    /// currently is. It is asserted against `governed` rather than a literal so
2542    /// it cannot drift away from the band.
2543    #[test]
2544    fn one_principal_alone_reaches_the_server_wide_limit() {
2545        use crate::config::{
2546            DEFAULT_MAX_CONCURRENT_TRANSFERS as CONFIGURED,
2547            DEFAULT_MAX_TRANSFERS_PER_PRINCIPAL as FLOOR,
2548        };
2549        let gate = Arc::new(Admission::new(Some(CONFIGURED), Some(FLOOR)));
2550        let governed = gate.governed_limit().expect("a governed limit");
2551        // The top of the band is the CONFIGURED ceiling, not the design's
2552        // default. This assertion read `MAX_LIMIT` while the band was capped at
2553        // 100, which made a server told 256 serve 100 and called it correct —
2554        // the test agreed with the bug because both had the same number in them.
2555        assert_eq!(
2556            governed, CONFIGURED,
2557            "a fresh governor starts at the top of its band, and the top is what \
2558             the operator configured"
2559        );
2560        assert!(
2561            governed > FLOOR,
2562            "a test where the governed limit is not above the per-principal floor cannot tell a \
2563             floor from a wall"
2564        );
2565
2566        let held: Vec<_> = (0..governed)
2567            .map(|i| {
2568                gate.try_admit("solo", "team", OpClass::LsRemote)
2569                    .unwrap_or_else(|refusal| {
2570                        panic!(
2571                            "transfer {i} of {governed} was refused on an EMPTY server holding \
2572                             one identity: {refusal}. The per-principal number is a floor, not a \
2573                             wall; a single-key client that stops at {FLOOR} is the old invisible ceiling, back"
2574                        )
2575                    })
2576            })
2577            .collect();
2578        assert_eq!(held.len(), governed);
2579
2580        let stats = gate.stats();
2581        assert_eq!(
2582            stats.admitted, governed as u64,
2583            "the server-wide limit in force must be reachable by ONE identity"
2584        );
2585        assert_eq!(
2586            stats.refused_per_principal, 0,
2587            "not one of these {governed} may be refused for fairness: there was nobody to be \
2588             fair to"
2589        );
2590        assert_eq!(stats.peak_in_flight, governed as u64);
2591        assert_eq!(stats.limit, governed, "the stats must report what refuses");
2592
2593        // …and the server-wide gate is still a gate. Without this the test
2594        // above would also pass on an Admission that had lost both limits.
2595        let full = gate
2596            .try_admit("solo", "team", OpClass::LsRemote)
2597            .expect_err("one past the governed limit");
2598        assert_eq!(full, Refusal::AtCapacity { limit: governed });
2599        drop(held);
2600    }
2601
2602    /// **Direction B of the max-min bar.** Once the server IS contended, a
2603    /// greedy principal is held to its share and a quiet one is still served.
2604    ///
2605    /// `(4, 1)` so the contention threshold (75 % → 3 of 4) is reachable in a
2606    /// handful of admissions, and so the share (`4 / 2 = 2`) is strictly above
2607    /// the configured floor of 1 — a share equal to the floor could not tell
2608    /// max-min fairness from the old wall.
2609    #[test]
2610    fn a_greedy_principal_is_held_to_its_share_once_the_server_is_contended() {
2611        let gate = Arc::new(Admission::new(Some(4), Some(1)));
2612
2613        // The quiet one arrives first, so it is *active* and has a claim.
2614        let _polite = gate
2615            .try_admit("polite", "team", OpClass::LsRemote)
2616            .expect("polite's first");
2617        // Greedy takes its share: 4 slots / 2 identities = 2.
2618        let _g1 = gate
2619            .try_admit("greedy", "team", OpClass::LsRemote)
2620            .expect("greedy's first");
2621        let _g2 = gate
2622            .try_admit("greedy", "team", OpClass::LsRemote)
2623            .expect("greedy's second is still its share");
2624
2625        // Three of four are held, so the server is over the 75 % threshold and
2626        // fairness is in force. Greedy has its two.
2627        let refused = gate
2628            .try_admit("greedy", "team", OpClass::LsRemote)
2629            .expect_err("greedy's third is past its share of a contended server");
2630        assert_eq!(
2631            refused,
2632            Refusal::PrincipalAtCapacity {
2633                limit: 2,
2634                active_principals: 2
2635            },
2636            "the refusal must name the SHARE in force (4/2), not the configured floor (1), or an \
2637             operator reads a number that matches nothing they set"
2638        );
2639        assert!(
2640            refused.client_message().contains("per-principal limit"),
2641            "installtest's `refusals_in_log` tells the two refusals apart by this exact phrase; \
2642             losing it silently zeroes the server-side witness: {}",
2643            refused.client_message()
2644        );
2645
2646        // The whole reason fairness exists: the fourth slot was NOT taken by
2647        // greedy, and it is there for somebody else.
2648        let _other = gate
2649            .try_admit("polite", "team", OpClass::LsRemote)
2650            .expect("the slot greedy was refused must be available to the quiet principal");
2651
2652        let stats = gate.stats();
2653        assert_eq!(stats.refused_per_principal, 1);
2654        assert_eq!(
2655            stats.refused_at_capacity, 0,
2656            "the server had a free slot; attributing this to capacity sends an operator to raise \
2657             the wrong number"
2658        );
2659        assert_eq!(stats.in_flight, 4);
2660    }
2661
2662    /// The share is *work-conserving*: two identities on a server that is
2663    /// nowhere near full do not each get half of it.
2664    ///
2665    /// Without the contention threshold this is what max-min fairness costs —
2666    /// a busy principal capped at `server_wide / active` while most of the box
2667    /// sits idle, which is the same "binds when there is nobody to be fair to"
2668    /// mistake in a smaller costume.
2669    #[test]
2670    fn fairness_does_not_bind_while_there_are_free_slots() {
2671        let gate = Arc::new(Admission::new(Some(100), Some(1)));
2672        let _quiet = gate
2673            .try_admit("quiet", "team", OpClass::LsRemote)
2674            .expect("one transfer from a quiet identity");
2675
2676        // An equal share of 100 between 2 identities is 50. The server is only
2677        // going to 74 % full, so the share must not be enforced at all.
2678        let busy: Vec<_> = (0..73)
2679            .map(|i| {
2680                gate.try_admit("busy", "team", OpClass::LsRemote)
2681                    .unwrap_or_else(|r| panic!("transfer {i} refused at {}% full: {r}", i + 2))
2682            })
2683            .collect();
2684        assert_eq!(
2685            busy.len(),
2686            73,
2687            "74 of 100 held, so nothing may be refused yet"
2688        );
2689        assert_eq!(gate.stats().refused_per_principal, 0);
2690
2691        // The very next one crosses the threshold — 75 of 100 occupied — and
2692        // the share engages at once: 100/2 = 50, and `busy` is already at 73.
2693        let refused = gate
2694            .try_admit("busy", "team", OpClass::LsRemote)
2695            .expect_err(
2696                "the transfer that makes the server 75% full is the first one weighed \
2697                         against the share",
2698            );
2699        assert_eq!(
2700            refused,
2701            Refusal::PrincipalAtCapacity {
2702                limit: 50,
2703                active_principals: 2
2704            }
2705        );
2706    }
2707
2708    /// The pinned mode: an operator who wants the old wall still has it.
2709    #[test]
2710    fn a_hard_cap_binds_on_an_empty_server() {
2711        let fair = Arc::new(Admission::new(Some(64), Some(2)));
2712        let _a = fair
2713            .try_admit("solo", "team", OpClass::LsRemote)
2714            .expect("first");
2715        let _b = fair
2716            .try_admit("solo", "team", OpClass::LsRemote)
2717            .expect("second");
2718        fair.try_admit("solo", "team", OpClass::LsRemote)
2719            .expect("a fair gate lets a lone principal past its floor on an idle server");
2720
2721        let hard = Arc::new(Admission::hard_capped(Some(64), Some(2)));
2722        let _c = hard
2723            .try_admit("solo", "team", OpClass::LsRemote)
2724            .expect("first");
2725        let _d = hard
2726            .try_admit("solo", "team", OpClass::LsRemote)
2727            .expect("second");
2728        let refused = hard
2729            .try_admit("solo", "team", OpClass::LsRemote)
2730            .expect_err("a pinned cap binds whether or not anyone else is there");
2731        assert_eq!(
2732            refused,
2733            Refusal::PrincipalAtCapacity {
2734                limit: 2,
2735                active_principals: 1
2736            }
2737        );
2738        assert!(
2739            !refused.client_message().contains("fair share"),
2740            "a pinned cap is not a share and must not be explained as one: {}",
2741            refused.client_message()
2742        );
2743        assert_eq!(hard.stats().refused_per_principal, 1);
2744        assert_eq!(fair.stats().refused_per_principal, 0);
2745    }
2746
2747    #[test]
2748    fn a_refused_transfer_does_not_leak_a_server_wide_slot() {
2749        // The subtle one: the server-wide permit is taken BEFORE the
2750        // per-principal check, so a per-principal refusal must give it back. If
2751        // it did not, a client hammering its own limit would drain the server.
2752        //
2753        // Driven through the *fair* path rather than a hard cap, because that
2754        // is what ships: `polite` first so the server is contended and greedy's
2755        // share (4/2 = 2) is what it hammers.
2756        let gate = Arc::new(Admission::new(Some(4), Some(1)));
2757        let _polite = gate
2758            .try_admit("polite", "team", OpClass::LsRemote)
2759            .expect("polite is active");
2760        let _g1 = gate
2761            .try_admit("greedy", "team", OpClass::LsRemote)
2762            .expect("greedy's share is 2");
2763        let _g2 = gate
2764            .try_admit("greedy", "team", OpClass::LsRemote)
2765            .expect("greedy's share is 2");
2766        for _ in 0..100 {
2767            gate.try_admit("greedy", "team", OpClass::LsRemote)
2768                .expect_err("refused every time");
2769        }
2770        // Three slots are held; the fourth must still be free for anybody at all.
2771        let _other = gate
2772            .try_admit("somebody-else", "team", OpClass::LsRemote)
2773            .expect("the fourth slot survived 100 refusals");
2774        assert_eq!(gate.stats().in_flight, 4);
2775        assert_eq!(gate.stats().refused_per_principal, 100);
2776    }
2777
2778    /// A principal that finishes its work must give its share back to the
2779    /// others, or the map's length — the divisor of every share — is a
2780    /// high-water mark rather than a live count.
2781    #[test]
2782    fn a_departed_principal_stops_shrinking_everybody_elses_share() {
2783        let gate = Arc::new(Admission::new(Some(4), Some(1)));
2784        let leaving = gate
2785            .try_admit("leaving", "team", OpClass::LsRemote)
2786            .expect("two identities are active");
2787        let _g1 = gate
2788            .try_admit("stayer", "team", OpClass::LsRemote)
2789            .expect("first");
2790        let _g2 = gate
2791            .try_admit("stayer", "team", OpClass::LsRemote)
2792            .expect("share of 2 with 2 active");
2793        gate.try_admit("stayer", "team", OpClass::LsRemote)
2794            .expect_err("held to 2 while `leaving` is still there");
2795
2796        drop(leaving);
2797        // Now one identity is active, so the share is the whole server again.
2798        let _g3 = gate
2799            .try_admit("stayer", "team", OpClass::LsRemote)
2800            .expect("with the other identity gone the share must widen back out");
2801        let _g4 = gate
2802            .try_admit("stayer", "team", OpClass::LsRemote)
2803            .expect("all four slots are reachable by the only identity left");
2804        assert_eq!(gate.stats().in_flight, 4);
2805    }
2806
2807    #[test]
2808    fn the_map_does_not_grow_with_every_principal_ever_seen() {
2809        let gate = Arc::new(Admission::new(Some(4), Some(1)));
2810        for i in 0..500 {
2811            let guard = gate
2812                .try_admit(&format!("principal-{i}"), "team", OpClass::LsRemote)
2813                .expect("admitted");
2814            drop(guard);
2815        }
2816        // Read through the gate's own lock rather than a counter, because the
2817        // thing being asserted is that the MAP is empty; a counter could agree
2818        // while the map leaked.
2819        let held = gate.per_principal.lock().expect("the per-principal map");
2820        assert!(
2821            held.is_empty(),
2822            "the per-principal map kept {} entries for principals with nothing in flight",
2823            held.len()
2824        );
2825    }
2826
2827    /// The three refusals must stay **tellable apart by their words**, because
2828    /// that is how the only cross-process witness of them counts.
2829    ///
2830    /// `installtest`'s `refusals_in_log` reads the server log and sorts
2831    /// refusals by the substrings `"concurrent git transfers"` and
2832    /// `"per-principal limit"` — the log records a refusal by its `Display`,
2833    /// which is the same sentence the client is handed. A third reason that
2834    /// collided with either phrase would be silently counted as the wrong one,
2835    /// and an operator would raise the wrong number: the concurrency limit,
2836    /// when the thing that ran out was memory.
2837    ///
2838    /// # Seen red
2839    ///
2840    /// By giving `Refusal::MemoryBudget` the natural wording, "the server is
2841    /// already serving its limit of N concurrent git transfers within its
2842    /// memory budget". Failure:
2843    ///
2844    /// ```text
2845    /// the memory refusal contains "concurrent git transfers", which
2846    /// installtest counts as a CAPACITY refusal; an operator reading that
2847    /// raises the concurrency limit and makes the memory problem worse
2848    /// ```
2849    #[test]
2850    fn the_three_refusals_do_not_borrow_each_others_words() {
2851        let capacity = Refusal::AtCapacity { limit: 37 }.client_message();
2852        let principal = Refusal::PrincipalAtCapacity {
2853            limit: 4,
2854            active_principals: 3,
2855        }
2856        .client_message();
2857        let memory = Refusal::MemoryBudget {
2858            class: OpClass::Clone,
2859            wanted: 2048 * 1024 * 1024,
2860            reserved: 6144 * 1024 * 1024,
2861            budget: 8192 * 1024 * 1024,
2862        }
2863        .client_message();
2864
2865        assert!(capacity.contains("concurrent git transfers"));
2866        assert!(!capacity.contains("per-principal limit"));
2867        assert!(principal.contains("per-principal limit"));
2868        assert!(!principal.contains("concurrent git transfers"));
2869
2870        assert!(
2871            !memory.contains("concurrent git transfers"),
2872            "the memory refusal contains \"concurrent git transfers\", which installtest counts \
2873             as a CAPACITY refusal; an operator reading that raises the concurrency limit and \
2874             makes the memory problem worse: {memory}"
2875        );
2876        assert!(
2877            !memory.contains("per-principal limit"),
2878            "the memory refusal contains \"per-principal limit\", so installtest counts a \
2879             server-wide memory exhaustion as one identity's fairness throttle: {memory}"
2880        );
2881        assert!(
2882            memory.contains("memory budget"),
2883            "the memory refusal must name the knob to turn: {memory}"
2884        );
2885        // All three are transient and every one of them must say so, or a CI
2886        // log reads a busy server as a broken one.
2887        for message in [&capacity, &principal, &memory] {
2888            assert!(
2889                message.contains("try again"),
2890                "a transient refusal that does not say `try again` reads as permanent: {message}"
2891            );
2892        }
2893    }
2894
2895    /// The work an operation would do, if it were admitted.
2896    ///
2897    /// Deliberately something with an **observable cost**: it allocates, and it
2898    /// records that it allocated. "No work was performed" asserted against a
2899    /// bool nobody sets is a hollow guard; asserted against a
2900    /// byte count that a real `Vec` really pushed into is not.
2901    #[derive(Default)]
2902    struct Work {
2903        runs: AtomicU64,
2904        bytes_allocated: AtomicU64,
2905    }
2906
2907    impl Work {
2908        /// One operation's buffer. A clone's real one is proportional to the
2909        /// pack; this stands in for it at a size a test can afford.
2910        fn perform(&self) -> Vec<u8> {
2911            let buffer = vec![0u8; 1024 * 1024];
2912            self.runs.fetch_add(1, Ordering::Relaxed);
2913            self.bytes_allocated
2914                .fetch_add(buffer.len() as u64, Ordering::Relaxed);
2915            buffer
2916        }
2917
2918        fn runs(&self) -> u64 {
2919            self.runs.load(Ordering::Relaxed)
2920        }
2921
2922        fn bytes(&self) -> u64 {
2923            self.bytes_allocated.load(Ordering::Relaxed)
2924        }
2925    }
2926
2927    /// **A denial happens AT admission, and nothing the operation would have
2928    /// done gets done.**
2929    ///
2930    /// This is the entire value of the gate. A refusal that arrives after the
2931    /// buffer has been built is not load shedding, it is a slower way of paying
2932    /// the same cost — and for the clone class that cost is gigabytes
2933    /// (2314 MB peak RSS, measured 2026-08-13).
2934    ///
2935    /// Both gates are driven, because they refuse on different paths and each
2936    /// could grow a bug the other's test would not see.
2937    ///
2938    /// # Seen red
2939    ///
2940    /// By moving `Work::perform` above the `try_admit` in the loop below, which
2941    /// is precisely the mistake a real caller makes when it opens the pack
2942    /// before asking whether it may. Failure:
2943    ///
2944    /// ```text
2945    /// 30 operations were refused, but the work ran 40 times and allocated
2946    /// 41943040 bytes. A refusal after the allocation is not load shedding
2947    /// ```
2948    #[test]
2949    fn a_denial_is_paid_for_by_nobody() {
2950        // Gate 1: the concurrency limit. A pinned band of 10, so the refusals
2951        // are the governor's and not the memory ledger's.
2952        let gate = Arc::new(Admission::new(Some(10), None));
2953        let work = Work::default();
2954        let mut held = Vec::new();
2955        let mut refused = 0;
2956        for _ in 0..40 {
2957            match gate.try_admit("solo", "team", OpClass::LsRemote) {
2958                Ok(guard) => held.push((guard, work.perform())),
2959                Err(_) => refused += 1,
2960            }
2961        }
2962        assert_eq!(refused, 30, "10 of 40 fit under the limit");
2963        assert_eq!(
2964            work.runs(),
2965            10,
2966            "{refused} operations were refused, but the work ran {} times and allocated {} \
2967             bytes. A refusal after the allocation is not load shedding",
2968            work.runs(),
2969            work.bytes()
2970        );
2971        assert_eq!(work.bytes(), 10 * 1024 * 1024);
2972
2973        // Gate 2: the memory budget. Four clones' worth, and eight arrivals.
2974        let clone = OpClass::Clone.admission_bytes();
2975        let gate = Arc::new(Admission::new(None, None).with_budget(clone * 4));
2976        assert!(
2977            gate.memory_budget().is_none(),
2978            "an unlimited concurrency gate has no budget to attach one to; the memory gate must \
2979             ride on the same governor or a configuration could get one without the other"
2980        );
2981
2982        let gate = Arc::new(Admission::with_memory_budget(Some(100), None, clone * 4));
2983        let work = Work::default();
2984        let mut held = Vec::new();
2985        let mut refused = Vec::new();
2986        for _ in 0..8 {
2987            match gate.try_admit("solo", "team", OpClass::Clone) {
2988                Ok(guard) => held.push((guard, work.perform())),
2989                Err(refusal) => refused.push(refusal),
2990            }
2991        }
2992        assert_eq!(work.runs(), 4, "a 4-clone budget admits 4 clones");
2993        assert_eq!(
2994            work.bytes(),
2995            4 * 1024 * 1024,
2996            "the 4 refused clones allocated {} bytes between them, which is 4 more than a \
2997             refusal at admission can possibly cost",
2998            work.bytes() - 4 * 1024 * 1024
2999        );
3000        assert_eq!(refused.len(), 4);
3001        assert!(
3002            refused.iter().all(|r| matches!(
3003                r,
3004                Refusal::MemoryBudget {
3005                    class: OpClass::Clone,
3006                    ..
3007                }
3008            )),
3009            "the refusals must name MEMORY, or an operator raises the concurrency limit and \
3010             makes it worse: {refused:?}"
3011        );
3012        let stats = gate.stats();
3013        assert_eq!(stats.refused_memory, 4);
3014        assert_eq!(
3015            stats.refused_at_capacity, 0,
3016            "the limit was 100 and 4 were held"
3017        );
3018        assert_eq!(stats.reserved_bytes, clone * 4);
3019    }
3020
3021    /// **An operation that ERRORS — or panics — still releases both its slot
3022    /// and its reservation.**
3023    ///
3024    /// Leak this and the server wedges after N failures: a budget that only
3025    /// comes back on the happy path is a budget that drains to zero the first
3026    /// time a client hangs up mid-clone, which is the single most ordinary
3027    /// thing a git client does.
3028    ///
3029    /// # Seen red
3030    ///
3031    /// By making `AdmissionGuard::drop` release only on the success path —
3032    /// `if !std::thread::panicking() { … }` around the governor release, which
3033    /// is exactly how a hand-written release-on-completion would behave.
3034    /// Failure:
3035    ///
3036    /// ```text
3037    /// 6 operations ended in an error or a panic and 6442450944 bytes are still
3038    /// reserved of 8589934592; the 7th would be refused, so the server has
3039    /// wedged
3040    /// ```
3041    ///
3042    /// Three panics, 2 GiB each: three quarters of the whole budget gone to
3043    /// operations that no longer exist.
3044    #[test]
3045    fn an_operation_that_fails_still_gives_its_reservation_back() {
3046        let clone = OpClass::Clone.admission_bytes();
3047        let gate = Arc::new(Admission::with_memory_budget(Some(100), None, clone * 4));
3048
3049        for _ in 0..3 {
3050            // The error path: a guard that goes out of scope through `?`-style
3051            // early return rather than through completion.
3052            let outcome: Result<(), &str> = (|| {
3053                let _guard = gate
3054                    .try_admit("solo", "team", OpClass::Clone)
3055                    .map_err(|_| "refused")?;
3056                Err("the transport died mid-pack")
3057            })();
3058            assert!(outcome.is_err());
3059        }
3060        for _ in 0..3 {
3061            // The panic path, which no explicit release call survives.
3062            let panicked = std::panic::catch_unwind(std::panic::AssertUnwindSafe(|| {
3063                let _guard = gate
3064                    .try_admit("solo", "team", OpClass::Clone)
3065                    .expect("admitted before the panic");
3066                panic!("the pack walk blew up");
3067            }));
3068            assert!(panicked.is_err());
3069        }
3070
3071        let stats = gate.stats();
3072        assert_eq!(
3073            stats.reserved_bytes,
3074            0,
3075            "6 operations ended in an error or a panic and {} bytes are still reserved of {}; \
3076             the 7th would be refused, so the server has wedged",
3077            stats.reserved_bytes,
3078            gate.memory_budget().unwrap_or(0)
3079        );
3080        assert_eq!(stats.in_flight, 0);
3081        assert_eq!(
3082            gate.governor().expect("a governor").occupied(),
3083            0,
3084            "the concurrency slots leaked even though the reservations did not"
3085        );
3086
3087        // The applied output: the budget is genuinely reusable, four times over.
3088        let held: Vec<_> = (0..4)
3089            .map(|i| {
3090                gate.try_admit("solo", "team", OpClass::Clone)
3091                    .unwrap_or_else(|r| panic!("clone {i} after 6 failures was refused: {r}"))
3092            })
3093            .collect();
3094        assert_eq!(held.len(), 4);
3095    }
3096
3097    /// A transfer that never reports entering service is COUNTED, not guessed
3098    /// at.
3099    ///
3100    /// The controller only sees what [`AdmissionGuard::entered_service`] files.
3101    /// A caller that forgets leaves it with no samples, and a CoDel controller
3102    /// with no samples sits at the top of its band forever — a gate that looks
3103    /// like it works and does not. This is the counter that makes that visible,
3104    /// and `tests/admission.rs` is where the real SSH path is held to zero.
3105    ///
3106    /// # Seen red
3107    ///
3108    /// By making `AdmissionGuard::drop` increment the counter unconditionally
3109    /// instead of only when the mark was never claimed. Failure:
3110    ///
3111    /// ```text
3112    /// assertion `left == right` failed: a transfer that DID report entering
3113    /// service was counted as unmeasured: 2
3114    ///   left: 2
3115    ///  right: 1
3116    /// ```
3117    #[test]
3118    fn a_transfer_that_never_enters_service_is_counted_as_unmeasured() {
3119        let gate = Arc::new(Admission::new(Some(100), None));
3120
3121        drop(
3122            gate.try_admit("solo", "team", OpClass::LsRemote)
3123                .expect("admitted"),
3124        );
3125        assert_eq!(
3126            gate.stats().unmeasured_sojourn,
3127            1,
3128            "a transfer that never said its work started filed no sojourn sample and must be \
3129             counted; a controller starved of samples grows to its ceiling and stays there"
3130        );
3131
3132        let guard = gate
3133            .try_admit("solo", "team", OpClass::LsRemote)
3134            .expect("admitted");
3135        guard.entered_service();
3136        drop(guard);
3137        assert_eq!(
3138            gate.stats().unmeasured_sojourn,
3139            1,
3140            "a transfer that DID report entering service was counted as unmeasured: {}",
3141            gate.stats().unmeasured_sojourn
3142        );
3143
3144        // …and the sample really reached the controller's window, rather than
3145        // merely flipping a flag. One sample opens a window; nothing has closed
3146        // one yet, so the limit has not moved and the class's window is live.
3147        let governor = gate.governor().expect("a governor");
3148        assert_eq!(governor.codel().shrinks() + governor.codel().grows(), 0);
3149        let guard = gate
3150            .try_admit("solo", "team", OpClass::LsRemote)
3151            .expect("admitted");
3152        guard.entered_service();
3153        drop(guard);
3154        assert_eq!(
3155            gate.stats().unmeasured_sojourn,
3156            1,
3157            "the one-shot mark must not turn a second measured transfer into an unmeasured one"
3158        );
3159    }
3160
3161    #[test]
3162    fn an_unlimited_gate_admits_everything_and_still_counts() {
3163        let gate = Arc::new(Admission::unlimited());
3164        let mut held = Vec::new();
3165        for _ in 0..1000 {
3166            held.push(
3167                gate.try_admit("anyone", "team", OpClass::LsRemote)
3168                    .expect("no limit means no refusal"),
3169            );
3170        }
3171        let stats = gate.stats();
3172        assert_eq!(stats.admitted, 1000);
3173        assert_eq!(stats.refused(), 0);
3174        assert_eq!(stats.in_flight, 1000);
3175        assert_eq!(stats.peak_in_flight, 1000);
3176    }
3177}