crates/gunnar-ssh/src/governor.rs
//! **The two load-shedding gates**: a CoDel delay governor (time) and a memory
//! reservation ledger (space). Both refuse *at admission*, before a single byte
//! of the operation's work has been allocated.
//!
//! [`Admission`](crate::Admission) is the one door; this module is the pair of
//! locks on it. An operation is admitted only if **both** pass:
//!
//! ```text
//! admit(class):
//! if occupied >= governed_limit -> DENY (time: the queue is standing)
//! if reserved + est(class) > budget -> DENY (space: the budget is spoken for)
//! occupied += 1; reserved += est(class) // released by Drop, on every path
//! ```
//!
//! # Gate 1 — CoDel, because it is the one that does not oscillate
//!
//! ## Sojourn, not queue length
//!
//! Queue *length* is not a saturation signal. A hundred-deep queue draining in
//! a millisecond is a healthy server doing its job; a three-deep queue that has
//! not moved in a second is a server that has stopped. What separates them is
//! how long an item **waited**, so that is what is measured: the interval
//! between admission (enqueue) and the moment the operation's work actually
//! starts running ([`AdmissionGuard::entered_service`](crate::AdmissionGuard::entered_service)).
//!
//! In gunnar that interval is a real queue and not a bookkeeping fiction. Every
//! transfer is `tokio::spawn`ed and then bridged onto a blocking-pool thread
//! (`crate::admission` module docs); the gap between "admitted" and "the task is
//! running" is the runtime's own ready queue plus the blocking-pool handoff.
//! That is the *invisible* ceiling behind the unbounded listener, measured directly
//! instead of inferred.
//!
//! ## Minimum, not average
//!
//! Over each [`INTERVAL`] window the controller keeps the **minimum** sojourn
//! per class, never the mean. One item that got through quickly is proof the
//! queue is not standing — it proves a path existed. An average is moved by a
//! single outlier, so an averaging controller shrinks on noise, then grows back
//! when the outlier ages out, then shrinks again: the oscillation CoDel exists
//! to remove.
//!
//! ## Per class, and this is the crux
//!
//! **One queue. One shared limit. Sojourn measured per class.**
//!
//! The trap is head-of-line blocking of the *signal*, not of the queue. Hold
//! one target for everything and a 25-minute `linux.git` clone — which does not
//! care about a 150 ms wait and whose client would never notice one — reads as a
//! catastrophically standing queue against a 5 ms bar, and the controller sheds
//! load that the box could have served. The clone poisons the delay signal for
//! the 10 ms `ls-remote` sharing its queue.
//!
//! So each class carries its own [`OpClass::target`], and the rule that sets it
//! is **half the class's measured median service time** — a client may be made
//! to wait at most long enough to add 50 % to its own latency. Those medians are
//! measured, not picked round: the reference desktop, 2026-08-13, gunnar in an
//! 8 GiB / 8-cpu cgroup.
//!
//! | class | measured median | target | note |
//! |---|---:|---:|---|
//! | ls-remote | 9–12 ms | **5 ms** | the base [`TARGET`] falls out of the measurement |
//! | fetch | 90 ms | 45 ms | `fetch`/`diverge` |
//! | push | 340 ms | 170 ms | `push`/`base` |
//! | clone | 372 ms | 186 ms | `clone`/`large`; xl was 4350 ms |
//! | gc | **NOT MEASURED** | 186 ms | no bakeoff row; borrows clone's, and says so |
//!
//! `ls-remote` is therefore the canary: it trips first, at exactly the 5 ms the
//! design asked for, and the slower classes trip only when the wait is large
//! enough to matter *to them*.
//!
//! ## The control law
//!
//! When a class's window closes, that class's window-minimum decides, and what
//! it moves is the **one shared limit**:
//!
//! * minimum above the class's target for the whole window → the queue is
//! standing → shrink multiplicatively, `limit -= max(limit/8, 1)`;
//! * otherwise → grow additively, `limit += 1`.
//!
//! Clamped to [`LimitBand`]: **[10, 100]** when no ceiling is configured, and
//! **[10, ceiling]** when one is — an operator's explicit number is never
//! lowered to the design's default. The `max(_, 1)` is
//! not cosmetic: without it a limit below 8 has an integer shrink step of zero
//! and the controller can never come down again.
//!
//! It starts at the top of the band. A load shedder has nothing to shed on an
//! idle server, and starting at the floor would make a cold server refuse work
//! it could do for the ~90 intervals (9 s) the additive climb takes.
//!
//! # Gate 2 — memory as a RESERVATION you own, not a measurement you take
//!
//! [`MemoryLedger`] is a budget and a counter. It does not ask the operating
//! system anything, and that is the design rather than a limitation of it:
//!
//! * a free-memory reading is **stale the instant you take it**; a reservation
//! you own is exact;
//! * it denies **before** the allocation, which is the whole principle — asking
//! "is memory low?" after an operation has built a multi-gigabyte buffer is
//! asking after the damage;
//! * it is the fixed-pool shape: admission is "is a slab free?".
//!
//! ## Why this gate is not theoretical for gunnar — measured
//!
//! From the same bakeoff table, peak anonymous RSS on `vs_forge_clone`, on
//! identical wire bytes: **gunnar 2314 MB, gitea 122 MB, Forgejo 167 MB**.
//! Memory is the resource gunnar actually exhausts. The cause is known —
//! `EmitEntry.stored` in `znippy-plugin-git/src/pack_walk.rs:931` is an owned
//! `Vec<u8>` per object, so a clone materialises the whole pack in RAM before
//! writing a byte — and it is being fixed elsewhere. It is not fixed here.
//!
//! ## The honest limitation, stated rather than papered over
//!
//! **A memory budget is only meaningful when an operation's footprint is bounded
//! and predictable, and a clone's is neither today.** It is proportional to the
//! pack, and for `linux.git` that is ~6.4 GB across ~27.6 M allocations. So
//! [`OpClass::estimated_bytes`] is a per-class **constant**, and for
//! [`OpClass::Clone`] that constant is a **guess** — the marginal figure from
//! one corpus, not a bound. Until `stored` becomes an extent into the mapped
//! archive rather than an owned copy, this gate does not bound clone memory
//! correctly, and nothing here should be read as claiming it does. What it does
//! do today is stop *unbounded concurrency* of a heavyweight class, which is a
//! smaller but real property.
//!
//! # OS-less, by law
//!
//! No `/proc`, no `sysinfo`, no `getrusage`, no CPU/disk/memory probing. The
//! only inputs are [`std::time::Instant`] and in-process counters, so this works
//! on bare metal and in a container with equal honesty. The corollary is stated
//! plainly on [`DEFAULT_MEMORY_BUDGET_BYTES`]: this code **cannot** know how much
//! RAM the box has, so the default is a constant an operator on a small machine
//! must lower.
//!
//! No rayon, no thread pool, no allocation on the decision path (the multicore
//! rule: gatling, never rayon): the
//! per-class windows are a fixed `[ClassWindow; OpClass::COUNT]` array behind one
//! mutex touched once per transfer, not once per byte.
use std::sync::atomic::{AtomicBool, AtomicU64, AtomicUsize, Ordering};
use std::sync::Mutex;
use std::time::{Duration, Instant};
use crate::command::GitService;
use gunnar_lfs::Operation;
/// The base sojourn a queue may stand at before it counts as standing: the
/// target for the *fastest* class, [`OpClass::LsRemote`], whose measured median
/// service time is 9–12 ms.
///
/// Every other class scales off its own measurement; see [`OpClass::target`].
pub const TARGET: Duration = Duration::from_millis(5);
/// How long a class's minimum sojourn must stay above its target before the
/// shared limit moves.
///
/// One window, one decision. Shorter and the controller reacts to a single slow
/// scheduling turn; longer and a genuinely standing queue keeps standing while
/// the controller thinks about it.
pub const INTERVAL: Duration = Duration::from_millis(100);
/// The floor the governed limit may never go below.
///
/// A load shedder that can shed itself to zero is a self-inflicted outage. Ten
/// concurrent git operations is a server that is still a server.
pub const MIN_LIMIT: usize = 10;
/// The ceiling the governed limit may never go above **when no budget was
/// declared**. A declared budget is its own ceiling; see
/// [`LimitBand::for_budget`].
pub const MAX_LIMIT: usize = 100;
/// A declared budget's floor is `budget / this`.
///
/// Half. The governor may shed a saturated box down to half of what the box was
/// measured to be able to hold, and no further: the budget's three terms —
/// cores, descriptors, `MemAvailable` — are hard resources, and a controller
/// reading a *latency* signal has no evidence about them. Below half, the
/// honest answer to a client is to serve it.
pub const BUDGET_FLOOR_DENOMINATOR: usize = 2;
/// The default self-imposed memory budget for admitted operations: 8 GiB.
///
/// **This is a number about gunnar, not about the machine.** By the OS-less law
/// this code cannot read the box's RAM, so it cannot size itself; 8 GiB is the
/// cgroup the bakeoff measurement above was taken in
/// (measured 2026-08-13), which makes it the one
/// figure with evidence behind it rather than a round guess.
///
/// An operator on a smaller box **must** lower it, and one on a larger box
/// should raise it. It covers only the footprints of operations admitted through
/// this gate: gunnar's resident baseline (~330 MB, the `ls-remote` rows, which
/// send no pack at all) sits outside the budget and must be left as headroom.
pub const DEFAULT_MEMORY_BUDGET_BYTES: u64 = 8 * 1024 * 1024 * 1024;
const MIB: u64 = 1024 * 1024;
/// How long an operation may **wait** for a slot before it is refused.
///
/// # Why there is a wait at all, when this module argued there must not be
///
/// The original text (still above, and still right about *unbounded* queues)
/// said a git client cannot see a queue, so a fast refusal beats a silent
/// wait. Two of its four premises did not survive measurement:
///
/// * **"Queueing keeps the expensive things alive."** So does refusing. A
/// refused client does not go away: it reconnects, and a reconnect is a
/// *whole* SSH handshake — a new TCP connection, a new key exchange, a new
/// signature verification — where a wait is one parked future. MEASURED on
/// the rig (2026-08-30): at 64 concurrent clients the refusal arm
/// spent more CPU per **served** operation than the waiting arm, because
/// three quarters of the handshakes bought nothing.
/// * **"A queue makes saturation invisible again."** Only an *unmeasured* one
/// does. This queue is bounded, counted
/// ([`crate::AdmissionStats::admitted_after_wait`],
/// [`crate::AdmissionStats::peak_wait_ms`]) and it is the CoDel controller's
/// input, so it is strictly more visible than the refusal it replaces.
///
/// Two seconds because that is well under what a client pays to retry — git
/// itself has no backoff, so the retry is the user's or CI's, in whole seconds
/// — and well under [`crate::SshConfig::exec_grace`] (60 s) and
/// [`crate::SshConfig::inactivity_timeout`] (600 s), so a waiting session can
/// never be disconnected by the timeouts that exist to kill idle ones.
///
/// `Duration::ZERO` restores the 2026-08-05 behaviour exactly: refuse on the
/// spot. That is what `--admission-wait-ms 0` is for, and it is the arm the
/// before/after table was measured against.
pub const DEFAULT_ADMISSION_WAIT: Duration = Duration::from_secs(2);
/// **How long an AUTHENTICATED caller may be queued for a slot**: 30 s.
///
/// The two seconds above are right for the anonymous caller this wait was
/// argued against, and wrong for a key. MEASURED on gunnar.rs, 2026-10-25:
/// every CI lane pushes and fetches as one key, many at once, and git-lfs's
/// pure-SSH `git-lfs-transfer` has no retry — a refused session falls back to
/// `git-lfs-authenticate` and the whole push fails with *"batch request: exec
/// request failed"*. A key's parked sessions are bounded by its own share, so
/// queueing it costs one future per session and saves a failed CI job.
///
/// Thirty, not more: well under [`crate::SshConfig::exec_grace`] (60 s), so a
/// queued session is never cut off by the timeout that kills idle ones.
pub const DEFAULT_KEYED_ADMISSION_WAIT: Duration = Duration::from_secs(30);
/// The share of a transfer's **wire bytes** charged to the memory budget as it
/// moves ([`crate::AdmissionGuard::reserve_more`]).
///
/// A quarter. MEASURED, not guessed: the spool bounds mean the pushed pack
/// itself is spooled past 64 MiB and read back as a read-only mmap (page cache,
/// reclaimable, and `MemAvailable` counts it as free), so what a transfer
/// really holds is the per-object bookkeeping — the walk's entry table
/// (~56 B/object), the closure check's offset set (~16 B), gunnar-wire's
/// visited sets (~100 B) — which came to ~2.5 GB against `linux.git`'s 8.58 GiB
/// pack (the memory limit). 2.5 / 8.58 is 0.29; a quarter
/// is that, rounded to a shift.
///
/// It is a *charge*, not a prediction. Its job is that a transfer which turns
/// out to be enormous stops **new** admissions long before the box is in
/// trouble, without a three-object push-create having to pay for the
/// possibility.
pub const WIRE_MEMORY_NUMERATOR: u64 = 1;
/// The denominator of [`WIRE_MEMORY_NUMERATOR`].
pub const WIRE_MEMORY_DENOMINATOR: u64 = 4;
/// How many bytes must cross the wire between two reservation top-ups.
///
/// The ledger is one atomic compare-and-swap, so this is not about the cost of
/// the call; it is about not turning a per-byte quantity into a per-byte
/// *decision*. 16 MiB is a quarter of the 64 MiB spool threshold, so a
/// transfer that is about to spool has already told the ledger about itself
/// three times.
pub const WIRE_RESERVE_STEP: u64 = 16 * MIB;
/// The three cost shapes a transfer can have, and the thing a lane floor
/// protects.
///
/// See [`OpClass::lane`] for why the split exists at all, and
/// [`Governor::lane_ceiling`] for what a floor actually reserves.
#[derive(Debug, Clone, Copy, PartialEq, Eq, PartialOrd, Ord, Hash)]
pub enum Lane {
/// A ref advertisement: no pack, milliseconds, 4 MiB.
Cheap,
/// A pack **out** of the store — clone and fetch. Mostly page cache.
Read,
/// A pack **in**, and the indexer behind it — push and gc.
Write,
}
impl Lane {
/// Every lane, in index order; the per-lane arrays are sized from this.
pub const ALL: [Lane; 3] = [Lane::Cheap, Lane::Read, Lane::Write];
/// How many lanes there are.
pub const COUNT: usize = Lane::ALL.len();
/// Index into the per-lane arrays.
pub const fn index(self) -> usize {
match self {
Lane::Cheap => 0,
Lane::Read => 1,
Lane::Write => 2,
}
}
/// The name a log or a counter uses.
pub const fn name(self) -> &'static str {
match self {
Lane::Cheap => "cheap",
Lane::Read => "read",
Lane::Write => "write",
}
}
/// The slots of the governed limit this lane is **guaranteed**, as
/// `limit * numerator / denominator`, never below [`Lane::floor_min`].
///
/// An eighth for `cheap` and a quarter for `read`: a `write` storm can take
/// at most five eighths of the box, so an `ls-remote` and a clone always
/// have somewhere to land. `write` reserves nothing, because it is the lane
/// that would starve the others and it is never the one that is starved —
/// whatever the other two are not using is its.
pub const fn floor_share(self) -> (usize, usize) {
match self {
Lane::Cheap => (1, 8),
Lane::Read => (1, 4),
Lane::Write => (0, 1),
}
}
/// The smallest floor a lane keeps however small the limit is.
///
/// Two, not one: one slot means a single stuck operation closes the lane,
/// and a lane floor exists precisely so that a lane cannot be closed by
/// what is happening in another one.
pub const fn floor_min(self) -> usize {
match self {
Lane::Cheap | Lane::Read => 2,
Lane::Write => 0,
}
}
/// This lane's guaranteed slots under a governed limit of `limit`.
pub fn floor(self, limit: usize) -> usize {
let (n, d) = self.floor_share();
(limit * n / d).max(self.floor_min()).min(limit)
}
}
/// Which kind of user-level git operation is being admitted.
///
/// The classes exist for **one** reason: a sojourn measured across a mixed queue
/// is a number about whichever operation happened to be slowest, so the delay
/// signal must be split even though the queue is not.
#[derive(Debug, Clone, Copy, PartialEq, Eq, PartialOrd, Ord, Hash)]
pub enum OpClass {
/// A ref advertisement and nothing else. The canary: fastest, cheapest, and
/// the first to notice a queue that has started to stand.
LsRemote,
/// An incremental fetch — a negotiated pack over an existing history.
Fetch,
/// A full clone. The expensive one, in time and (today, badly) in memory.
Clone,
/// A push, including LFS upload.
Push,
/// Repository maintenance.
Gc,
}
impl OpClass {
/// Every class, in index order. The per-class arrays are sized from this,
/// so a new variant that is not added here fails to compile rather than
/// silently sharing another class's window.
pub const ALL: [OpClass; 5] = [
OpClass::LsRemote,
OpClass::Fetch,
OpClass::Clone,
OpClass::Push,
OpClass::Gc,
];
/// How many classes there are; the width of every per-class array here.
pub const COUNT: usize = OpClass::ALL.len();
/// Index into the per-class arrays.
pub const fn index(self) -> usize {
match self {
OpClass::LsRemote => 0,
OpClass::Fetch => 1,
OpClass::Clone => 2,
OpClass::Push => 3,
OpClass::Gc => 4,
}
}
/// The name a log or a counter uses.
pub const fn name(self) -> &'static str {
match self {
OpClass::LsRemote => "ls-remote",
OpClass::Fetch => "fetch",
OpClass::Clone => "clone",
OpClass::Push => "push",
OpClass::Gc => "gc",
}
}
/// The sojourn **this class** may stand at before the shared limit shrinks.
///
/// Half the class's measured median service time
/// (measured on the reference desktop,
/// 2026-08-13): a client of
/// this class may be made to wait at most long enough to add 50 % to its own
/// latency. That rule, not a table of round numbers, is what stops a slow
/// class from shrinking the limit on a fast class's behalf — and it lands
/// [`OpClass::LsRemote`] exactly on [`TARGET`], because its measured median
/// is 10 ms.
pub const fn target(self) -> Duration {
match self {
// measured 9–12 ms → 5 ms, which IS `TARGET`.
OpClass::LsRemote => TARGET,
// measured 90 ms (`fetch`/`diverge`).
OpClass::Fetch => Duration::from_millis(45),
// measured 372 ms (`clone`/`large`; `xl` was 4350 ms, so this is the
// conservative end of the measured range).
OpClass::Clone => Duration::from_millis(186),
// measured 340 ms (`push`/`base`).
OpClass::Push => Duration::from_millis(170),
// NOT MEASURED: the bakeoff run of 2026-08-13 has no `vs_forge_gc`
// row, so this borrows clone's rather than inventing a figure. Say
// so here rather than let a reader assume it was measured.
OpClass::Gc => Duration::from_millis(186),
}
}
/// What this class reserves **at admission**, before it knows its size.
///
/// # This used to be the whole footprint, and that was the ceiling
///
/// Until 2026-08-30 this returned a per-class **constant equal to the whole
/// operation's measured peak** — 320 MiB for a push, 2 GiB for a clone —
/// and [`MemoryLedger`] refused anything that did not fit in
/// [`DEFAULT_MEMORY_BUDGET_BYTES`]. That is an arithmetic ceiling nobody
/// chose: 8 GiB / 320 MiB = **25 concurrent pushes on any box**, whatever
/// its RAM, and a three-object push-create reserved exactly as much as a
/// `linux.git` mirror.
///
/// MEASURED on the native desktop rig (gunnar-solo pinned to 4 cpus / 8 GiB,
/// 2026-08-30): 500 push-creates at 64 concurrent clients → **118–132 of
/// them refused `MemoryBudget`** while the server's RSS never passed
/// 1.6 GB of its 8 GiB. The gate was refusing three quarters of the work to
/// protect a budget that was 20 % used.
///
/// # So the reservation starts small and GROWS with the bytes on the wire
///
/// This is the *entry* cost — the session, the state machine, the ref
/// advertisement, the first buffers — and it is what admission is decided
/// on. Everything proportional to the transfer is added afterwards by
/// [`crate::AdmissionGuard::reserve_more`], which the transport calls as
/// the pack actually arrives ([`WIRE_MEMORY_NUMERATOR`]). A gate that
/// charges a small push for a large one refuses work the box can do; a
/// gate that never charges the large one takes the box down. Charging what
/// has actually moved is the only rule that is right in both directions.
///
/// The numbers below are marginal RSS per concurrent transfer, measured on
/// the same rig by walking the concurrency up with the gate wide open
/// (`rss-slope.tsv`, a scratch capture outside the repository), rounded up
/// to the next power of
/// two.
pub const fn admission_bytes(self) -> u64 {
match self {
// No pack at all — the ref advertisement and its framing.
OpClass::LsRemote => 4 * MIB,
// A negotiation and the first window of a pack.
OpClass::Fetch => 16 * MIB,
// Same entry cost as a fetch; a clone's real footprint is its pack,
// and its pack is charged as it is written.
OpClass::Clone => 16 * MIB,
// The receive state machine plus the first spool window. MEASURED
// 39 MB marginal per concurrent tiny push-create on the rig.
OpClass::Push => 32 * MIB,
// NOT MEASURED — no bakeoff row, no rig row. Sized at a push's,
// and flagged rather than asserted.
OpClass::Gc => 32 * MIB,
}
}
/// The **lane** this class is metered in.
///
/// One queue is still one queue, but a lane may not be starved by another:
/// an `ls-remote` that costs 4 MiB and 10 ms must not wait behind an 8 GiB
/// mirror push, and on a saturated box it did — every class shared one
/// counter against one limit, so a push storm took every slot.
///
/// Three lanes, because there are three cost shapes: the ref advertisement
/// (no pack), a read (a pack out of the store, mostly page cache), and a
/// write (a pack in, plus the indexer behind it).
pub const fn lane(self) -> Lane {
match self {
OpClass::LsRemote => Lane::Cheap,
OpClass::Fetch | OpClass::Clone => Lane::Read,
OpClass::Push | OpClass::Gc => Lane::Write,
}
}
/// The class an SSH `exec` request belongs to.
///
/// # What SSH cannot tell you, and it matters
///
/// `clone`, `fetch` and `ls-remote` are **the same exec line**:
/// `git-upload-pack '<path>'`. Which of the three it turns out to be is
/// decided later, by the protocol-v2 command the client sends inside the
/// stream (`ls-refs` versus `fetch`), and by whether the client had any
/// history to negotiate from. The admission gate runs before any of that,
/// by design — the entire value of this gate is refusing before the cost is
/// paid.
///
/// So over SSH, `git-upload-pack` is classified [`OpClass::Fetch`]: the
/// middle of the three, chosen deliberately rather than by accident.
/// Classifying it as `Clone` would give every `ls-remote` a 2 GiB
/// reservation and a 186 ms delay target; as `LsRemote` it would give every
/// clone a 4 MiB reservation and a 5 ms target, which is the poisoned signal
/// this module exists to avoid.
///
/// Callers that **do** know — the HTTP smart transport, which sees
/// `GET /info/refs` separately from `POST /git-upload-pack`, and `gunnar gc`,
/// which is not a transport at all — pass the exact class to
/// [`Admission::try_admit`](crate::Admission::try_admit) instead of going
/// through this function.
pub fn for_ssh_service(service: GitService) -> OpClass {
match service {
GitService::UploadPack => OpClass::Fetch,
GitService::ReceivePack => OpClass::Push,
GitService::LfsTransfer(Operation::Upload) => OpClass::Push,
GitService::LfsTransfer(Operation::Download) => OpClass::Fetch,
}
}
}
impl std::fmt::Display for Lane {
fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
f.write_str(self.name())
}
}
impl std::fmt::Display for OpClass {
fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
f.write_str(self.name())
}
}
/// The range the governed limit is clamped to.
///
/// Spelled as a type rather than two constants because it has to be **narrowed
/// by an operator's configured ceiling**: a server told `--max-concurrent-transfers 4`
/// must serve 4, not the design's floor of 10. See [`LimitBand::for_ceiling`].
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub struct LimitBand {
/// The smallest the controller may shrink to.
pub min: usize,
/// The largest it may grow to, and where it starts.
pub max: usize,
}
impl LimitBand {
/// The design's band: **[10, 100]**.
pub const DEFAULT: LimitBand = LimitBand {
min: MIN_LIMIT,
max: MAX_LIMIT,
};
/// The band for a server whose operator pinned a hard ceiling.
///
/// The operator's number is the ceiling, and the FLOOR is clamped under it:
/// pinning 4 gives `[4, 4]`, a governor that cannot move, which is exactly
/// what "I told you 4" should mean. A configured ceiling below
/// [`MIN_LIMIT`] must not be silently RAISED — a gate admitting more than
/// it was told to is the one failure mode a limit may not have.
///
/// **The ceiling is NOT capped at [`MAX_LIMIT`], and that was a bug.** The
/// first form did `MAX_LIMIT.min(ceiling)`, so a server configured with the
/// shipped default of 256 got a band of `[10, 100]` and quietly served 100.
/// Lowering an operator's explicit number is the same defect as raising it,
/// in the other direction: both mean the configured value is not the value.
/// `MAX_LIMIT` is the design's default for a server that has NOT been told a
/// number — it is not an opinion that outranks one that has.
pub fn for_ceiling(ceiling: usize) -> LimitBand {
LimitBand {
min: MIN_LIMIT.min(ceiling),
max: ceiling,
}
}
/// The band for a server whose ceiling is a **declared, measured budget**
/// rather than a round number an operator typed.
///
/// # Why the floor moves with the ceiling — MEASURED
///
/// [`LimitBand::for_ceiling`] pins the floor at [`MIN_LIMIT`] = 10. With the
/// shipped ceiling of 256 that is a band of `[10, 256]`, and the controller
/// walks the whole of it: multiplicative decrease is `limit -= limit/8`, so
/// 256 → 224 → 196 → … → 24 → 21 → **19** → 17 … On the appliance under a
/// 64-client push storm every 100 ms window stands, and the measurement
/// took the gate refusing by name at exactly that — `already serving its
/// limit of 19`, 268 of 1000 push-creates admitted. **19 was not a constant
/// anybody chose; it is a point on a descent.**
///
/// The descent does not stop because shedding does not fix what the
/// controller is measuring. The sojourn it reads was the gap between
/// admission and the spawned task's first line — the tokio ready queue and
/// the blocking-pool handoff — and that gap is set by how many clients are
/// *connected and doing crypto*, not by how many transfers are admitted. A
/// refused client reconnects, so a refusal makes the input worse. A
/// controller with no authority over its own input runs to its floor.
///
/// Both halves are fixed here. The controller's input is now the
/// **admission wait** — the queue this gate actually owns
/// ([`DEFAULT_ADMISSION_WAIT`]) — and its floor is a share of the budget
/// rather than a constant: below [`BUDGET_FLOOR_DENOMINATOR`]`⁻¹` of a
/// number derived from cores, descriptors and `MemAvailable`, the box is
/// provably not the bottleneck and refusing costs more than serving.
pub fn for_budget(budget: usize) -> LimitBand {
LimitBand {
min: (budget / BUDGET_FLOOR_DENOMINATOR).clamp(MIN_LIMIT.min(budget), budget),
max: budget,
}
}
fn clamp(self, limit: usize) -> usize {
limit.clamp(self.min, self.max)
}
}
/// One class's sliding window.
///
/// `opened` is `None` until the first sample, so construction never reads the
/// clock and a test can drive the whole controller off `Instant`s of its own.
#[derive(Debug, Clone, Copy)]
struct ClassWindow {
opened: Option<Instant>,
/// The **minimum** sojourn seen since the window opened. Never a sum, never
/// a count: an average would let one outlier shrink the limit.
min: Option<Duration>,
}
impl ClassWindow {
const EMPTY: ClassWindow = ClassWindow {
opened: None,
min: None,
};
}
/// The CoDel controller: per-class delay windows, one shared limit.
#[derive(Debug)]
pub struct Codel {
band: LimitBand,
/// The one shared limit. Read on every admission, written once per closed
/// window.
limit: AtomicUsize,
/// Fixed-size and preallocated: no map, no allocation, no growth.
windows: Mutex<[ClassWindow; OpClass::COUNT]>,
shrinks: AtomicU64,
grows: AtomicU64,
}
impl Codel {
/// A controller starting at the top of its band.
pub fn new(band: LimitBand) -> Codel {
Codel {
band,
limit: AtomicUsize::new(band.max),
windows: Mutex::new([ClassWindow::EMPTY; OpClass::COUNT]),
shrinks: AtomicU64::new(0),
grows: AtomicU64::new(0),
}
}
/// The band this controller is clamped to.
pub fn band(&self) -> LimitBand {
self.band
}
/// The shared limit right now.
pub fn limit(&self) -> usize {
self.limit.load(Ordering::Relaxed)
}
/// How many times a standing queue has shrunk the limit.
pub fn shrinks(&self) -> u64 {
self.shrinks.load(Ordering::Relaxed)
}
/// How many times a draining queue has grown it.
pub fn grows(&self) -> u64 {
self.grows.load(Ordering::Relaxed)
}
/// Record one item's sojourn, and adjust the shared limit if this closed
/// `class`'s window.
pub fn record(&self, class: OpClass, sojourn: Duration) {
self.record_at(class, sojourn, Instant::now());
}
/// [`Codel::record`] with the clock supplied.
///
/// The seam a test drives: `Instant` cannot be constructed, but it can be
/// offset, so a test captures one `Instant::now()` and hands this window
/// boundaries to the microsecond instead of sleeping for them.
pub fn record_at(&self, class: OpClass, sojourn: Duration, now: Instant) {
let mut windows = self.windows.lock().unwrap_or_else(|e| e.into_inner());
let window = &mut windows[class.index()];
let opened = *window.opened.get_or_insert(now);
window.min = Some(match window.min {
Some(previous) => previous.min(sojourn),
None => sojourn,
});
if now.saturating_duration_since(opened) < INTERVAL {
return;
}
// The window is closed. Its MINIMUM decides, against THIS class's
// target — the head-of-line guard, in one line.
let standing = window.min.is_some_and(|min| min > class.target());
*window = ClassWindow {
opened: Some(now),
min: None,
};
drop(windows);
if standing {
self.shrink();
} else {
self.grow();
}
}
/// Multiplicative decrease: `limit -= max(limit/8, 1)`.
///
/// The `max(_, 1)` is load-bearing. `limit/8` is integer division, so at a
/// limit of 7 the step would be 0 and a standing queue could never bring the
/// limit down — the controller would be dead in exactly the state it exists
/// for.
fn shrink(&self) {
let band = self.band;
let _ = self
.limit
.fetch_update(Ordering::Relaxed, Ordering::Relaxed, |limit| {
Some(band.clamp(limit.saturating_sub((limit / 8).max(1))))
});
self.shrinks.fetch_add(1, Ordering::Relaxed);
}
/// Additive increase: `limit += 1`.
fn grow(&self) {
let band = self.band;
let _ = self
.limit
.fetch_update(Ordering::Relaxed, Ordering::Relaxed, |limit| {
Some(band.clamp(limit.saturating_add(1)))
});
self.grows.fetch_add(1, Ordering::Relaxed);
}
}
/// Gate 2: a budget and a counter. Never a probe.
#[derive(Debug)]
pub struct MemoryLedger {
budget: u64,
reserved: AtomicU64,
}
impl MemoryLedger {
/// A ledger with `budget` bytes to give out.
pub fn new(budget: u64) -> MemoryLedger {
MemoryLedger {
budget,
reserved: AtomicU64::new(0),
}
}
/// The budget.
pub fn budget(&self) -> u64 {
self.budget
}
/// Bytes currently reserved by operations in flight.
pub fn reserved(&self) -> u64 {
self.reserved.load(Ordering::Relaxed)
}
/// Reserve `bytes`, or refuse.
///
/// A compare-and-swap loop rather than "load, check, add": between a load
/// and an add, two concurrent admissions both see room and both take it, and
/// a budget that can be exceeded by the number of racing threads is not a
/// budget. Nothing is allocated here and nothing is allocated by the caller
/// before here — the refusal arrives before the cost.
pub fn try_reserve(&self, bytes: u64) -> bool {
self.reserved
.fetch_update(Ordering::Relaxed, Ordering::Relaxed, |reserved| {
let wanted = reserved.saturating_add(bytes);
(wanted <= self.budget).then_some(wanted)
})
.is_ok()
}
/// Charge `bytes` **whether or not they fit**, and say which it was.
///
/// The reserve path for an operation that is already running: see
/// [`Growth`] for why this cannot refuse. The counter is allowed to exceed
/// the budget, and while it does, [`MemoryLedger::try_reserve`] refuses
/// every new admission — which is the whole mechanism by which one enormous
/// transfer quiets the box instead of taking it down.
pub fn charge(&self, bytes: u64) -> bool {
let before = self.reserved.fetch_add(bytes, Ordering::Relaxed);
before.saturating_add(bytes) <= self.budget
}
/// Give `bytes` back.
///
/// Saturating, because a ledger that can go negative under a double release
/// would hand out the whole budget twice.
pub fn release(&self, bytes: u64) {
let _ = self
.reserved
.fetch_update(Ordering::Relaxed, Ordering::Relaxed, |reserved| {
Some(reserved.saturating_sub(bytes))
});
}
}
/// Why the governor shed an operation.
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub enum Shed {
/// Gate 1: the governed concurrency limit is full.
Concurrency {
/// The limit in force at the moment of the refusal — the *governed*
/// one, which is not the configured one and moves on its own. Reported
/// rather than assumed, or an operator reads a number in a log that
/// matches nothing in a config file.
limit: usize,
/// How many operations were in flight.
in_flight: usize,
},
/// Gate 1b: this operation's **lane** is full, though the server as a
/// whole is not. The other lanes' floors are what is holding the slots; see
/// [`Lane::floor`].
LaneFull {
/// The lane that was full.
lane: Lane,
/// What that lane may hold under the governed limit right now.
ceiling: usize,
/// The governed server-wide limit.
limit: usize,
},
/// Gate 1c: the claim asked for a slot the caller had told this governor to
/// **hold back** ([`Governor::claim_reserving`]), though the server as a
/// whole is not full.
///
/// Its own variant rather than [`Shed::Concurrency`] with a smaller `limit`
/// in it, because reporting a reduced limit as *the* limit is the exact
/// failure the unbounded listener is about: an operator would read `limit: 2` out of a log
/// on a box configured for 256 and conclude the gate was broken. Both
/// numbers travel, and the caller that supplied the reserve is the one that
/// gets to name the rule.
Reserved {
/// Operations in flight when the claim failed.
occupied: usize,
/// The **governed** server-wide limit — the real one, unreduced.
limit: usize,
/// How many slots of it the caller asked to be held back from this
/// claim.
reserve: usize,
},
/// Gate 2: the memory budget is spoken for.
Memory {
/// The class whose estimate did not fit.
class: OpClass,
/// What [`OpClass::estimated_bytes`] asked for.
wanted: u64,
/// What was already reserved.
reserved: u64,
/// The budget.
budget: u64,
},
}
/// What an admitted operation holds: a concurrency slot and a memory
/// reservation, plus the instant it was admitted.
///
/// Deliberately **not** a `Drop` guard. The one guard is
/// [`AdmissionGuard`](crate::AdmissionGuard), which owns this; two `Drop`s over
/// one reservation is how a ledger comes to be released twice.
#[derive(Debug, Clone, Copy)]
pub struct Claim {
/// The class, needed to release the right reservation and to file the
/// sojourn under the right window.
pub class: OpClass,
/// What was reserved **at admission** ([`OpClass::admission_bytes`]).
/// Everything charged afterwards is counted by the guard that owns this
/// claim, so the release is `bytes + grown` and cannot drift.
pub bytes: u64,
/// Enqueue time — when the operation first asked for a slot, before any
/// wait. The sojourn is measured from here.
pub enqueued: Instant,
}
/// What a top-up of a claim's reservation did.
///
/// It is deliberately not a `Result`: an operation that is already running
/// cannot be refused, and pretending otherwise would put a failure on a path
/// with nothing to do about it. What an overdraw does instead is close the door
/// behind it — `reserved > budget` makes [`MemoryLedger::try_reserve`] refuse
/// every *new* admission until this transfer finishes, which is the graceful
/// degradation the budget exists for.
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub enum Growth {
/// The charge fitted inside the budget.
Fitted {
/// What this transfer now holds.
held: u64,
},
/// The charge did not fit and was taken anyway. New admissions are refused
/// until it is given back.
Overdrawn {
/// What this transfer now holds.
held: u64,
/// What the whole server now holds.
reserved: u64,
/// The budget it is over.
budget: u64,
},
}
/// Both gates, and the occupancy they meter.
#[derive(Debug)]
pub struct Governor {
codel: Codel,
memory: MemoryLedger,
/// Operations in flight. This *is* the queue occupancy — incremented inside
/// the same compare-and-swap that checks the limit, so there is no window in
/// which two admissions both see room.
occupied: AtomicUsize,
/// The same count, split by [`Lane`], so a lane floor can be enforced.
/// Fixed-size and preallocated: no map, no allocation on the decision path.
per_lane: [AtomicUsize; Lane::COUNT],
/// How many operations of each lane are **parked** in
/// [`crate::Admission::admit_within`] right now.
///
/// This is the demand signal, and it is the whole reason lane floors are
/// work-conserving rather than a second wall: see
/// [`Governor::lane_ceiling`].
waiting: [AtomicUsize; Lane::COUNT],
shed_concurrency: AtomicU64,
shed_lane: AtomicU64,
shed_memory: AtomicU64,
overdrawn: AtomicU64,
}
impl Governor {
/// A governor over `band` with `budget` bytes of memory to give out.
pub fn new(band: LimitBand, budget: u64) -> Governor {
Governor {
codel: Codel::new(band),
memory: MemoryLedger::new(budget),
occupied: AtomicUsize::new(0),
per_lane: [const { AtomicUsize::new(0) }; Lane::COUNT],
waiting: [const { AtomicUsize::new(0) }; Lane::COUNT],
shed_concurrency: AtomicU64::new(0),
shed_lane: AtomicU64::new(0),
shed_memory: AtomicU64::new(0),
overdrawn: AtomicU64::new(0),
}
}
/// The delay controller.
pub fn codel(&self) -> &Codel {
&self.codel
}
/// The memory ledger.
pub fn memory(&self) -> &MemoryLedger {
&self.memory
}
/// The governed limit right now.
pub fn limit(&self) -> usize {
self.codel.limit()
}
/// Operations in flight right now.
pub fn occupied(&self) -> usize {
self.occupied.load(Ordering::Relaxed)
}
/// Operations in flight in one lane.
pub fn occupied_in(&self, lane: Lane) -> usize {
self.per_lane[lane.index()].load(Ordering::Relaxed)
}
/// **How many slots `lane` may hold right now**: the governed limit, less
/// the part of every *other* lane's floor that somebody is actually
/// **waiting** for.
///
/// # A floor that reserves an idle lane's slots is the old bug again
///
/// The per-principal cap used to bind unconditionally, and measured on the
/// 32-core desktop that meant 32 served and 32 refused against
/// **224 idle slots**
/// (`crate::admission`). A lane floor that reserved an eighth of the box for
/// `ls-remote` whether or not any `ls-remote` existed would be the same
/// mistake with a different noun: it would refuse a push the box could
/// serve, to keep a slot warm for a request nobody had made.
///
/// So the floor is claimed **on demand**, and demand is now observable:
/// since 2026-08-30 an operation over the limit *parks*
/// ([`crate::Admission::admit_within`]) instead of being refused, so the
/// number of parked operations per lane is a live, exact count of who is
/// asking. A lane with nobody waiting reserves nothing; the instant one
/// `ls-remote` parks, one slot of the cheap floor is held back and the next
/// release goes to it rather than to the push storm that would otherwise
/// take every slot for as long as it lasted.
///
/// `min(floor, waiting)` is what makes it a floor and not a queue: a lane
/// can never reserve more than [`Lane::floor`] however many of its
/// operations are waiting, so `cheap` cannot starve `write` either.
pub fn lane_ceiling(&self, lane: Lane, limit: usize) -> usize {
let reserved_elsewhere: usize = Lane::ALL
.iter()
.filter(|other| **other != lane)
.map(|other| other.floor(limit).min(self.waiting_in(*other)))
.sum();
limit.saturating_sub(reserved_elsewhere).max(1)
}
/// Operations of `lane` parked waiting for a slot right now.
pub fn waiting_in(&self, lane: Lane) -> usize {
self.waiting[lane.index()].load(Ordering::Relaxed)
}
/// Say that one operation of `lane` has started waiting.
///
/// Paired with [`Governor::left_wait`] by
/// [`crate::Admission::admit_within`], which is the only caller and which
/// pairs them on every path including cancellation.
pub fn entered_wait(&self, lane: Lane) {
self.waiting[lane.index()].fetch_add(1, Ordering::Relaxed);
}
/// Say that one operation of `lane` has stopped waiting.
pub fn left_wait(&self, lane: Lane) {
let _ = self.waiting[lane.index()].fetch_update(
Ordering::Relaxed,
Ordering::Relaxed,
|waiting| Some(waiting.saturating_sub(1)),
);
}
/// Refusals by gate: (concurrency, lane, memory).
pub fn shed_counts(&self) -> (u64, u64, u64) {
(
self.shed_concurrency.load(Ordering::Relaxed),
self.shed_lane.load(Ordering::Relaxed),
self.shed_memory.load(Ordering::Relaxed),
)
}
/// How many transfers have charged the ledger past its budget while
/// running ([`Growth::Overdrawn`]).
pub fn overdrawn(&self) -> u64 {
self.overdrawn.load(Ordering::Relaxed)
}
/// **The admission decision.** Time first, then the lane, then space, then
/// take all three.
///
/// Nothing is allocated on any refusal path, and nothing the operation
/// needs is allocated on the success path either: a [`Claim`] is three
/// `Copy` words. That is the point of the gate — the refusal has to arrive
/// before the cost, or it is only a slower way of paying it.
pub fn try_claim(&self, class: OpClass) -> Result<Claim, Shed> {
self.claim(class, 0, true)
}
/// [`Governor::try_claim`] or [`Governor::attempt`], chosen by the caller.
pub fn claim_for(&self, class: OpClass, count: bool) -> Result<Claim, Shed> {
self.claim(class, 0, count)
}
/// [`Governor::claim_for`] against a limit `reserve` slots **lower** than
/// the governed one, so that this claim can never take one of the last
/// `reserve` slots.
///
/// # Why the reserve is enforced here and not by the caller
///
/// A caller can read [`Governor::occupied`], compare it to a threshold and
/// only then ask for a claim — and that is a race, for exactly the reason
/// written over the compare-and-swap below: `N` racing admissions all read
/// the same occupancy and all decide they fit. Measured, on a four-slot gate
/// with two hundred anonymous readers: strangers holding *transient* claims
/// while they made that decision filled the server-wide gate, and a keyed
/// caller arriving into that window was refused by gate 1 with
/// `AtCapacity` — the reserve that existed to protect it having never been
/// consulted. A limit that is checked outside the atomic that enforces it is
/// not a limit.
///
/// So the reserve rides in the same `fetch_update`. `N` racing strangers can
/// then no more collectively exceed `limit - reserve` than they can exceed
/// `limit`, and a refused one never holds a slot at all — which matters
/// twice, because the refusal path of a flood is the path an attacker is
/// trying to make expensive.
///
/// # Why this moves no shed counter
///
/// The three [`Governor::shed_counts`] counters count *this* governor's own
/// gates — the limit it controls, the lane floors it computes, the ledger it
/// keeps. The reserve is none of those: it is a number the caller handed in,
/// for a policy the caller owns, and the caller counts it
/// (`AdmissionStats::refused_keyed_reserve`). Counting it here as a
/// concurrency shed would tell an operator the CoDel limit was full when it
/// was nowhere near, and counting it here under a fourth name would give the
/// same event two owners.
pub fn claim_reserving(
&self,
class: OpClass,
reserve: usize,
count: bool,
) -> Result<Claim, Shed> {
self.claim(class, reserve, count)
}
/// [`Governor::try_claim`] without moving the shed counters.
///
/// The retry inside [`crate::Admission::admit_within`] asks this door many
/// times for **one** operation. Counting every ask would make
/// `shed_concurrency` a count of *polls* rather than of refusals — the
/// counter would read 21 where one client was refused once, which is worse
/// than no counter. So the wait loop attempts without counting and counts
/// once, at the end, when the refusal is final.
pub fn attempt(&self, class: OpClass) -> Result<Claim, Shed> {
self.claim(class, 0, false)
}
fn claim(&self, class: OpClass, reserve: usize, count: bool) -> Result<Claim, Shed> {
let limit = self.codel.limit();
// What THIS claim may take. `reserve` is 0 for every ordinary caller,
// in which case the two are the same number and nothing below changes.
// See [`Governor::claim_reserving`] for why the reserve is applied here
// rather than checked by the caller beforehand.
let available = limit.saturating_sub(reserve);
// Check and take in ONE compare-and-swap. "Load, compare, then add"
// lets N racing admissions all read `occupied == limit - 1`.
let taken = self
.occupied
.fetch_update(Ordering::Relaxed, Ordering::Relaxed, |occupied| {
(occupied < available).then_some(occupied + 1)
});
if let Err(occupied) = taken {
if reserve > 0 {
// The caller's rule refused this, not the governor's, so the
// governor's counters do not move and the governor's variant is
// not used — even when the box happens to ALSO be full. Reported
// this way round on purpose: a caller that supplied a reserve
// has a policy reason for it, and a refusal that flipped between
// two reasons depending on how full the box happened to be would
// be unreadable in a log and, in `Admission`'s case, would decide
// whether the caller gets parked.
return Err(Shed::Reserved {
occupied,
limit,
reserve,
});
}
if count {
self.shed_concurrency.fetch_add(1, Ordering::Relaxed);
}
return Err(Shed::Concurrency {
limit,
in_flight: occupied,
});
}
// The lane, second: the server-wide limit is the resource that actually
// runs out, and a lane floor is only interesting once it does.
let lane = class.lane();
let ceiling = self.lane_ceiling(lane, limit);
let lane_taken = self.per_lane[lane.index()].fetch_update(
Ordering::Relaxed,
Ordering::Relaxed,
|held| (held < ceiling).then_some(held + 1),
);
if lane_taken.is_err() {
self.occupied.fetch_sub(1, Ordering::Relaxed);
if count {
self.shed_lane.fetch_add(1, Ordering::Relaxed);
}
return Err(Shed::LaneFull {
lane,
ceiling,
limit,
});
}
let bytes = class.admission_bytes();
if !self.memory.try_reserve(bytes) {
// Hand the concurrency slot AND the lane slot straight back: a slot
// leaked on the memory path is a server that refuses everything an
// hour after start-up.
self.per_lane[lane.index()].fetch_sub(1, Ordering::Relaxed);
self.occupied.fetch_sub(1, Ordering::Relaxed);
if count {
self.shed_memory.fetch_add(1, Ordering::Relaxed);
}
return Err(Shed::Memory {
class,
wanted: bytes,
reserved: self.memory.reserved(),
budget: self.memory.budget(),
});
}
Ok(Claim {
class,
bytes,
enqueued: Instant::now(),
})
}
/// **Charge a running transfer for what has actually crossed the wire.**
///
/// Called by [`crate::AdmissionGuard::reserve_more`] and by nothing else,
/// so the guard stays the single owner of "what this operation holds" and
/// the release cannot drift. See [`Growth`] for why it cannot refuse and
/// [`WIRE_MEMORY_NUMERATOR`] for what fraction of the wire is charged.
pub fn charge(&self, held_after: u64, extra: u64) -> Growth {
if self.memory.charge(extra) {
Growth::Fitted { held: held_after }
} else {
self.overdrawn.fetch_add(1, Ordering::Relaxed);
Growth::Overdrawn {
held: held_after,
reserved: self.memory.reserved(),
budget: self.memory.budget(),
}
}
}
/// Give a claim's slot and reservation back. Called from exactly one place:
/// [`AdmissionGuard`](crate::AdmissionGuard)'s `Drop`, which passes what it
/// grew the reservation to on top of [`Claim::bytes`].
pub fn release(&self, claim: &Claim, grown: u64) {
self.memory.release(claim.bytes.saturating_add(grown));
let lane = claim.class.lane();
let _ = self.per_lane[lane.index()].fetch_update(
Ordering::Relaxed,
Ordering::Relaxed,
|held| Some(held.saturating_sub(1)),
);
let _ = self
.occupied
.fetch_update(Ordering::Relaxed, Ordering::Relaxed, |occupied| {
Some(occupied.saturating_sub(1))
});
}
/// File one **admission wait** against its class's window.
///
/// This is the controller's input, and the change of input is the point:
/// see [`LimitBand::for_budget`] for what it used to be (the runtime's
/// spawn-to-run gap) and why a controller that cannot move its own input
/// runs to its floor. A wait of zero — the ordinary case on an unloaded
/// server — is a sample too, and it is the one that grows the limit back.
pub fn record_sojourn(&self, class: OpClass, sojourn: Duration) {
self.codel.record(class, sojourn);
}
}
/// A one-shot latch for "this operation's work has actually started".
///
/// Its own type so the "was the sojourn ever measured?" question has one
/// answer-holder rather than a bool copied into two structs.
#[derive(Debug, Default)]
pub struct ServiceMark(AtomicBool);
impl ServiceMark {
/// Claim the mark. `true` the first time, `false` for every later call, so a
/// caller that marks twice files one sample rather than two.
pub fn claim(&self) -> bool {
!self.0.swap(true, Ordering::Relaxed)
}
/// Whether the mark was ever claimed.
pub fn claimed(&self) -> bool {
self.0.load(Ordering::Relaxed)
}
}
#[cfg(test)]
mod tests {
use super::*;
/// Drive `class` through exactly one closed window at `sojourn`.
///
/// Two samples: one to open the window, one at `INTERVAL` later to close it.
/// Returns the instant the window closed, so a caller can chain windows.
fn one_window(codel: &Codel, class: OpClass, sojourn: Duration, from: Instant) -> Instant {
codel.record_at(class, sojourn, from);
let closed = from + INTERVAL;
codel.record_at(class, sojourn, closed);
closed
}
/// **A standing queue shrinks the limit.**
///
/// # Seen red
///
/// By changing `Codel::record_at`'s decision to
/// `let standing = window.min.is_some_and(|min| min > INTERVAL);` — i.e.
/// comparing the window minimum against the window length instead of the
/// class's target, the natural "off by one constant" of this controller.
/// Failure:
///
/// ```text
/// a queue standing at 50ms against a 5ms target did not shrink the limit:
/// it is still 100 after 8 closed windows (shrinks=0)
/// ```
#[test]
fn a_standing_queue_shrinks_the_limit() {
let codel = Codel::new(LimitBand::DEFAULT);
assert_eq!(
codel.limit(),
MAX_LIMIT,
"a governor starts at the top of its band"
);
// Ten times the ls-remote target, for eight windows.
let mut at = Instant::now();
for _ in 0..8 {
at = one_window(&codel, OpClass::LsRemote, Duration::from_millis(50), at);
}
assert!(
codel.limit() < MAX_LIMIT,
"a queue standing at 50ms against a {:?} target did not shrink the limit: it is \
still {} after 8 closed windows (shrinks={})",
OpClass::LsRemote.target(),
codel.limit(),
codel.shrinks()
);
assert_eq!(
codel.shrinks(),
8,
"every closed standing window must shrink"
);
// 100 → 88 → 78 → 69 → 61 → 54 → 48 → 42 → 37: multiplicative, not -8.
assert_eq!(
codel.limit(),
37,
"the decrease must be MULTIPLICATIVE (limit -= limit/8); an additive one would \
leave a different number here and would not back off fast enough to matter"
);
assert_eq!(codel.grows(), 0);
}
/// **A draining queue grows the limit.**
///
/// # Seen red
///
/// By making `Codel::grow` a no-op (`fn grow(&self) { }`). Failure:
///
/// ```text
/// a queue draining at 1ms did not grow the limit: 10 after 20 closed
/// windows, and a controller that only shrinks is a ratchet
/// ```
#[test]
fn a_draining_queue_grows_the_limit() {
let codel = Codel::new(LimitBand::DEFAULT);
// Start from the floor so growth is visible: shrink it there first.
let mut at = Instant::now();
for _ in 0..40 {
at = one_window(&codel, OpClass::LsRemote, Duration::from_millis(50), at);
}
assert_eq!(
codel.limit(),
MIN_LIMIT,
"the standing phase must reach the floor"
);
for _ in 0..20 {
at = one_window(&codel, OpClass::LsRemote, Duration::from_millis(1), at);
}
assert!(
codel.limit() > MIN_LIMIT,
"a queue draining at 1ms did not grow the limit: {} after 20 closed windows, and a \
controller that only shrinks is a ratchet",
codel.limit()
);
assert_eq!(
codel.limit(),
MIN_LIMIT + 20,
"growth must be ADDITIVE, one per closed draining window"
);
}
/// **The limit never leaves its band**, in either direction, however long
/// the load runs.
///
/// # Seen red
///
/// By dropping the clamp from `Codel::shrink`
/// (`Some(limit.saturating_sub((limit / 8).max(1)))`). Failure:
///
/// ```text
/// the limit left the band: 9 is outside [10, 100]
/// ```
///
/// It fails on the FIRST step past the floor, not after drifting to zero,
/// which is what makes the assertion inside the loop worth having: a check
/// only at the end would report `0` and hide when it started.
#[test]
fn the_limit_never_leaves_the_band() {
let codel = Codel::new(LimitBand::DEFAULT);
let mut at = Instant::now();
// Far past the floor.
for _ in 0..200 {
at = one_window(&codel, OpClass::LsRemote, Duration::from_secs(1), at);
let limit = codel.limit();
assert!(
(MIN_LIMIT..=MAX_LIMIT).contains(&limit),
"the limit left the band: {limit} is outside [{MIN_LIMIT}, {MAX_LIMIT}]"
);
}
assert_eq!(codel.limit(), MIN_LIMIT);
// …and far past the ceiling.
for _ in 0..500 {
at = one_window(&codel, OpClass::LsRemote, Duration::from_micros(1), at);
let limit = codel.limit();
assert!(
(MIN_LIMIT..=MAX_LIMIT).contains(&limit),
"the limit left the band: {limit} is outside [{MIN_LIMIT}, {MAX_LIMIT}]"
);
}
assert_eq!(codel.limit(), MAX_LIMIT);
}
/// **THE HEAD-OF-LINE TEST.** A slow class does not shrink the limit on a
/// fast class's behalf.
///
/// A `linux.git` clone waiting 150 ms is a clone that has not noticed. An
/// `ls-remote` waiting 150 ms has had its latency multiplied by fifteen.
/// One number cannot say both, so the target is per class — and this test is
/// the difference between the two designs.
///
/// # Seen red
///
/// By replacing the per-class bar in `Codel::record_at` with the single
/// shared one the naive design has:
/// `let standing = window.min.is_some_and(|min| min > TARGET);`
/// Failure:
///
/// ```text
/// a clone waiting 150ms — well inside its own 186ms target, and a wait its
/// client cannot perceive — shrank the shared limit to 42. That is the
/// slow class poisoning the fast class's delay signal, which is the exact
/// trap per-class sojourn exists to close (shrinks=8)
/// ```
#[test]
fn a_slow_class_does_not_shrink_the_limit_on_a_fast_class_behalf() {
let codel = Codel::new(LimitBand::DEFAULT);
let mut at = Instant::now();
// Eight windows of clones waiting 150 ms — thirty times the ls-remote
// target, and comfortably INSIDE clone's own 186 ms target.
assert!(Duration::from_millis(150) > OpClass::LsRemote.target());
assert!(Duration::from_millis(150) < OpClass::Clone.target());
for _ in 0..8 {
at = one_window(&codel, OpClass::Clone, Duration::from_millis(150), at);
}
assert_eq!(
codel.limit(),
MAX_LIMIT,
"a clone waiting 150ms — well inside its own {:?} target, and a wait its client \
cannot perceive — shrank the shared limit to {}. That is the slow class poisoning \
the fast class's delay signal, which is the exact trap per-class sojourn exists to \
close (shrinks={})",
OpClass::Clone.target(),
codel.limit(),
codel.shrinks()
);
assert_eq!(codel.shrinks(), 0);
// The other half of the bar, or the test above would also pass on a
// controller that had simply stopped shrinking: the SAME 150 ms, filed
// under ls-remote, must shrink hard.
for _ in 0..8 {
at = one_window(&codel, OpClass::LsRemote, Duration::from_millis(150), at);
}
assert!(
codel.limit() < MAX_LIMIT,
"150ms of ls-remote sojourn is thirty times its {:?} target and MUST shed load; a \
controller that shrinks for nobody is not a controller",
OpClass::LsRemote.target()
);
assert_eq!(codel.shrinks(), 8);
// …and the clone class is still healthy while the shared limit came
// down. One queue, one limit, two signals.
let before = codel.limit();
let shrinks_before = codel.shrinks();
for _ in 0..4 {
at = one_window(&codel, OpClass::Clone, Duration::from_millis(150), at);
}
assert!(
codel.limit() > before,
"the clone class must still read as DRAINING and grow the shared limit back from \
{before}, even though its absolute sojourn is thirty times the ls-remote target; \
it is {} instead",
codel.limit()
);
assert_eq!(
codel.shrinks(),
shrinks_before,
"not one of those clone windows may be read as standing"
);
}
/// Each class keeps its **own** window: a sample filed under one class may
/// not open, close or lower another's.
///
/// # Seen red
///
/// By indexing every class into slot 0 in `Codel::record_at`
/// (`&mut windows[0]`). Failure:
///
/// ```text
/// assertion `left == right` failed: push's own windows must close and grow
/// left: 10
/// right: 9
/// ```
///
/// The tenth growth is the ls-remote sample's own window being closed by a
/// push, which is exactly the crossing this test forbids — one class's
/// sample opening and closing another's window.
#[test]
fn a_sample_lands_in_its_own_class_window_only() {
let codel = Codel::new(LimitBand::DEFAULT);
let at = Instant::now();
// Open ls-remote's window, then push a full INTERVAL of PUSH samples
// through. Nothing may close ls-remote's window but an ls-remote sample.
codel.record_at(OpClass::LsRemote, Duration::from_millis(50), at);
for step in 0..10 {
codel.record_at(
OpClass::Push,
Duration::from_millis(1),
at + INTERVAL * (step + 1),
);
}
// Push closed its own windows and they were healthy, so it GREW.
assert_eq!(codel.grows(), 9, "push's own windows must close and grow");
assert_eq!(
codel.shrinks(),
0,
"a push sample closed the ls-remote window: shrinks={} grows={} after one ls-remote \
sample that could not have closed anything",
codel.shrinks(),
codel.grows()
);
}
/// **Reserving up to the budget admits; the next operation is denied.**
///
/// # Seen red
///
/// By making `MemoryLedger::try_reserve` compare `reserved` rather than
/// `reserved + bytes` against the budget — the classic off-by-one-operation
/// budget check, which lets the last admission overshoot by a whole
/// estimate. Failure:
///
/// ```text
/// the ledger admitted 5 clones against a 4-clone budget: 10737418240
/// reserved of 8589934592
/// ```
#[test]
fn the_budget_admits_up_to_itself_and_then_denies() {
let clone = OpClass::Clone.admission_bytes();
let ledger = MemoryLedger::new(clone * 4);
let mut admitted = 0;
while ledger.try_reserve(clone) {
admitted += 1;
assert!(
admitted <= 4,
"the ledger admitted {admitted} clones against a 4-clone budget: {} reserved of {}",
ledger.reserved(),
ledger.budget()
);
}
assert_eq!(admitted, 4, "a 4-clone budget must admit exactly 4 clones");
assert_eq!(ledger.reserved(), ledger.budget());
// …and a SMALLER operation still fits nowhere, which is the property a
// "there is some room" check would lose.
assert!(
!ledger.try_reserve(1),
"a full ledger handed out one more byte"
);
}
/// **A completed operation releases its reservation and the next one is
/// then admitted.**
///
/// # Seen red
///
/// By making `MemoryLedger::release` a no-op. Failure:
///
/// ```text
/// the 4th clone finished and gave its 2147483648 bytes back, but the next
/// clone was still refused: 8589934592 reserved of 8589934592. A budget
/// that is never released is a server that wedges after N operations
/// ```
#[test]
fn a_completed_operation_releases_its_reservation() {
let clone = OpClass::Clone.admission_bytes();
let ledger = MemoryLedger::new(clone * 4);
for _ in 0..4 {
assert!(ledger.try_reserve(clone));
}
assert!(!ledger.try_reserve(clone), "the budget is full");
ledger.release(clone);
assert!(
ledger.try_reserve(clone),
"the 4th clone finished and gave its {clone} bytes back, but the next clone was \
still refused: {} reserved of {}. A budget that is never released is a server that \
wedges after N operations",
ledger.reserved(),
ledger.budget()
);
}
/// **The budget is never exceeded, however hard it is raced.**
///
/// # This test was HOLLOW twice, and both failures are worth keeping
///
/// 1. First written as 64 `thread::spawn`s in a loop, it **passed against a
/// deliberately broken `try_reserve`**: spawning a thread costs orders of
/// magnitude more than the three instructions being raced, so the racers
/// arrived one at a time. A concurrency test whose threads never overlap
/// is a test of `thread::spawn`.
/// 2. Adding a `Barrier` was not enough either — five runs, still green.
/// One release of 64 threads gives the bug exactly one chance to happen
/// inside a window a few nanoseconds wide.
///
/// What makes it fail is **many chances at the boundary**: a budget of four
/// slabs, every thread cycling reserve/release, so the "is there room for
/// one more" edge is crossed millions of times instead of once. The
/// invariant is checked by the racers themselves, not after the fact.
///
/// # Seen red
///
/// By rewriting `try_reserve` as load / compare / `fetch_add`, which is what
/// it looks like it could be. On the reference desktop, 2026-08-14:
///
/// ```text
/// 32 racers cycling 20000 times each drove the ledger to 23068672 bytes
/// against a budget of 4194304: it was exceeded by 18874368 bytes, which
/// is a budget that the number of concurrent admissions can walk straight
/// past
/// ```
///
/// Five and a half times the budget. The naive version does not overshoot
/// by one slab, it overshoots by however many threads are in the window.
#[test]
fn the_budget_is_not_exceeded_by_racing_admissions() {
use std::sync::{Arc, Barrier};
let slab = 1024 * 1024u64;
let slabs = 4u64;
let racers = 32;
let rounds = 20_000;
let budget = slab * slabs;
let ledger = Arc::new(MemoryLedger::new(budget));
// The high-water mark any racer ever SAW, which is the applied output:
// a budget is not a budget if a reader can observe it exceeded.
let peak = Arc::new(AtomicU64::new(0));
let gate = Arc::new(Barrier::new(racers));
// Plain std threads: these are one-shot racers, not data-parallel work,
// so there is nothing for gatling to rotate over and rayon is forbidden
// outright (the multicore rule: gatling, never rayon).
let threads: Vec<_> = (0..racers)
.map(|_| {
let ledger = Arc::clone(&ledger);
let peak = Arc::clone(&peak);
let gate = Arc::clone(&gate);
std::thread::spawn(move || {
gate.wait();
for _ in 0..rounds {
if ledger.try_reserve(slab) {
peak.fetch_max(ledger.reserved(), Ordering::Relaxed);
ledger.release(slab);
}
}
})
})
.collect();
for thread in threads {
thread.join().expect("racer");
}
assert_eq!(
ledger.reserved(),
0,
"every reservation was released, so the ledger must be empty"
);
assert!(
peak.load(Ordering::Relaxed) <= budget,
"{} racers cycling {rounds} times each drove the ledger to {} bytes against a budget \
of {budget}: it was exceeded by {} bytes, which is a budget that the number of \
concurrent admissions can walk straight past",
racers,
peak.load(Ordering::Relaxed),
peak.load(Ordering::Relaxed) - budget
);
// …and the racers really did reach the boundary, or the assertion above
// passed on a ledger nobody ever filled.
assert_eq!(
peak.load(Ordering::Relaxed),
budget,
"the ledger never actually filled ({} of {budget}), so the boundary this test is \
about was never crossed",
peak.load(Ordering::Relaxed)
);
}
/// A memory refusal must give the **concurrency** slot straight back.
///
/// # Seen red
///
/// By deleting the `self.occupied.fetch_sub(1, ...)` from the memory arm of
/// `Governor::try_claim`. Failure:
///
/// ```text
/// 40 memory refusals leaked 40 concurrency slots: occupied is 44 with 4
/// operations actually in flight, so the gate now refuses on TIME for a
/// reason that has nothing to do with time
/// ```
#[test]
fn a_memory_refusal_does_not_leak_a_concurrency_slot() {
let clone = OpClass::Clone.admission_bytes();
let governor = Governor::new(LimitBand::DEFAULT, clone * 4);
let held: Vec<_> = (0..4)
.map(|i| {
governor
.try_claim(OpClass::Clone)
.unwrap_or_else(|shed| panic!("clone {i} of 4 was shed: {shed:?}"))
})
.collect();
for _ in 0..40 {
let shed = governor
.try_claim(OpClass::Clone)
.expect_err("the budget is full");
assert!(matches!(shed, Shed::Memory { .. }), "{shed:?}");
}
assert_eq!(
governor.occupied(),
4,
"40 memory refusals leaked {} concurrency slots: occupied is {} with 4 operations \
actually in flight, so the gate now refuses on TIME for a reason that has nothing \
to do with time",
governor.occupied() - 4,
governor.occupied()
);
for claim in &held {
governor.release(claim, 0);
}
assert_eq!(governor.occupied(), 0);
assert_eq!(governor.memory().reserved(), 0);
}
/// The operator's ceiling is the upper bound of the whole band, floor
/// included.
///
/// # Seen red
///
/// By writing `LimitBand::for_ceiling` as `{ min: MIN_LIMIT, max: MAX_LIMIT.min(ceiling) }`
/// — the obvious version. Failure:
///
/// ```text
/// assertion `left == right` failed: a server told to serve 4 has a band of
/// [10, 4] and a limit of 4: the governor admits more than the operator
/// configured
/// left: LimitBand { min: 10, max: 4 }
/// right: LimitBand { min: 4, max: 4 }
/// ```
///
/// An inverted band — `min` above `max` — which `usize::clamp` panics on
/// the moment the controller tries to move. The band has to be well formed
/// at construction, not corrected later.
#[test]
fn a_configured_ceiling_below_the_floor_is_still_the_ceiling() {
let band = LimitBand::for_ceiling(4);
assert_eq!(
band,
LimitBand { min: 4, max: 4 },
"a server told to serve 4 has a band of [{}, {}] and a limit of {}: the governor \
admits more than the operator configured",
band.min,
band.max,
Codel::new(band).limit()
);
assert_eq!(Codel::new(band).limit(), 4);
// ★ A ceiling ABOVE the design default is honoured, not capped.
//
// The first form did `MAX_LIMIT.min(ceiling)`, so the shipped default of
// 256 became a band of [10, 100] and the server quietly served 100.
// Lowering an operator's explicit number is the same defect as raising
// it. Seen RED by restoring that `.min()`: left `LimitBand { min: 10,
// max: 100 }`, right `LimitBand { min: 10, max: 256 }`.
assert_eq!(
LimitBand::for_ceiling(256),
LimitBand { min: 10, max: 256 },
"a configured ceiling of 256 was lowered to the design's default"
);
assert_eq!(LimitBand::for_ceiling(32), LimitBand { min: 10, max: 32 });
// With no ceiling configured, the design's band still applies.
assert_eq!(LimitBand::DEFAULT, LimitBand { min: 10, max: 100 });
}
/// Every class has a distinct index, or two classes share a window and the
/// head-of-line guard is silently off.
#[test]
fn every_class_has_its_own_window_slot() {
let mut seen = [false; OpClass::COUNT];
for class in OpClass::ALL {
assert!(
!std::mem::replace(&mut seen[class.index()], true),
"{class} shares window slot {} with another class",
class.index()
);
}
assert!(seen.iter().all(|&s| s));
}
/// The service mark is one-shot, so a caller that marks twice files one
/// sojourn sample rather than two.
#[test]
fn the_service_mark_is_one_shot() {
let mark = ServiceMark::default();
assert!(!mark.claimed());
assert!(mark.claim(), "the first claim wins");
assert!(!mark.claim(), "the second claim must not");
assert!(mark.claimed());
}
}