crates/admit/src/lib.rs
//! **Admission: what this work would cost, whether it gets in, and what it
//! actually cost** — one trait for every product that serves bytes to a
//! consumer it has to share a box with.
//!
//! ```text
//! plan(work, who, usage) -> Plan pure: the estimate, the class, a sentence
//! admit(&plan) -> Guard never blocks: a guard, or a Refusal with a reason
//! | Refusal
//! record(guard, &actual) feeds the usage ledger and the per-kind estimate
//! ```
//!
//! # Why three calls and not one
//!
//! `plan` has **no side effect**, so an admin console and a customer can ask
//! "what would a clone of this cost me, and would I get it at full speed?"
//! without taking anything. `admit` is the only call that takes a slot, and it
//! is handed the plan rather than recomputing one, so the sentence a refusal
//! quotes is the one the caller already saw. `record` is the only writer of
//! what was learnt: usage for the budget, and an EWMA per [`WorkKind`] for the
//! next estimate.
//!
//! # Never a waiting queue for interactive git
//!
//! A class maps to a **share of concurrent slots** plus a **fast refusal with a
//! reason**; `admit` never parks the caller. The reasoning is gunnar's
//! (`gunnar-ssh/src/admission.rs`, "Why this refuses instead of queueing"): a
//! git client retries but cannot see a queue, a queued transfer keeps its
//! socket and session alive (the very resource a flood exhausts), the
//! inactivity timeouts fight a queue, and a queue makes saturation invisible.
//! [`Class::Queued`] exists as a CLASS so a product that genuinely owns a queue
//! for non-interactive work (holger's pull-through) can say so in the plan;
//! nothing in this crate queues.
//!
//! # The two implementations
//!
//! * [`PassThrough`] — always [`Class::Full`], never refuses; `record` only
//! feeds the estimate. What a product runs when the operator wrote
//! `limit = "none"`, and what it is when built without its share feature.
//! * [`UsageShare`] — rolling-window bytes per consumer (uploads + downloads +
//! upstream fetch). Below the limit → `Full`; above → `Reduced`, which is
//! the anonymous share. Anonymous → `Floor`, with a per-day-token byte
//! budget. Defaults: [`DEFAULT_LIMIT_BYTES`] per [`DEFAULT_WINDOW_DAYS`]
//! days, [`DEFAULT_ANONYMOUS_PER_DAY`] per anonymous token per day.
//!
//! # Anonymous is a DAY TOKEN, never an address
//!
//! [`visitors`] is holger's privacy construction moved here AS IS: a client is
//! reduced to `blake3::keyed_hash(key, address)[..16]` under a key drawn from
//! the OS RNG at the start of each UTC day and overwritten at its end. Within
//! a day one client is one [`DayToken`] (so a per-day budget is right);
//! across days nobody — this process included — can link two tokens or test
//! one against a candidate address, because the key that would make that
//! possible no longer exists. [`DayKey`] is the same construction with the
//! key in atomics under a sequence lock, for the admission path: one token per
//! CONNECTION, no mutex.
//!
//! # Multi-core: no global lock on the admission path
//!
//! `plan` + `admit` run ONCE per transfer and `record` once when its guard is
//! done — nothing per byte, nothing in a streaming loop. On that path:
//!
//! * usage is a [`Table`] — sharded per core, preallocated, one `AtomicU64`
//! word per consumer-day; a 30-day sum is 30 loads, a charge one CAS; only a
//! consumer's FIRST sight takes its shard's insert lock (and evicts idle
//! consumers there, off the hot path);
//! * the estimate is an [`Estimator`] striped per core, `f64` bits in atomics;
//! * a class slot is one `fetch_add` + compare, given back with `fetch_sub`;
//! * durability is not here: a product with a durable ledger (gunnar's flow
//! ledger) hands a [`Seed`] and the table is its cache.
use std::fmt;
pub mod daykey;
pub mod estimate;
mod share;
pub mod table;
pub mod visitors;
pub use daykey::DayKey;
pub use estimate::{EstimateRow, Estimator};
pub use share::{parse_bytes, Seed, Settings, SettingsError, Slots, UsageShare, DEFAULT_MAX_CONSUMERS};
pub use table::Table;
/// Seconds in one UTC day — the day token's window and the usage window's bucket.
pub const DAY_SECS: u64 = visitors::WINDOW_SECS;
/// **Rolling-window byte budget for a named consumer**: 100 GB (SI) per window.
pub const DEFAULT_LIMIT_BYTES: u64 = 100_000_000_000;
/// **The rolling window, in UTC days.**
pub const DEFAULT_WINDOW_DAYS: u32 = 30;
/// **Bytes one anonymous day token may move per UTC day**: 5 GB (SI).
pub const DEFAULT_ANONYMOUS_PER_DAY: u64 = 5_000_000_000;
/// **What kind of work** is asking. The estimate is learnt per kind, because a
/// push and a clone of the same repository cost different things in different
/// directions.
#[derive(Clone, Copy, Debug, PartialEq, Eq, Hash, PartialOrd, Ord)]
pub enum WorkKind {
/// A full clone (upload-pack with no haves).
Clone,
/// An incremental fetch.
Fetch,
/// A push (receive-pack).
Push,
/// An artifact download (a registry blob, a crate, an LFS object).
ArtifactDownload,
/// **Pull-through**: fetching from an upstream on a consumer's behalf.
PullThrough,
/// Mirroring one asset from an upstream.
MirrorAsset,
/// Promoting an artifact between repositories/stages.
Promote,
/// Anything else a product meters.
Other,
}
impl WorkKind {
/// Every kind, in a fixed order (the estimator's table index).
pub const ALL: [WorkKind; 8] = [
WorkKind::Clone,
WorkKind::Fetch,
WorkKind::Push,
WorkKind::ArtifactDownload,
WorkKind::PullThrough,
WorkKind::MirrorAsset,
WorkKind::Promote,
WorkKind::Other,
];
/// The word the plan's sentence uses.
pub fn name(self) -> &'static str {
match self {
WorkKind::Clone => "clone",
WorkKind::Fetch => "fetch",
WorkKind::Push => "push",
WorkKind::ArtifactDownload => "download",
WorkKind::PullThrough => "pull-through",
WorkKind::MirrorAsset => "mirror",
WorkKind::Promote => "promote",
WorkKind::Other => "request",
}
}
pub(crate) fn index(self) -> usize {
self as usize
}
}
impl fmt::Display for WorkKind {
fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
f.write_str(self.name())
}
}
/// **The work being asked for**, with the size hint the caller has (a
/// repository's packed size, a blob's `Content-Length`, a manifest's layer
/// sum). `None` when the caller knows nothing — the estimate then falls back
/// to what this kind has cost on average.
#[derive(Clone, Copy, Debug, PartialEq, Eq)]
pub struct Work {
pub kind: WorkKind,
pub size_hint: Option<u64>,
}
impl Work {
pub fn new(kind: WorkKind, size_hint: Option<u64>) -> Work {
Work { kind, size_hint }
}
}
/// **An anonymous consumer for one UTC day** — see [`visitors`]. Made by
/// [`DayKey::token`], so it is only ever the keyed hash of an address, never
/// the address.
#[derive(Clone, Copy, Debug, PartialEq, Eq, Hash)]
pub struct DayToken {
/// The UTC day (`unix / DAY_SECS`) whose key minted this token.
pub day: u64,
pub(crate) token: [u8; 16],
}
impl DayToken {
/// The first four bytes as hex — the same label shape holger draws a
/// puller under (`client 1a2b3c4d`). Stable within the day only.
pub fn label(&self) -> String {
let t = &self.token;
format!("client {:02x}{:02x}{:02x}{:02x}", t[0], t[1], t[2], t[3])
}
/// Test seam: a token from raw bytes, for a caller that has no address.
#[doc(hidden)]
pub fn from_raw(day: u64, token: [u8; 16]) -> DayToken {
DayToken { day, token }
}
}
/// **Who is asking.** Usage is counted per consumer.
#[derive(Clone, Debug, PartialEq, Eq, Hash)]
pub enum Consumer {
/// An authenticated principal, by the id the product knows it as.
Principal(String),
/// A namespace (tenant) — the unit gunnar bills and meters.
Namespace(String),
/// A paired peer (a replica, a mirror partner).
Peer(String),
/// A stranger, as today's [`DayToken`].
Anonymous(DayToken),
}
impl Consumer {
pub fn is_anonymous(&self) -> bool {
matches!(self, Consumer::Anonymous(_))
}
/// The usage-ledger key for a named consumer; `None` for anonymous, whose
/// usage lives only in the per-day token tally and dies with the key.
pub fn key(&self) -> Option<String> {
match self {
Consumer::Principal(p) => Some(format!("principal:{p}")),
Consumer::Namespace(n) => Some(format!("namespace:{n}")),
Consumer::Peer(p) => Some(format!("peer:{p}")),
Consumer::Anonymous(_) => None,
}
}
}
impl fmt::Display for Consumer {
fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
match self {
Consumer::Principal(p) => write!(f, "principal {p}"),
Consumer::Namespace(n) => write!(f, "namespace {n}"),
Consumer::Peer(p) => write!(f, "peer {p}"),
Consumer::Anonymous(t) => write!(f, "anonymous {}", t.label()),
}
}
}
/// **What a piece of work costs**, in the six units a box actually runs out of.
#[derive(Clone, Copy, Debug, Default, PartialEq, Eq)]
pub struct Cost {
pub cpu_ms: u64,
pub disk_read: u64,
pub disk_write: u64,
pub net_in: u64,
pub net_out: u64,
/// Bytes fetched from an upstream on the consumer's behalf.
pub upstream_fetch: u64,
}
impl Cost {
/// **The bytes charged to a consumer's budget**: uploads + downloads +
/// upstream fetch. Disk is the box's cost, not the consumer's traffic.
pub fn usage_bytes(&self) -> u64 {
self.net_in
.saturating_add(self.net_out)
.saturating_add(self.upstream_fetch)
}
pub(crate) fn fields(&self) -> [u64; 6] {
[
self.cpu_ms,
self.disk_read,
self.disk_write,
self.net_in,
self.net_out,
self.upstream_fetch,
]
}
pub(crate) fn from_fields(f: [u64; 6]) -> Cost {
Cost {
cpu_ms: f[0],
disk_read: f[1],
disk_write: f[2],
net_in: f[3],
net_out: f[4],
upstream_fetch: f[5],
}
}
}
/// **Usage over a window**: bytes moved and the window they were moved in.
/// As an input to [`Admit::plan`] it is what the consumer has USED; in
/// [`Plan::budget_left`] it is what is LEFT (`u64::MAX` when unlimited).
#[derive(Clone, Copy, Debug, PartialEq, Eq)]
pub struct Usage {
pub bytes: u64,
pub window_days: u32,
}
impl Usage {
pub fn new(bytes: u64, window_days: u32) -> Usage {
Usage { bytes, window_days }
}
/// Nothing used, over the default window.
pub fn none() -> Usage {
Usage { bytes: 0, window_days: DEFAULT_WINDOW_DAYS }
}
pub fn unlimited(window_days: u32) -> Usage {
Usage { bytes: u64::MAX, window_days }
}
pub fn is_unlimited(&self) -> bool {
self.bytes == u64::MAX
}
}
/// **The class a plan puts work in** — each maps to a SHARE of concurrent
/// slots in the product's gate, never to a place in a queue (see the module
/// docs).
#[derive(Clone, Copy, Debug, PartialEq, Eq, Hash, PartialOrd, Ord)]
pub enum Class {
/// Within budget: the ordinary share.
Full,
/// Over budget: the anonymous share.
Reduced,
/// Anonymous: the smallest share, plus a per-day-token byte budget.
Floor,
/// Non-interactive work the product will run from its own queue.
Queued,
}
impl Class {
pub fn name(self) -> &'static str {
match self {
Class::Full => "Full",
Class::Reduced => "Reduced",
Class::Floor => "Floor",
Class::Queued => "Queued",
}
}
}
impl fmt::Display for Class {
fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
f.write_str(self.name())
}
}
/// **The answer to "what would this cost, and which class does it get"** —
/// pure, taken nowhere.
#[derive(Clone, Debug, PartialEq, Eq)]
pub struct Plan {
pub class: Class,
pub estimate: Cost,
/// One human sentence, e.g. `100 GB/30d used 63 GB; this clone ~1.2 GB; Full`.
pub why: String,
/// What is left of the budget after the usage passed in (not after this
/// work — the estimate is a guess, the usage is a fact).
pub budget_left: Usage,
/// The work and the consumer the plan is for, so `admit` needs nothing else.
pub work: Work,
pub who: Consumer,
}
/// **A slot taken per a plan.** Handed back to [`Admit::record`] when the work
/// is done; dropping it without recording releases the slot and learns nothing.
#[derive(Debug)]
pub struct Guard {
pub(crate) class: Class,
pub(crate) work: Work,
pub(crate) who: Consumer,
pub(crate) slot: Option<share::SlotHold>,
}
impl Guard {
/// A guard that holds no slot — what [`PassThrough`] hands out.
pub fn unslotted(plan: &Plan) -> Guard {
Guard { class: plan.class, work: plan.work, who: plan.who.clone(), slot: None }
}
pub fn class(&self) -> Class {
self.class
}
pub fn work(&self) -> Work {
self.work
}
pub fn consumer(&self) -> &Consumer {
&self.who
}
/// Whether this guard holds one of the policy's own class slots (released
/// when it is dropped or recorded).
pub fn holds_slot(&self) -> bool {
self.slot.is_some()
}
}
/// **Why work was not admitted**, with the number that was hit.
#[derive(Clone, Debug, PartialEq, Eq)]
pub enum Refusal {
/// This anonymous day token has moved its whole day's budget.
DayTokenSpent { used: u64, limit: u64 },
/// The class's share of concurrent slots is full.
ClassFull { class: Class, in_flight: usize, share: usize },
}
impl fmt::Display for Refusal {
fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
match self {
Refusal::DayTokenSpent { used, limit } => write!(
f,
"anonymous {}/day used {}; try again after 00:00 UTC or authenticate",
short_bytes(*limit),
short_bytes(*used)
),
Refusal::ClassFull { class, in_flight, share } => write!(
f,
"the {class} share is full ({in_flight} of {share} in flight); retry shortly"
),
}
}
}
impl Refusal {
/// **Seconds until asking again can succeed**, for a `Retry-After`: a spent
/// day token until the next 00:00 UTC (the token and its budget both turn
/// over then), a full class share one second — a slot frees when any
/// transfer finishes, and a refusal that asked for more would be a queue
/// by another name.
pub fn retry_after_secs(&self, now_unix: u64) -> u64 {
match self {
Refusal::DayTokenSpent { .. } => DAY_SECS - now_unix % DAY_SECS,
Refusal::ClassFull { .. } => 1,
}
}
}
impl std::error::Error for Refusal {}
/// **The admission trait.** `Send + Sync` so one instance serves every door.
pub trait Admit: Send + Sync {
/// What this work would cost, and which class it gets — pure, no side effect.
fn plan(&self, work: &Work, who: &Consumer, usage: &Usage) -> Plan;
/// Take a slot per the plan (never blocks: a guard, or a refusal with a reason).
fn admit(&self, plan: &Plan) -> Result<Guard, Refusal>;
/// What it actually cost, afterwards — feeds usage and the meters.
fn record(&self, guard: Guard, actual: &Cost);
/// The learnt estimate, one row per [`WorkKind`] — for the admin view.
fn estimates(&self) -> Vec<EstimateRow> {
Vec::new()
}
/// The policy's name for a read-out (`pass-through`, `usage-share`).
fn policy(&self) -> &'static str;
/// The byte budget per window for a named consumer, or `None` when unlimited.
fn limit(&self) -> Option<Usage> {
None
}
/// **What `who` has used over the policy's window**, from the policy's own
/// table (lock-free for a consumer already seen). A policy with no table
/// knows nothing and says zero.
fn usage(&self, who: &Consumer) -> Usage {
let _ = who;
Usage::none()
}
}
/// **Always Full; only records.** The type a product is when it was built
/// without a share, or configured with `limit = "none"`.
#[derive(Debug, Default)]
pub struct PassThrough {
estimator: Estimator,
}
impl PassThrough {
pub fn new() -> PassThrough {
PassThrough::default()
}
}
impl Admit for PassThrough {
fn plan(&self, work: &Work, who: &Consumer, usage: &Usage) -> Plan {
let estimate = self.estimator.estimate(work);
Plan {
class: Class::Full,
estimate,
why: sentence(None, usage, work, &estimate, Class::Full),
budget_left: Usage::unlimited(usage.window_days),
work: *work,
who: who.clone(),
}
}
fn admit(&self, plan: &Plan) -> Result<Guard, Refusal> {
Ok(Guard::unslotted(plan))
}
fn record(&self, guard: Guard, actual: &Cost) {
self.estimator.learn(&guard.work, actual);
}
fn estimates(&self) -> Vec<EstimateRow> {
self.estimator.rows()
}
fn policy(&self) -> &'static str {
"pass-through"
}
}
/// `100 GB`, `1.2 GB`, `512 B` — [`nornir_humanize::bytes_si`] with a whole
/// number's `.0` dropped, which is how a person writes a budget. The `.0` is
/// cut in place: one allocation, the formatter's own.
pub fn short_bytes(n: u64) -> String {
let mut s = nornir_humanize::bytes_si(n);
if let Some(i) = s.find(".0 ") {
s.replace_range(i..i + 2, "");
}
s
}
/// **The plan's sentence.** One writer for both implementations, so the admin
/// view, the customer and a refusal all read the same words. Written into ONE
/// preallocated string; the only other allocations are the byte formatter's.
///
/// `limit`: the budget per window for this consumer, `None` when unlimited.
pub(crate) fn sentence(
limit: Option<Usage>,
used: &Usage,
work: &Work,
estimate: &Cost,
class: Class,
) -> String {
use std::fmt::Write as _;
let mut out = String::with_capacity(96);
let _ = match limit {
Some(l) if l.window_days == 1 => {
write!(out, "{}/day used {}", short_bytes(l.bytes), short_bytes(used.bytes))
}
Some(l) => write!(
out,
"{}/{}d used {}",
short_bytes(l.bytes),
l.window_days,
short_bytes(used.bytes)
),
None => write!(out, "no limit; {}d used {}", used.window_days, short_bytes(used.bytes)),
};
let est = estimate.usage_bytes();
let _ = if est == 0 {
write!(out, "; this {} ~? (no estimate yet); {class}", work.kind)
} else {
write!(out, "; this {} ~{}; {class}", work.kind, short_bytes(est))
};
out
}
#[cfg(test)]
mod tests {
use super::*;
#[test]
fn pass_through_is_always_full_and_never_refuses() {
let p = PassThrough::new();
let who = Consumer::Namespace("acme".into());
let used = Usage::new(u64::MAX / 2, 30);
let plan = p.plan(&Work::new(WorkKind::Clone, Some(1_200_000_000)), &who, &used);
assert_eq!(plan.class, Class::Full);
assert!(plan.budget_left.is_unlimited());
for _ in 0..1000 {
let g = p.admit(&plan).expect("pass-through never refuses");
p.record(g, &Cost::default());
}
}
#[test]
fn the_sentence_reads_like_the_design() {
let s = sentence(
Some(Usage::new(100_000_000_000, 30)),
&Usage::new(63_000_000_000, 30),
&Work::new(WorkKind::Clone, None),
&Cost { net_out: 1_200_000_000, ..Cost::default() },
Class::Full,
);
assert_eq!(s, "100 GB/30d used 63 GB; this clone ~1.2 GB; Full");
}
#[test]
fn usage_bytes_is_traffic_not_disk() {
let c = Cost {
cpu_ms: 9,
disk_read: 1000,
disk_write: 1000,
net_in: 1,
net_out: 2,
upstream_fetch: 4,
};
assert_eq!(c.usage_bytes(), 7);
assert_eq!(Cost::from_fields(c.fields()), c);
}
#[test]
fn the_policies_are_send_and_sync_and_object_safe() {
fn is<T: Send + Sync>() {}
is::<PassThrough>();
is::<UsageShare>();
is::<DayKey>();
is::<Table>();
is::<Guard>();
is::<std::sync::Arc<dyn Admit>>();
}
#[test]
fn retry_after_is_the_next_midnight_for_a_spent_token() {
let r = Refusal::DayTokenSpent { used: 1, limit: 1 };
assert_eq!(r.retry_after_secs(20_000 * DAY_SECS + 3600), DAY_SECS - 3600);
assert_eq!(Refusal::ClassFull { class: Class::Full, in_flight: 1, share: 1 }.retry_after_secs(7), 1);
}
#[test]
fn refusals_say_what_was_hit() {
let r = Refusal::DayTokenSpent { used: 5_100_000_000, limit: 5_000_000_000 };
assert_eq!(
r.to_string(),
"anonymous 5 GB/day used 5.1 GB; try again after 00:00 UTC or authenticate"
);
let r = Refusal::ClassFull { class: Class::Reduced, in_flight: 5, share: 5 };
assert!(r.to_string().contains("Reduced share is full (5 of 5"));
}
}