server/lib/src/admission.rs
//! **Download admission** — who may take how much of the door, asked through
//! the shared trait `edda_admit::Admit`.
//!
//! Until this module holger had ONE admission: the paid upload quota
//! (`metered::Quota`, asked before a PUT body is read). Reads had none — the
//! accept loop's connection ceiling is server-wide, so one client with a
//! parallel downloader could hold every slot and every other consumer waited
//! behind it.
//!
//! ## The shape
//!
//! * **Who** — [`Who`]: an authenticated principal, a paired downstream holger
//! ([`crate::peers`], known by its certificate), or an anonymous client named
//! only by its per-day token (`edda_admit::DayKey` — a keyed hash whose key is
//! rotated, lock-free, at UTC midnight; no address is kept).
//! * **The decision, ONCE per request** — `Admit::usage` → `Admit::plan` →
//! `Admit::admit` at the door, `Admit::record` when the response is done.
//! Nothing per byte. The DEFAULT is `PassThrough` (every plan `Full`, nothing
//! refused — today's behaviour); `UsageShare` (holger feature `admit-share`)
//! is the paid one: 100 GB / 30 d `Full` per customer, then `Reduced`;
//! anonymous `Floor`, 5 GB / day per day token. Its usage table is edda's
//! sharded, preallocated, lock-free `Table`.
//! * **The share** — holger's own, per consumer per class: at most
//! [`AdmissionConfig::ron_share_full`] requests in flight for one `Full`
//! consumer, fewer for `Reduced` and `Floor`, and
//! [`AdmissionConfig::ron_max_in_flight`] for the whole door (one atomic).
//! The per-consumer counts live in a table SHARDED per core — no global lock.
//! Over its share a request is answered AT ONCE with `503` + `Retry-After`.
//! * **The book** — [`CostBook`]: per consumer, per UTC day, per [`WorkKind`],
//! the estimated and actual [`CostTally`], sharded per core like the share
//! table. It is what the admin view prints, what is flushed to redb every
//! minute and at shutdown ([`CostBook::flush`], rows of kind `admit_usage` in
//! the job ledger `jobs/jobs.redb`), and what `UsageShare` is SEEDED from on
//! first sight of a consumer after a restart ([`CostBook::seed_days`]). The
//! request path only touches memory.
//!
//! ## Unconfigured ⇒ unchanged
//!
//! No `admission:` block in the RON ⇒ [`DoorAdmission::pass_through`]: the
//! trait is still asked (so the book is kept and the admin view has numbers),
//! but no share is enforced and no request is ever refused here.
use crate::lock::LockOrRecover as _;
use std::collections::{BTreeMap, HashMap};
use std::hash::{Hash, Hasher};
use std::sync::atomic::{AtomicU32, Ordering};
use std::sync::{Arc, Mutex, OnceLock};
use serde::{Deserialize, Serialize};
use edda_admit as admit;
pub use edda_admit::DayToken;
/// One UTC day, the book's bucket.
const DAY_SECS: u64 = 86_400;
/// How many days of history a consumer's usage covers (`UsageShare`'s window).
pub const WINDOW_DAYS: u64 = 30;
/// Consumers ONE shard of the book keeps at most. An anonymous token is
/// per-day, so a shard would otherwise grow with every client of every day;
/// past the cap the anonymous rows not touched today are dropped first.
const SHARD_CAP: usize = 16_384;
/// The jera `kind` the book's durable rows carry. NOT `holger_`-prefixed, so
/// the job queue's recovery sweep and its `list` never see them.
pub const USAGE_KIND: &str = "admit_usage";
/// Their jera status: not a word jera's pruner knows, so it never drops one.
const USAGE_STATUS: &str = "ledger";
// ─────────────────────────────────────────────────────────────────────────────
// Sharding
// ─────────────────────────────────────────────────────────────────────────────
/// **A map sharded per core**: `4 × cores` (power of two) independent locks, a
/// key's shard picked by its hash. Two requests contend only when their
/// consumers hash to one shard — there is no lock every request takes.
struct Sharded<V> {
shards: Box<[Mutex<HashMap<String, V>>]>,
mask: usize,
}
impl<V> Sharded<V> {
fn new() -> Self {
let cores = std::thread::available_parallelism()
.map(|n| n.get())
.unwrap_or(4);
let n = (cores * 4).next_power_of_two().clamp(8, 1024);
Self {
shards: (0..n).map(|_| Mutex::new(HashMap::new())).collect(),
mask: n - 1,
}
}
fn shard(&self, key: &str) -> std::sync::MutexGuard<'_, HashMap<String, V>> {
let mut h = std::collections::hash_map::DefaultHasher::new();
key.hash(&mut h);
self.shards[(h.finish() as usize) & self.mask].lock_or_recover()
}
fn each_shard(&self) -> impl Iterator<Item = std::sync::MutexGuard<'_, HashMap<String, V>>> {
self.shards.iter().map(|s| s.lock_or_recover())
}
}
// ─────────────────────────────────────────────────────────────────────────────
// Who
// ─────────────────────────────────────────────────────────────────────────────
/// **A consumer, as holger can name one.**
#[derive(Clone, Debug, PartialEq, Eq, Hash)]
pub enum Who {
/// An authenticated principal (OIDC `sub`, mTLS CN, a static token's name).
Principal(String),
/// A paired downstream holger, by its pairing name.
Peer(String),
/// An anonymous client: its per-day token (`edda_admit::visitors`, a keyed
/// hash whose key dies at UTC midnight). Tomorrow it is a different token.
Anonymous(DayToken),
}
impl Who {
/// The book's key: `principal:<sub>`, `peer:<name>`, `anon:<token>`.
pub fn key(&self) -> String {
match self {
Who::Principal(s) => format!("principal:{s}"),
Who::Peer(s) => format!("peer:{s}"),
Who::Anonymous(t) => {
format!("anon:{}:{}", t.day, t.label().trim_start_matches("client "))
}
}
}
/// Parse a NAMED book key back (the admin door's `?consumer=`). An
/// anonymous key cannot be turned back into its token — by design.
pub fn from_key(key: &str) -> Option<Who> {
let (kind, name) = key.split_once(':')?;
match kind {
"principal" => Some(Who::Principal(name.to_string())),
"peer" => Some(Who::Peer(name.to_string())),
_ => None,
}
}
/// Resolve the consumer the doors see: a paired peer first (its
/// certificate is the strongest statement), then a principal other than
/// `anonymous`, then the anonymous per-day token of the client address.
pub fn resolve(peer: Option<&str>, principal: Option<&str>, anon_token: DayToken) -> Who {
if let Some(p) = peer.filter(|p| !p.is_empty()) {
return Who::Peer(p.to_string());
}
if let Some(p) = principal.filter(|p| !p.is_empty() && *p != "anonymous") {
return Who::Principal(p.to_string());
}
Who::Anonymous(anon_token)
}
fn to_consumer(&self) -> admit::Consumer {
match self {
Who::Principal(s) => admit::Consumer::Principal(s.clone()),
Who::Peer(s) => admit::Consumer::Peer(s.clone()),
Who::Anonymous(t) => admit::Consumer::Anonymous(*t),
}
}
/// The named consumer back from edda's (`UsageShare`'s seed asks by it).
/// An anonymous one is `None`: its usage is its day token's, never seeded.
pub fn of_consumer(c: &admit::Consumer) -> Option<Who> {
match c {
admit::Consumer::Principal(s) => Some(Who::Principal(s.clone())),
admit::Consumer::Peer(s) => Some(Who::Peer(s.clone())),
admit::Consumer::Namespace(s) => Some(Who::Principal(s.clone())),
admit::Consumer::Anonymous(_) => None,
}
}
}
// ─────────────────────────────────────────────────────────────────────────────
// Work and cost
// ─────────────────────────────────────────────────────────────────────────────
/// **What a request asks the door to do.**
#[derive(Clone, Copy, Debug, PartialEq, Eq, Hash, PartialOrd, Ord, Serialize, Deserialize)]
#[serde(rename_all = "snake_case")]
pub enum WorkKind {
/// A read of an artifact (`GET`, `FetchArtifact`, `StreamArtifact`).
Download,
/// A write (`PUT`/`POST`, `PutArtifact`).
Upload,
/// The upstream half of a proxy read (a `Job::Fetch`, or an inline walk).
PullThrough,
/// One mirror asset (`Job::MirrorAsset`).
MirrorAsset,
}
impl WorkKind {
pub fn word(self) -> &'static str {
match self {
WorkKind::Download => "download",
WorkKind::Upload => "upload",
WorkKind::PullThrough => "pull_through",
WorkKind::MirrorAsset => "mirror_asset",
}
}
/// The shared kind: a download is edda's `ArtifactDownload`, an upload its
/// `Push` (one estimator per product, so the names do not collide).
pub fn shared(self) -> admit::WorkKind {
match self {
WorkKind::Download => admit::WorkKind::ArtifactDownload,
WorkKind::Upload => admit::WorkKind::Push,
WorkKind::PullThrough => admit::WorkKind::PullThrough,
WorkKind::MirrorAsset => admit::WorkKind::MirrorAsset,
}
}
fn to_work(self, bytes: u64) -> admit::Work {
admit::Work::new(self.shared(), (bytes > 0).then_some(bytes))
}
}
/// holger's own copy of `admit::Cost` — the same six fields, summable and
/// serializable, so the book and the admin door do not depend on the trait
/// crate's derives.
#[derive(Clone, Copy, Debug, Default, PartialEq, Eq, Serialize, Deserialize)]
pub struct CostTally {
pub cpu_ms: u64,
pub disk_read: u64,
pub disk_write: u64,
pub net_in: u64,
pub net_out: u64,
pub upstream_fetch: u64,
}
impl CostTally {
pub fn add(&mut self, o: &CostTally) {
self.cpu_ms += o.cpu_ms;
self.disk_read += o.disk_read;
self.disk_write += o.disk_write;
self.net_in += o.net_in;
self.net_out += o.net_out;
self.upstream_fetch += o.upstream_fetch;
}
/// Bytes that crossed a wire for this consumer — what a share is sold in.
pub fn bytes(&self) -> u64 {
self.net_in + self.net_out + self.upstream_fetch
}
fn from_cost(c: &admit::Cost) -> Self {
Self {
cpu_ms: c.cpu_ms,
disk_read: c.disk_read,
disk_write: c.disk_write,
net_in: c.net_in,
net_out: c.net_out,
upstream_fetch: c.upstream_fetch,
}
}
fn to_cost(self) -> admit::Cost {
admit::Cost {
cpu_ms: self.cpu_ms,
disk_read: self.disk_read,
disk_write: self.disk_write,
net_in: self.net_in,
net_out: self.net_out,
upstream_fetch: self.upstream_fetch,
}
}
}
/// The admission class, as holger prints and schedules it.
#[derive(Clone, Copy, Debug, PartialEq, Eq, Serialize, Deserialize)]
#[serde(rename_all = "snake_case")]
pub enum ClassWord {
Full,
Reduced,
Floor,
Queued,
}
impl ClassWord {
fn of(c: &admit::Class) -> Self {
match c {
admit::Class::Full => ClassWord::Full,
admit::Class::Reduced => ClassWord::Reduced,
admit::Class::Floor => ClassWord::Floor,
admit::Class::Queued => ClassWord::Queued,
}
}
/// The pull-through queue priority for this class (`jobs::PRIORITY_*`
/// values): a `Reduced`/`Floor`/`Queued` consumer's fetch waits behind a
/// `Full` one's.
pub fn priority(self) -> u8 {
match self {
ClassWord::Full => 0,
ClassWord::Reduced => 1,
ClassWord::Floor => 2,
ClassWord::Queued => 3,
}
}
}
// ─────────────────────────────────────────────────────────────────────────────
// The book
// ─────────────────────────────────────────────────────────────────────────────
/// One work kind's numbers for one consumer.
#[derive(Clone, Debug, Default, PartialEq, Eq, Serialize, Deserialize)]
pub struct WorkTally {
pub requests: u64,
pub refused: u64,
pub estimated: CostTally,
pub actual: CostTally,
}
#[derive(Default)]
struct ConsumerRow {
/// `day → actual cost` for the last [`WINDOW_DAYS`].
days: BTreeMap<u64, CostTally>,
by_work: BTreeMap<WorkKind, WorkTally>,
last_class: Option<ClassWord>,
last_why: String,
in_flight: u32,
touched_day: u64,
/// The last day this consumer made a request IN THIS PROCESS (0 = none):
/// `touched_day` is also set by a load, so it is no evidence of a request.
seen_day: u64,
/// Charged since the last [`CostBook::flush`].
dirty: bool,
}
/// **Per-consumer usage**, sharded per core. One per process ([`book`]).
pub struct CostBook {
rows: Sharded<ConsumerRow>,
}
impl Default for CostBook {
fn default() -> Self {
Self {
rows: Sharded::new(),
}
}
}
/// One consumer's usage as the admin door prints it.
#[derive(Clone, Debug, Default, PartialEq, Eq, Serialize, Deserialize)]
pub struct UsageView {
pub consumer: String,
/// Actual cost over the last [`WINDOW_DAYS`] days.
pub window: CostTally,
/// Actual cost today (UTC).
pub today: CostTally,
pub by_work: BTreeMap<String, WorkTally>,
/// The class the last plan gave it, and why.
pub class: Option<ClassWord>,
pub why: String,
pub in_flight: u32,
}
/// The durable row: one per NAMED consumer (an anonymous token dies at
/// midnight, so there is nothing to keep it under).
#[derive(Serialize, Deserialize)]
struct UsageRecord {
consumer: String,
days: Vec<(u64, CostTally)>,
}
static BOOK: OnceLock<CostBook> = OnceLock::new();
/// The process's book (the doors, the proxy and the job runner share it).
pub fn book() -> &'static CostBook {
BOOK.get_or_init(CostBook::default)
}
/// **The process's consumers, today** (the estate board, lane A13).
pub fn consumers() -> Option<crate::topology::ConsumerCount> {
Some(book().consumers_on(today()))
}
fn today() -> u64 {
crate::visitors::now_unix() / DAY_SECS
}
fn window_sum(r: &ConsumerRow, day: u64) -> (CostTally, CostTally) {
let first = day.saturating_sub(WINDOW_DAYS - 1);
let mut window = CostTally::default();
for (_, c) in r.days.range(first..=day) {
window.add(c);
}
(window, r.days.get(&day).copied().unwrap_or_default())
}
impl CostBook {
fn with_row<R>(&self, key: &str, f: impl FnOnce(&mut ConsumerRow) -> R) -> R {
let mut shard = self.rows.shard(key);
let day = today();
if !shard.contains_key(key) && shard.len() >= SHARD_CAP {
shard.retain(|k, r| !k.starts_with("anon:") || r.touched_day >= day);
}
let row = shard.entry(key.to_string()).or_default();
row.touched_day = day;
f(row)
}
/// Charge `cost` for `work` to `key` on day `day`.
fn charge_on(&self, key: &str, work: WorkKind, cost: &CostTally, day: u64) {
self.with_row(key, |r| {
r.days.entry(day).or_default().add(cost);
let first = day.saturating_sub(WINDOW_DAYS - 1);
r.days.retain(|d, _| *d >= first);
r.by_work.entry(work).or_default().actual.add(cost);
r.dirty = true;
})
}
/// Charge the actual cost of finished work to `key`, today.
pub fn charge(&self, key: &str, work: WorkKind, cost: &CostTally) {
self.charge_on(key, work, cost, today());
}
/// The proxy took `bytes` from an upstream for `key` (the job runner's case:
/// a `Job::Fetch` has no ticket).
pub fn add_upstream_fetch(&self, key: &str, bytes: u64) {
let cost = CostTally {
upstream_fetch: bytes,
..Default::default()
};
self.charge(key, WorkKind::PullThrough, &cost);
}
fn note_plan(&self, key: &str, work: WorkKind, class: ClassWord, why: &str, est: &CostTally) {
self.with_row(key, |r| {
r.seen_day = r.touched_day;
let w = r.by_work.entry(work).or_default();
w.requests += 1;
w.estimated.add(est);
r.last_class = Some(class);
if r.last_why != why {
r.last_why = why.to_string();
}
})
}
fn note_refused(&self, key: &str, work: WorkKind) {
self.with_row(key, |r| r.by_work.entry(work).or_default().refused += 1)
}
fn set_in_flight(&self, key: &str, n: u32) {
self.with_row(key, |r| r.in_flight = n)
}
/// How many NAMED consumers the book holds (0 on a fresh boot with an
/// empty ledger — the cue to seed from the audit log).
pub fn named_rows(&self) -> usize {
self.rows
.each_shard()
.map(|s| s.keys().filter(|k| !k.starts_with("anon:")).count())
.sum()
}
/// **Every consumer, counted** (lane A13, the estate board): named
/// principals and paired peers with a request or a charge in the last
/// [`WINDOW_DAYS`] — the charges come back from the ledger after a restart —
/// and the anonymous day tokens of `day`, which exist for that day only.
/// Shard by shard, never all shards at once.
pub fn consumers_on(&self, day: u64) -> crate::topology::ConsumerCount {
let first = day.saturating_sub(WINDOW_DAYS - 1);
let anon_today = format!("anon:{day}:");
let mut c = crate::topology::ConsumerCount {
counted: true,
window_days: WINDOW_DAYS as u32,
..Default::default()
};
for shard in self.rows.each_shard() {
for (k, r) in shard.iter() {
let active = r.seen_day >= first || r.days.range(first..=day).next().is_some();
if k.starts_with("principal:") && active {
c.principals += 1;
} else if k.starts_with("peer:") && active {
c.peers += 1;
} else if k.starts_with(&anon_today) {
c.anonymous_today += 1;
}
}
}
c
}
/// `(window, today)` actual cost of `key`.
pub fn usage_of(&self, key: &str) -> (CostTally, CostTally) {
let shard = self.rows.shard(key);
shard
.get(key)
.map(|r| window_sum(r, today()))
.unwrap_or_default()
}
/// **`UsageShare`'s seed**: `(day, bytes)` of `key` from `first_day` to
/// `today` — what this book held (loaded from redb) when the policy first
/// saw the consumer in this process.
pub fn seed_days(&self, key: &str, first_day: u64, today: u64) -> Vec<(u64, u64)> {
let shard = self.rows.shard(key);
shard
.get(key)
.map(|r| {
r.days
.range(first_day..=today)
.map(|(d, c)| (*d, c.bytes()))
.collect()
})
.unwrap_or_default()
}
/// One consumer's view.
pub fn view(&self, key: &str) -> Option<UsageView> {
let shard = self.rows.shard(key);
let r = shard.get(key)?;
let (window, today_cost) = window_sum(r, today());
Some(UsageView {
consumer: key.to_string(),
window,
today: today_cost,
by_work: r
.by_work
.iter()
.map(|(k, v)| (k.word().to_string(), v.clone()))
.collect(),
class: r.last_class,
why: r.last_why.clone(),
in_flight: r.in_flight,
})
}
/// Every consumer's view, heaviest window first; `limit == 0` = all. The
/// admin path: shard by shard, never all shards at once.
pub fn views(&self, limit: usize) -> Vec<UsageView> {
let day = today();
let mut out = Vec::new();
for shard in self.rows.each_shard() {
for (k, r) in shard.iter() {
let (window, today_cost) = window_sum(r, day);
out.push(UsageView {
consumer: k.clone(),
window,
today: today_cost,
by_work: r
.by_work
.iter()
.map(|(k, v)| (k.word().to_string(), v.clone()))
.collect(),
class: r.last_class,
why: r.last_why.clone(),
in_flight: r.in_flight,
});
}
}
out.sort_by(|a, b| {
b.window
.bytes()
.cmp(&a.window.bytes())
.then(a.consumer.cmp(&b.consumer))
});
if limit > 0 {
out.truncate(limit);
}
out
}
/// **Write every named consumer charged since the last flush** to the job
/// ledger (one `admit_usage` row each, overwritten in place). Off the
/// request path: the minute timer and the shutdown call it. Rows flushed.
pub fn flush(&self, ledger: &jera::jobs::JobStore) -> anyhow::Result<usize> {
let mut dirty: Vec<UsageRecord> = Vec::new();
for mut shard in self.rows.each_shard() {
for (k, r) in shard.iter_mut() {
if r.dirty && !k.starts_with("anon:") {
r.dirty = false;
dirty.push(UsageRecord {
consumer: k.clone(),
days: r.days.iter().map(|(d, c)| (*d, *c)).collect(),
});
}
}
}
let now = crate::audit::now_nanos() / 1_000;
for rec in &dirty {
let row = jera::jobs::JobRecord {
job_id: format!("{USAGE_KIND}:{}", rec.consumer),
kind: USAGE_KIND.to_string(),
target: rec.consumer.clone(),
workspace: "holger".to_string(),
status: USAGE_STATUS.to_string(),
ts_start_micros: now,
ts_end_micros: None,
elapsed_ms: None,
detail_json: serde_json::to_string(rec).unwrap_or_default(),
result_ref: String::new(),
parent_id: None,
node: None,
phase: None,
actor: None,
actor_method: None,
};
ledger.submit(&row)?;
}
Ok(dirty.len())
}
/// **Load the flushed rows** (boot, before the doors serve). Rows loaded;
/// `0` means the ledger holds none yet and the audit log is the seed.
pub fn load(&self, ledger: &jera::jobs::JobStore) -> anyhow::Result<usize> {
let day = today();
let first = day.saturating_sub(WINDOW_DAYS - 1);
let mut n = 0;
for row in ledger.list(&jera::jobs::JobSelector::All)? {
if row.kind != USAGE_KIND {
continue;
}
let Ok(rec) = serde_json::from_str::<UsageRecord>(&row.detail_json) else {
continue;
};
self.with_row(&rec.consumer, |r| {
for (d, c) in rec.days.iter().filter(|(d, _)| *d >= first) {
*r.days.entry(*d).or_default() = *c;
}
});
n += 1;
}
Ok(n)
}
/// **Seed from the audit log** — the first boot with this code, when the
/// ledger has no `admit_usage` rows yet: every successful download
/// (`net_out`) and upload (`net_in`) of the last [`WINDOW_DAYS`] by a
/// principal or a paired peer. Anonymous rows are NOT seeded — yesterday's
/// token is gone by design. Returns records charged.
pub fn seed(&self, events: &[crate::audit::AuditEvent], now_unix: u64) -> usize {
let day_now = now_unix / DAY_SECS;
let first = day_now.saturating_sub(WINDOW_DAYS - 1);
let mut n = 0;
for e in events {
if e.ident.is_empty() || e.ident == "anonymous" || !(200..300).contains(&e.status) {
continue;
}
let day = (e.ts_nanos.max(0) as u64 / 1_000_000_000) / DAY_SECS;
if day < first || day > day_now {
continue;
}
let (work, cost) = match e.action {
crate::audit::AuditAction::Download => (
WorkKind::Download,
CostTally {
net_out: e.bytes,
..Default::default()
},
),
crate::audit::AuditAction::Upload => (
WorkKind::Upload,
CostTally {
net_in: e.bytes,
..Default::default()
},
),
_ => continue,
};
let who = match e.ident.strip_prefix(crate::peers::PEER_PRINCIPAL_PREFIX) {
Some(peer) => Who::Peer(peer.to_string()),
None => Who::Principal(e.ident.clone()),
};
self.charge_on(&who.key(), work, &cost, day);
n += 1;
}
n
}
}
// ─────────────────────────────────────────────────────────────────────────────
// The door
// ─────────────────────────────────────────────────────────────────────────────
fn default_max_in_flight() -> u32 {
256
}
fn default_share_full() -> u32 {
64
}
fn default_share_reduced() -> u32 {
16
}
fn default_share_floor() -> u32 {
4
}
fn default_retry_after() -> u64 {
2
}
fn default_pull_wait_ms() -> u64 {
2_000
}
fn default_limit() -> String {
"100 GB".to_string()
}
fn default_window_days() -> u32 {
WINDOW_DAYS as u32
}
fn default_anon_per_day() -> String {
"5 GB".to_string()
}
/// Which `Admit` implementation decides the class.
#[derive(Clone, Copy, Debug, Default, PartialEq, Eq, Serialize, Deserialize)]
pub enum AdmitPolicy {
/// Every plan `Full`, nothing refused (the default, and today's behaviour).
#[default]
PassThrough,
/// `UsageShare`: `ron_limit` / `ron_window_days` `Full` per named
/// consumer, then `Reduced`; anonymous → `Floor`, `ron_anonymous_per_day`.
/// Needs the `admit-share` feature; naming it in a build without is a boot
/// refusal.
UsageShare,
}
/// **The `admission:` block of the RON config.** Absent ⇒ no share is
/// enforced and no request is refused here.
#[derive(Clone, Debug, PartialEq, Eq, Serialize, Deserialize)]
#[serde(deny_unknown_fields)]
pub struct AdmissionConfig {
#[serde(default)]
pub ron_policy: AdmitPolicy,
/// Requests in flight through the doors, all consumers together.
#[serde(default = "default_max_in_flight")]
pub ron_max_in_flight: u32,
/// Requests in flight for ONE consumer planned `Full`.
#[serde(default = "default_share_full")]
pub ron_share_full: u32,
/// … planned `Reduced`.
#[serde(default = "default_share_reduced")]
pub ron_share_reduced: u32,
/// … planned `Floor` or `Queued`.
#[serde(default = "default_share_floor")]
pub ron_share_floor: u32,
/// The `Retry-After` of a request refused for want of a share.
#[serde(default = "default_retry_after")]
pub ron_retry_after_secs: u64,
/// Proxy reads of artifacts that are not local go through the durable job
/// queue (`Job::Fetch`) and are answered `503` + `Retry-After` while they
/// are fetched. Off ⇒ the inline walk, as before.
#[serde(default)]
pub ron_pull_through_queue: bool,
/// How long a `Full` request waits for its queued fetch before it is
/// answered 503. Any other class never waits: it gets its 503 at once and
/// holds no connection.
#[serde(default = "default_pull_wait_ms")]
pub ron_pull_through_wait_ms: u64,
/// `UsageShare`: bytes per window a named consumer moves at `Full`
/// (`"100 GB"`, `"none"` = unlimited).
#[serde(default = "default_limit")]
pub ron_limit: String,
/// `UsageShare`: the rolling window, UTC days.
#[serde(default = "default_window_days")]
pub ron_window_days: u32,
/// `UsageShare`: bytes one anonymous day token may move per UTC day.
#[serde(default = "default_anon_per_day")]
pub ron_anonymous_per_day: String,
}
impl Default for AdmissionConfig {
fn default() -> Self {
Self {
ron_policy: AdmitPolicy::PassThrough,
ron_max_in_flight: default_max_in_flight(),
ron_share_full: default_share_full(),
ron_share_reduced: default_share_reduced(),
ron_share_floor: default_share_floor(),
ron_retry_after_secs: default_retry_after(),
ron_pull_through_queue: false,
ron_pull_through_wait_ms: default_pull_wait_ms(),
ron_limit: default_limit(),
ron_window_days: default_window_days(),
ron_anonymous_per_day: default_anon_per_day(),
}
}
}
impl AdmissionConfig {
fn share_of(&self, class: ClassWord) -> u32 {
match class {
ClassWord::Full => self.ron_share_full,
ClassWord::Reduced => self.ron_share_reduced,
ClassWord::Floor | ClassWord::Queued => self.ron_share_floor,
}
.max(1)
}
/// The `Admit` this block names, or a refusal by name. `UsageShare` is
/// SEEDED from the book (itself loaded from redb at boot): a consumer's
/// window survives a restart.
pub fn build_admit(&self) -> anyhow::Result<Arc<dyn admit::Admit>> {
match self.ron_policy {
AdmitPolicy::PassThrough => Ok(Arc::new(admit::PassThrough::new())),
#[cfg(feature = "admit-share")]
AdmitPolicy::UsageShare => {
let size = |field: &str, s: &str| {
admit::parse_bytes(s).map_err(|e| anyhow::anyhow!("admission.{field}: {e}"))
};
let seed = |who: &admit::Consumer, first: u64, today: u64| -> Vec<(u64, u64)> {
match Who::of_consumer(who) {
Some(w) => book().seed_days(&w.key(), first, today),
None => Vec::new(),
}
};
Ok(admit::Settings {
limit: size("ron_limit", &self.ron_limit)?,
window_days: self.ron_window_days.max(1),
anonymous_per_day: size("ron_anonymous_per_day", &self.ron_anonymous_per_day)?,
// holger keeps its OWN per-consumer share table (below); the
// policy decides the class and the anonymous day budget.
slots: None,
..admit::Settings::default()
}
.build_seeded(Some(Arc::new(seed))))
}
#[cfg(not(feature = "admit-share"))]
AdmitPolicy::UsageShare => anyhow::bail!(
"admission.ron_policy = UsageShare, but this holger was built without the \
`admit-share` feature — refusing to start with a weaker policy than the one \
the config names"
),
}
}
}
/// A request refused at the door: answer `503` + `Retry-After`.
#[derive(Clone, Debug, PartialEq, Eq)]
pub struct Busy {
pub retry_after_secs: u64,
pub why: String,
}
impl std::fmt::Display for Busy {
fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
write!(f, "{} (retry in {} s)", self.why, self.retry_after_secs)
}
}
/// **The door's admission**: the trait, the day key, the share table, the
/// queue. ONE per process, shared by both transports.
pub struct DoorAdmission {
admit: Arc<dyn admit::Admit>,
/// The anonymous consumer's day key — lock-free, rotated at UTC midnight.
day_key: admit::DayKey,
/// `None` ⇒ pass-through: nothing is refused, no share enforced.
cfg: Option<AdmissionConfig>,
/// Per consumer: requests in flight. Sharded per core.
in_flight: Sharded<u32>,
/// All consumers: one atomic, claimed by CAS against the ceiling.
total: AtomicU32,
#[cfg(feature = "jobs")]
pull: OnceLock<Arc<crate::pull_through::PullThrough>>,
}
impl DoorAdmission {
/// Today's behaviour: `PassThrough`, no shares, no queue.
pub fn pass_through() -> Arc<Self> {
Arc::new(Self::with(Arc::new(admit::PassThrough::new()), None))
}
/// From the RON block (or its absence).
pub fn from_config(cfg: Option<&AdmissionConfig>) -> anyhow::Result<Arc<Self>> {
match cfg {
None => Ok(Self::pass_through()),
Some(c) => Ok(Arc::new(Self::with(c.build_admit()?, Some(c.clone())))),
}
}
/// Any `Admit` over a share table (a test's seam).
pub fn with(admit: Arc<dyn admit::Admit>, cfg: Option<AdmissionConfig>) -> Self {
Self {
admit,
day_key: admit::DayKey::new(crate::visitors::now_unix()),
cfg,
in_flight: Sharded::new(),
total: AtomicU32::new(0),
#[cfg(feature = "jobs")]
pull: OnceLock::new(),
}
}
pub fn config(&self) -> Option<&AdmissionConfig> {
self.cfg.as_ref()
}
/// **The anonymous consumer's token** for `client`, today. Lock-free.
pub fn day_token(&self, client: crate::visitors::Client<'_>) -> DayToken {
self.day_key.token(client, crate::visitors::now_unix())
}
/// Install the queued pull-through (only when the block asks for it).
#[cfg(feature = "jobs")]
pub fn attach_pull_through(&self, pt: Arc<crate::pull_through::PullThrough>) {
let _ = self.pull.set(pt);
}
#[cfg(feature = "jobs")]
pub fn pull_through(&self) -> Option<&Arc<crate::pull_through::PullThrough>> {
self.pull.get()
}
/// Requests in flight through the doors, all consumers.
pub fn in_flight_total(&self) -> u32 {
self.total.load(Ordering::Relaxed)
}
/// The policy's name (`pass-through`, `usage-share`).
pub fn policy(&self) -> &'static str {
self.admit.policy()
}
/// The policy's learnt estimate per work kind, for the admin view.
pub fn estimates(&self) -> Vec<EstimateView> {
self.admit
.estimates()
.into_iter()
.map(|r| EstimateView {
kind: r.kind.name().to_string(),
samples: r.samples,
hinted: r.hinted,
mean: CostTally::from_cost(&r.mean),
per_gb: r.per_gb.as_ref().map(CostTally::from_cost),
})
.collect()
}
/// **Plan only** — what `Admit::plan` answers for `who` doing `work` now.
/// The admin door's `admit plan`.
pub fn plan(&self, who: &Who, work: WorkKind, estimate_bytes: u64) -> PlanView {
let consumer = who.to_consumer();
let usage = self.admit.usage(&consumer);
let p = self
.admit
.plan(&work.to_work(estimate_bytes), &consumer, &usage);
PlanView {
consumer: who.key(),
work: work.word().to_string(),
class: ClassWord::of(&p.class),
estimate: CostTally::from_cost(&p.estimate),
why: p.why.clone(),
used: usage.bytes,
budget_left: (!p.budget_left.is_unlimited()).then_some(p.budget_left.bytes),
share: self
.cfg
.as_ref()
.map(|c| c.share_of(ClassWord::of(&p.class))),
policy: self.admit.policy().to_string(),
}
}
/// **Admit one request**, or refuse it at once. ONCE per request: usage →
/// plan → admit here, record when the ticket drops. No global lock: the
/// policy's table is edda's lock-free one, this door's per-consumer count
/// is one shard, the door total one atomic.
pub fn enter(
self: &Arc<Self>,
who: &Who,
work: WorkKind,
estimate_bytes: u64,
) -> Result<Ticket, Busy> {
let key = who.key();
let consumer = who.to_consumer();
let usage = self.admit.usage(&consumer);
let plan = self
.admit
.plan(&work.to_work(estimate_bytes), &consumer, &usage);
let class = ClassWord::of(&plan.class);
let estimate = CostTally::from_cost(&plan.estimate);
book().note_plan(&key, work, class, &plan.why, &estimate);
let retry_after_secs = self
.cfg
.as_ref()
.map(|c| c.ron_retry_after_secs)
.unwrap_or(1)
.max(1);
let guard = match self.admit.admit(&plan) {
Ok(g) => g,
Err(refusal) => {
book().note_refused(&key, work);
return Err(Busy {
retry_after_secs: refusal.retry_after_secs(crate::visitors::now_unix()).max(1),
why: format!("admission refused {key}: {refusal}"),
});
}
};
if let Some(cfg) = &self.cfg {
// The door ceiling: one CAS loop on one atomic.
let ceiling = cfg.ron_max_in_flight.max(1);
let claimed = self
.total
.fetch_update(Ordering::AcqRel, Ordering::Acquire, |n| {
(n < ceiling).then_some(n + 1)
});
if let Err(total) = claimed {
book().note_refused(&key, work);
self.admit.record(guard, &admit::Cost::default());
return Err(Busy {
retry_after_secs,
why: format!("the door is full ({total} in flight, ceiling {ceiling})"),
});
}
// The consumer's share: one shard, checked and taken together.
let share = cfg.share_of(class);
let mine = {
let mut shard = self.in_flight.shard(&key);
let n = shard.entry(key.clone()).or_insert(0);
if *n >= share {
let held = *n;
if held == 0 {
shard.remove(&key);
}
Err(held)
} else {
*n += 1;
Ok(*n)
}
};
match mine {
Ok(n) => book().set_in_flight(&key, n),
Err(held) => {
self.total.fetch_sub(1, Ordering::AcqRel);
book().note_refused(&key, work);
self.admit.record(guard, &admit::Cost::default());
return Err(Busy {
retry_after_secs,
why: format!(
"{key} has {held} request(s) in flight; its {class:?} share is {share}"
),
});
}
}
}
Ok(Ticket {
door: Arc::clone(self),
key,
work,
class,
guard: Some(guard),
actual: Arc::new(Mutex::new(CostTally::default())),
counted: self.cfg.is_some(),
})
}
fn leave(&self, key: &str) {
let left = {
let mut shard = self.in_flight.shard(key);
match shard.get_mut(key) {
Some(n) if *n > 1 => {
*n -= 1;
*n
}
_ => {
shard.remove(key);
0
}
}
};
self.total.fetch_sub(1, Ordering::AcqRel);
book().set_in_flight(key, left);
}
}
/// One work kind's learnt estimate (`Admit::estimates`), as the admin door prints it.
#[derive(Clone, Debug, PartialEq, Eq, Serialize, Deserialize)]
pub struct EstimateView {
pub kind: String,
pub samples: u64,
pub hinted: u64,
pub mean: CostTally,
pub per_gb: Option<CostTally>,
}
/// `admit plan` for one consumer, as the admin door prints it.
#[derive(Clone, Debug, PartialEq, Eq, Serialize, Deserialize)]
pub struct PlanView {
pub consumer: String,
pub work: String,
pub class: ClassWord,
pub estimate: CostTally,
pub why: String,
/// Bytes the policy counts as used in its window (`Admit::usage`).
pub used: u64,
/// Bytes left in the window (`None` = unlimited).
pub budget_left: Option<u64>,
/// The concurrent-request share this class gets per consumer (`None` =
/// no share enforced: no `admission:` block).
pub share: Option<u32>,
/// `pass-through` | `usage-share`.
pub policy: String,
}
/// **One admitted request.** Drop it when the response is done: that is the
/// `Admit::record` and the release of the share.
pub struct Ticket {
door: Arc<DoorAdmission>,
key: String,
work: WorkKind,
class: ClassWord,
guard: Option<admit::Guard>,
/// This request's own cost — its own lock, never shared with another request.
actual: Arc<Mutex<CostTally>>,
counted: bool,
}
impl Ticket {
pub fn consumer(&self) -> &str {
&self.key
}
pub fn class(&self) -> ClassWord {
self.class
}
pub fn add_net_out(&self, bytes: u64) {
self.actual.lock_or_recover().net_out += bytes;
}
pub fn add_net_in(&self, bytes: u64) {
self.actual.lock_or_recover().net_in += bytes;
}
/// **Run a backend walk for this request**: the upstream bytes a proxy takes
/// are charged to this ticket, and — with the queued pull-through configured
/// — a primary miss becomes a durable `Job::Fetch` at this class's priority.
pub fn walk<R>(&self, f: impl FnOnce() -> R) -> R {
crate::pull_through::charged_ticket(self, || {
#[cfg(feature = "jobs")]
if let Some(pt) = self.door.pull_through() {
return crate::pull_through::queued(pt, self.class.priority(), f);
}
f()
})
}
/// The sink `crate::pull_through` charges upstream bytes into while a walk
/// runs for this ticket, so they reach `Admit::record` with the rest.
pub fn sink(&self) -> Arc<Mutex<CostTally>> {
Arc::clone(&self.actual)
}
}
impl Drop for Ticket {
fn drop(&mut self) {
let actual = *self.actual.lock_or_recover();
if let Some(g) = self.guard.take() {
self.door.admit.record(g, &actual.to_cost());
}
// The upstream half is booked under `pull_through`, the rest under the
// request's own kind, so the admin view can tell them apart.
let mut own = actual;
let upstream = std::mem::take(&mut own.upstream_fetch);
if own != CostTally::default() {
book().charge(&self.key, self.work, &own);
}
if upstream > 0 {
book().charge(
&self.key,
WorkKind::PullThrough,
&CostTally {
upstream_fetch: upstream,
..Default::default()
},
);
}
if self.counted {
self.door.leave(&self.key);
}
}
}
impl std::fmt::Debug for Ticket {
fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
f.debug_struct("Ticket")
.field("consumer", &self.key)
.field("class", &self.class)
.finish()
}
}
#[cfg(test)]
mod tests {
use super::*;
use std::sync::atomic::AtomicUsize;
use std::time::{Duration, Instant};
/// A policy that plans by consumer kind: named `Full`, anonymous `Floor`.
struct ByKind;
impl admit::Admit for ByKind {
fn plan(
&self,
work: &admit::Work,
who: &admit::Consumer,
usage: &admit::Usage,
) -> admit::Plan {
let class = if who.is_anonymous() {
admit::Class::Floor
} else {
admit::Class::Full
};
admit::Plan {
class,
estimate: admit::Cost::default(),
why: format!("{class:?} by kind"),
budget_left: admit::Usage::unlimited(usage.window_days),
work: *work,
who: who.clone(),
}
}
fn admit(&self, plan: &admit::Plan) -> Result<admit::Guard, admit::Refusal> {
Ok(admit::Guard::unslotted(plan))
}
fn record(&self, _: admit::Guard, _: &admit::Cost) {}
fn policy(&self) -> &'static str {
"by-kind"
}
}
fn shares(full: u32, floor: u32, max: u32) -> AdmissionConfig {
AdmissionConfig {
ron_share_full: full,
ron_share_reduced: floor,
ron_share_floor: floor,
ron_max_in_flight: max,
ron_retry_after_secs: 3,
..AdmissionConfig::default()
}
}
fn anon(door: &DoorAdmission, ip: &str) -> Who {
let ip: std::net::IpAddr = ip.parse().unwrap();
Who::Anonymous(door.day_token(crate::visitors::Client::Ip(ip)))
}
/// § PassThrough reproduces today's behaviour: no share, nothing refused,
/// however many requests one consumer holds at once.
#[test]
fn pass_through_refuses_nothing() {
let door = DoorAdmission::pass_through();
let who = Who::Principal("pt-everyone".into());
let held: Vec<Ticket> = (0..2_000)
.map(|_| {
door.enter(&who, WorkKind::Download, 0)
.expect("never refused")
})
.collect();
assert_eq!(door.in_flight_total(), 0, "pass-through counts no share");
assert!(held.iter().all(|t| t.class() == ClassWord::Full));
}
/// § A Full consumer gets its share; past it, a fast 503 with the
/// configured Retry-After; another consumer is untouched; a freed slot is
/// taken again.
#[test]
fn a_consumer_past_its_share_gets_a_fast_503() {
let door = Arc::new(DoorAdmission::with(
Arc::new(ByKind),
Some(shares(4, 1, 100)),
));
let alice = Who::Principal("share-alice".into());
let mut held: Vec<Ticket> = (0..4)
.map(|_| door.enter(&alice, WorkKind::Download, 0).unwrap())
.collect();
let t0 = Instant::now();
let busy = door.enter(&alice, WorkKind::Download, 0).unwrap_err();
assert!(
t0.elapsed() < Duration::from_millis(50),
"refused at once, not parked"
);
assert_eq!(busy.retry_after_secs, 3);
assert!(busy.why.contains("share is 4"), "{}", busy.why);
let bob = Who::Principal("share-bob".into());
let _b = door
.enter(&bob, WorkKind::Download, 0)
.expect("bob has his own share");
held.pop();
door.enter(&alice, WorkKind::Download, 0)
.expect("a freed slot is taken again");
let v = book().view(&alice.key()).unwrap();
assert!(v.by_work["download"].refused >= 1);
}
/// § Floor (anonymous) gets a SMALLER share than Full.
#[test]
fn floor_gets_a_smaller_share_than_full() {
let door = Arc::new(DoorAdmission::with(
Arc::new(ByKind),
Some(shares(8, 2, 100)),
));
let stranger = anon(&door, "203.0.113.50");
let a: Vec<_> = (0..2)
.map(|_| door.enter(&stranger, WorkKind::Download, 0).unwrap())
.collect();
assert_eq!(a[0].class(), ClassWord::Floor);
assert!(
door.enter(&stranger, WorkKind::Download, 0).is_err(),
"Floor's share is 2"
);
let named = Who::Principal("floor-vs-full".into());
let b: Vec<_> = (0..8)
.map(|_| door.enter(&named, WorkKind::Download, 0).unwrap())
.collect();
assert_eq!(b.len(), 8, "Full's share is 8");
}
/// § DOWNLOAD ADMISSION SHARES UNDER LOAD: 32 threads, 4 consumers (one
/// anonymous), hammering enter/hold/drop. No consumer ever holds more than
/// its share, the door never more than its ceiling, and the over-share
/// requests were refused rather than queued.
#[test]
fn shares_hold_under_load() {
let door = Arc::new(DoorAdmission::with(
Arc::new(ByKind),
Some(shares(6, 2, 16)),
));
let who = [
Who::Principal("load-a".into()),
Who::Principal("load-b".into()),
Who::Peer("load-peer".into()),
anon(&door, "198.51.100.77"),
];
let live: Vec<AtomicUsize> = (0..4).map(|_| AtomicUsize::new(0)).collect();
let peak: Vec<AtomicUsize> = (0..4).map(|_| AtomicUsize::new(0)).collect();
let door_peak = AtomicUsize::new(0);
let (ok, refused) = (AtomicUsize::new(0), AtomicUsize::new(0));
std::thread::scope(|s| {
for t in 0..32 {
let (door, who, live, peak, door_peak, ok, refused) =
(&door, &who, &live, &peak, &door_peak, &ok, &refused);
s.spawn(move || {
for i in 0..400 {
let c = (t + i) % 4;
match door.enter(&who[c], WorkKind::Download, 0) {
Ok(ticket) => {
let n = live[c].fetch_add(1, Ordering::SeqCst) + 1;
peak[c].fetch_max(n, Ordering::SeqCst);
door_peak
.fetch_max(door.in_flight_total() as usize, Ordering::SeqCst);
ok.fetch_add(1, Ordering::SeqCst);
std::thread::yield_now();
live[c].fetch_sub(1, Ordering::SeqCst);
drop(ticket);
}
Err(_) => {
refused.fetch_add(1, Ordering::SeqCst);
}
}
}
});
}
});
for p in peak.iter().take(3) {
assert!(
p.load(Ordering::SeqCst) <= 6,
"a Full consumer held {}",
p.load(Ordering::SeqCst)
);
}
assert!(
peak[3].load(Ordering::SeqCst) <= 2,
"the anonymous one held {}",
peak[3].load(Ordering::SeqCst)
);
assert!(
door_peak.load(Ordering::SeqCst) <= 16,
"the door held {}",
door_peak.load(Ordering::SeqCst)
);
assert!(ok.load(Ordering::SeqCst) > 0 && refused.load(Ordering::SeqCst) > 0);
assert_eq!(door.in_flight_total(), 0, "every share was released");
eprintln!(
"shares_hold_under_load: admitted {} refused {} peak per consumer {:?} door peak {}",
ok.load(Ordering::SeqCst),
refused.load(Ordering::SeqCst),
peak.iter()
.map(|p| p.load(Ordering::SeqCst))
.collect::<Vec<_>>(),
door_peak.load(Ordering::SeqCst)
);
}
/// § The book survives a restart: flushed to a redb job ledger, loaded by a
/// fresh book, and `UsageShare` would be seeded with the same days.
#[test]
fn the_book_flushes_to_redb_and_loads_back() {
let dir = tempfile::tempdir().unwrap();
let store = jera::jobs::JobStore::open(dir.path()).unwrap();
let first = CostBook::default();
first.charge(
"principal:durable-dora",
WorkKind::Download,
&CostTally {
net_out: 1_000,
..Default::default()
},
);
first.charge(
"principal:durable-dora",
WorkKind::PullThrough,
&CostTally {
upstream_fetch: 500,
..Default::default()
},
);
first.charge(
"anon:1:deadbeef",
WorkKind::Download,
&CostTally {
net_out: 7,
..Default::default()
},
);
assert_eq!(
first.flush(&store).unwrap(),
1,
"the named consumer, not the day token"
);
assert_eq!(first.flush(&store).unwrap(), 0, "nothing dirty twice");
let second = CostBook::default();
assert_eq!(second.load(&store).unwrap(), 1);
assert_eq!(second.usage_of("principal:durable-dora").0.bytes(), 1_500);
let d = today();
assert_eq!(
second.seed_days("principal:durable-dora", d.saturating_sub(29), d),
vec![(d, 1_500)]
);
}
/// § With `UsageShare`, a named consumer past its window budget is planned
/// `Reduced` and gets the smaller share; its usage came from the seed.
#[test]
fn usage_share_reduces_a_consumer_past_its_budget() {
let heavy = "principal:usage-share-heavy";
book().charge(
heavy,
WorkKind::Download,
&CostTally {
net_out: 2_000,
..Default::default()
},
);
let seed = |who: &admit::Consumer, first: u64, today: u64| -> Vec<(u64, u64)> {
Who::of_consumer(who)
.map(|w| book().seed_days(&w.key(), first, today))
.unwrap_or_default()
};
let share = admit::Settings {
limit: Some(1_000),
slots: None,
..admit::Settings::default()
}
.build_seeded(Some(Arc::new(seed)));
let door = Arc::new(DoorAdmission::with(share, Some(shares(4, 1, 100))));
let who = Who::Principal("usage-share-heavy".into());
let t = door.enter(&who, WorkKind::Download, 0).unwrap();
assert_eq!(
t.class(),
ClassWord::Reduced,
"2 000 B used of a 1 000 B window"
);
assert!(
door.enter(&who, WorkKind::Download, 0).is_err(),
"Reduced's share is 1"
);
let light = Who::Principal("usage-share-light".into());
assert_eq!(
door.enter(&light, WorkKind::Download, 0).unwrap().class(),
ClassWord::Full
);
}
fn pct(sorted: &[u64], p: f64) -> u64 {
sorted[((sorted.len() as f64 - 1.0) * p).round() as usize]
}
/// Run `threads × ops` enter+drop on `door`, each thread its own consumer;
/// `(p50 ns, p99 ns, ops/s)`.
fn hammer(door: &Arc<DoorAdmission>, threads: usize, ops: usize) -> (u64, u64, f64) {
let t0 = Instant::now();
let mut all: Vec<u64> = std::thread::scope(|s| {
let hs: Vec<_> = (0..threads)
.map(|t| {
let door = Arc::clone(door);
s.spawn(move || {
let who = Who::Principal(format!("bench-{t}"));
let mut lat = Vec::with_capacity(ops);
for _ in 0..ops {
let a = Instant::now();
let ticket = door
.enter(&who, WorkKind::Download, 4096)
.expect("admitted");
ticket.add_net_out(4096);
drop(ticket);
lat.push(a.elapsed().as_nanos() as u64);
}
lat
})
})
.collect();
hs.into_iter().flat_map(|h| h.join().unwrap()).collect()
});
let secs = t0.elapsed().as_secs_f64();
all.sort_unstable();
(
pct(&all, 0.50),
pct(&all, 0.99),
(threads * ops) as f64 / secs,
)
}
/// § THE LOAD TEST (owner, 2026-09-28): many concurrent requests across N
/// threads through the admission layer — enter (usage → plan → admit →
/// share) and drop (record → book → release) — p50/p99 latency and
/// throughput, PassThrough against UsageShare. Printed for the report;
/// asserted only loosely (a debug build on a loaded box is not a bench),
/// and that nothing is refused or leaked.
#[test]
fn admission_layer_load_p50_p99_pass_through_vs_usage_share() {
let threads = std::thread::available_parallelism()
.map(|n| n.get())
.unwrap_or(4)
.clamp(2, 16);
let ops = 20_000;
let big = shares(1_000, 1_000, 100_000);
let pass = Arc::new(DoorAdmission::with(
Arc::new(admit::PassThrough::new()),
Some(big.clone()),
));
let share = Arc::new(DoorAdmission::with(
admit::Settings {
slots: None,
..admit::Settings::default()
}
.build(),
Some(big),
));
let (p50a, p99a, tpa) = hammer(&pass, threads, ops);
let (p50b, p99b, tpb) = hammer(&share, threads, ops);
eprintln!(
"ADMISSION LOAD ({threads} threads x {ops} ops, enter+record, {} build): \
PassThrough p50 {p50a} ns p99 {p99a} ns {tpa:.0} ops/s | \
UsageShare p50 {p50b} ns p99 {p99b} ns {tpb:.0} ops/s",
if cfg!(debug_assertions) {
"debug"
} else {
"release"
}
);
assert_eq!(pass.in_flight_total(), 0);
assert_eq!(share.in_flight_total(), 0);
assert!(
tpa > 1_000.0 && tpb > 1_000.0,
"throughput collapsed: {tpa} / {tpb}"
);
}
/// ★ **Every consumer is counted** (lane A13): a named principal or peer once
/// however many requests it makes, while it has a request or a charge in the
/// window — the charges reload from the ledger, so a restart keeps them — and
/// an anonymous client by its day token, today only.
#[test]
fn every_consumer_is_counted_once_named_in_the_window_and_anonymous_today() {
let b = CostBook::default();
let d = today();
for _ in 0..3 {
b.note_plan(
"principal:alice",
WorkKind::Download,
ClassWord::Full,
"",
&CostTally::default(),
);
}
b.note_plan(
"principal:bob",
WorkKind::Upload,
ClassWord::Full,
"",
&CostTally::default(),
);
b.note_plan(
"peer:onprem-1",
WorkKind::Download,
ClassWord::Full,
"",
&CostTally::default(),
);
b.note_plan(
&format!("anon:{d}:aa"),
WorkKind::Download,
ClassWord::Floor,
"",
&CostTally::default(),
);
b.note_plan(
&format!("anon:{d}:bb"),
WorkKind::Download,
ClassWord::Floor,
"",
&CostTally::default(),
);
b.note_plan(
&format!("anon:{}:cc", d - 1),
WorkKind::Download,
ClassWord::Floor,
"",
&CostTally::default(),
);
// Charged 40 days ago only: out of the window.
b.charge_on(
"principal:gone",
WorkKind::Download,
&CostTally {
net_out: 1,
..Default::default()
},
d - 40,
);
let c = b.consumers_on(d);
assert_eq!(
(
c.counted,
c.principals,
c.peers,
c.anonymous_today,
c.window_days
),
(true, 2, 1, 2, 30),
"{c:?}"
);
// A restart: the ledger brings the charged named consumers back, never a day token.
let dir = tempfile::tempdir().unwrap();
let store = jera::jobs::JobStore::open(dir.path()).unwrap();
b.charge(
"principal:alice",
WorkKind::Download,
&CostTally {
net_out: 10,
..Default::default()
},
);
b.charge(
&format!("anon:{d}:aa"),
WorkKind::Download,
&CostTally {
net_out: 10,
..Default::default()
},
);
b.flush(&store).unwrap();
let after = CostBook::default();
after.load(&store).unwrap();
let c = after.consumers_on(d);
assert_eq!(
(c.principals, c.peers, c.anonymous_today),
(1, 0, 0),
"alice was charged and is kept; bob only asked; tokens die: {c:?}"
);
}
}