use std::collections::HashMap;
use std::sync::atomic::{AtomicU64, Ordering};
use std::sync::{Arc, Mutex};
use std::time::Duration;
use crate::governor::{
Claim, Governor, Growth, Lane, LimitBand, OpClass, ServiceMark, Shed, DEFAULT_ADMISSION_WAIT,
DEFAULT_MEMORY_BUDGET_BYTES, WIRE_MEMORY_DENOMINATOR, WIRE_MEMORY_NUMERATOR, WIRE_RESERVE_STEP,
};
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub enum Refusal {
AtCapacity {
limit: usize,
},
PrincipalAtCapacity {
limit: usize,
active_principals: usize,
},
LaneAtCapacity {
lane: Lane,
ceiling: usize,
limit: usize,
},
AnonymousAtCapacity {
limit: usize,
},
KeyedReserve {
occupied: usize,
limit: usize,
reserve: usize,
},
MemoryBudget {
class: OpClass,
wanted: u64,
reserved: u64,
budget: u64,
},
Paused {
pause: Pause,
},
#[cfg(feature = "admit-share")]
ReducedShare {
ceiling: usize,
busy: bool,
},
}
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub enum Pause {
Growing,
}
impl Pause {
pub const fn words(self) -> &'static str {
match self {
Pause::Growing => GROWING_MESSAGE,
}
}
pub const fn reason(self) -> &'static str {
match self {
Pause::Growing => "paused-growing",
}
}
}
pub const GROWING_MESSAGE: &str = "The server is growing its storage to make room for new \
members. Pushes are paused — please try again in a few \
minutes.";
impl Refusal {
pub fn client_message(&self) -> String {
match self {
Refusal::AtCapacity { limit } => format!(
"the server is already serving its limit of {limit} concurrent git transfers; \
try again in a moment"
),
Refusal::PrincipalAtCapacity {
limit,
active_principals,
} if *active_principals > 1 => format!(
"you already have {limit} git transfers in flight, which is your fair share of \
this server while {active_principals} identities are using it; that is the \
per-principal limit in force right now, and it rises again as the server \
empties; try again in a moment"
),
Refusal::PrincipalAtCapacity { limit, .. } => format!(
"you already have {limit} git transfers in flight on this server, which is the \
per-principal limit; try again in a moment"
),
Refusal::LaneAtCapacity {
lane,
ceiling,
limit,
} => format!(
"this server is holding {lane} operations to {ceiling} of its {limit} slots so \
that the other kinds of request still land; try again in a moment"
),
Refusal::AnonymousAtCapacity { limit } => format!(
"this server is already serving {limit} unauthenticated readers, which is all it \
offers without a key; try again in a moment, or push a key and clone as \
yourself"
),
Refusal::KeyedReserve {
occupied,
limit,
reserve,
} => format!(
"this server is {occupied} of {limit} slots busy and keeps its last {reserve} for \
callers with a key; try again in a moment, or push a key and clone as yourself"
),
Refusal::MemoryBudget {
class,
wanted,
reserved,
budget,
} => format!(
"this server holds a memory budget of {} MiB for git operations in flight, {} MiB \
of it is reserved, and a {class} needs {} MiB more than is left; try again in a \
moment",
budget / (1024 * 1024),
reserved / (1024 * 1024),
wanted / (1024 * 1024),
),
Refusal::Paused { pause } => pause.words().to_owned(),
#[cfg(feature = "admit-share")]
Refusal::ReducedShare { ceiling, busy } => {
format!(
"this namespace is over its transfer budget for the window, so it is served at \
the reduced share of {ceiling} git transfers at once{}; try again in a moment, \
or ask for a larger budget",
if *busy { ", and none while the server is this busy" } else { "" }
)
}
}
}
pub const fn reason(&self) -> &'static str {
match self {
Refusal::AtCapacity { .. } => "admission-gate-full",
Refusal::LaneAtCapacity { .. } => "admission-lane-full",
Refusal::PrincipalAtCapacity { .. } => "admission-principal-share",
Refusal::AnonymousAtCapacity { .. } => "admission-anonymous-ceiling",
Refusal::KeyedReserve { .. } => "admission-keyed-reserve",
Refusal::MemoryBudget { .. } => "admission-memory-budget",
Refusal::Paused { pause } => pause.reason(),
#[cfg(feature = "admit-share")]
Refusal::ReducedShare { .. } => "admission-reduced-share",
}
}
}
impl std::fmt::Display for Refusal {
fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
f.write_str(&self.client_message())
}
}
#[derive(Debug, Clone, Copy, PartialEq, Eq, Default)]
pub struct AdmissionStats {
pub admitted: u64,
pub refused_at_capacity: u64,
pub refused_per_principal: u64,
pub refused_memory: u64,
pub refused_anonymous: u64,
pub refused_keyed_reserve: u64,
#[cfg(feature = "admit-share")]
pub refused_reduced: u64,
pub refused_paused: u64,
pub in_flight: u64,
pub peak_in_flight: u64,
pub limit: usize,
pub reserved_bytes: u64,
pub admitted_after_wait: u64,
pub waited_ms_total: u64,
pub peak_wait_ms: u64,
pub refused_after_wait: u64,
pub refused_lane: u64,
pub overdrawn: u64,
pub grown_bytes: u64,
pub handoff_ms_total: u64,
pub peak_handoff_ms: u64,
pub unmeasured_sojourn: u64,
}
impl AdmissionStats {
pub fn refused(&self) -> u64 {
self.refused_at_capacity
+ self.refused_lane
+ self.refused_per_principal
+ self.refused_memory
+ self.refused_anonymous
+ self.refused_keyed_reserve
+ self.refused_paused
+ self.refused_reduced()
}
#[cfg(feature = "admit-share")]
fn refused_reduced(&self) -> u64 {
self.refused_reduced
}
#[cfg(not(feature = "admit-share"))]
fn refused_reduced(&self) -> u64 {
0
}
pub fn mean_wait_ms(&self) -> u64 {
self.waited_ms_total
.checked_div(self.admitted_after_wait.max(1))
.unwrap_or(0)
}
}
#[derive(Debug, Default)]
struct Counters {
admitted: AtomicU64,
refused_at_capacity: AtomicU64,
refused_lane: AtomicU64,
refused_per_principal: AtomicU64,
refused_memory: AtomicU64,
refused_anonymous: AtomicU64,
refused_keyed_reserve: AtomicU64,
refused_paused: AtomicU64,
#[cfg(feature = "admit-share")]
refused_reduced: AtomicU64,
in_flight: AtomicU64,
peak_in_flight: AtomicU64,
unmeasured_sojourn: AtomicU64,
admitted_after_wait: AtomicU64,
waited_ms_total: AtomicU64,
peak_wait_ms: AtomicU64,
refused_after_wait: AtomicU64,
overdrawn: AtomicU64,
grown_bytes: AtomicU64,
handoff_ms_total: AtomicU64,
peak_handoff_ms: AtomicU64,
}
pub const CONTENTION_PERCENT: usize = 75;
pub const DEFAULT_MAX_ANONYMOUS_TRANSFERS: usize = 5;
const RETRY_INTERVAL: Duration = crate::governor::INTERVAL;
const WAIT_ROW_INTERVAL_MS: u64 = 1000;
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
enum Count {
Yes,
No,
}
impl Count {
fn bump(self, counter: &AtomicU64) {
if self == Count::Yes {
counter.fetch_add(1, Ordering::Relaxed);
}
}
}
#[derive(Debug)]
pub struct Admission {
governor: Option<Governor>,
max_concurrent: Option<usize>,
max_per_principal: Option<usize>,
hard_per_principal: bool,
anonymous_ceiling: Option<usize>,
per_principal: Mutex<HashMap<String, usize>>,
counters: Arc<Counters>,
slot_freed: tokio::sync::Notify,
wait: Duration,
keyed_wait: Option<Duration>,
epoch: std::time::Instant,
next_wait_row_ms: AtomicU64,
pause: std::sync::atomic::AtomicU8,
flow: Option<Arc<gunnar_core::FlowLedger>>,
#[cfg(feature = "admit-share")]
policy: Policy,
}
#[cfg(feature = "admit-share")]
#[derive(Clone)]
struct Policy(Arc<dyn edda_admit::Admit>);
#[cfg(feature = "admit-share")]
impl std::fmt::Debug for Policy {
fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
f.write_str(self.0.policy())
}
}
pub fn work_kind(class: OpClass) -> edda_admit::WorkKind {
match class {
OpClass::Clone => edda_admit::WorkKind::Clone,
OpClass::Fetch => edda_admit::WorkKind::Fetch,
OpClass::Push => edda_admit::WorkKind::Push,
OpClass::LsRemote | OpClass::Gc => edda_admit::WorkKind::Other,
}
}
impl Admission {
pub fn new(max_concurrent: Option<usize>, max_per_principal: Option<usize>) -> Admission {
Admission::with_memory_budget(
max_concurrent,
max_per_principal,
DEFAULT_MEMORY_BUDGET_BYTES,
)
}
pub fn with_flow(mut self, flow: Arc<gunnar_core::FlowLedger>) -> Admission {
self.flow = Some(flow);
self
}
pub fn flow(&self) -> Option<&Arc<gunnar_core::FlowLedger>> {
self.flow.as_ref()
}
pub fn with_memory_budget(
max_concurrent: Option<usize>,
max_per_principal: Option<usize>,
memory_budget: u64,
) -> Admission {
Admission {
governor: max_concurrent
.map(|budget| Governor::new(LimitBand::for_budget(budget), memory_budget)),
max_concurrent,
max_per_principal,
hard_per_principal: false,
anonymous_ceiling: Some(DEFAULT_MAX_ANONYMOUS_TRANSFERS),
per_principal: Mutex::new(HashMap::new()),
counters: Arc::new(Counters::default()),
slot_freed: tokio::sync::Notify::new(),
wait: DEFAULT_ADMISSION_WAIT,
keyed_wait: None,
epoch: std::time::Instant::now(),
next_wait_row_ms: AtomicU64::new(0),
pause: std::sync::atomic::AtomicU8::new(0),
flow: None,
#[cfg(feature = "admit-share")]
policy: Policy(Arc::new(edda_admit::PassThrough::new())),
}
}
#[cfg(feature = "admit-share")]
pub fn with_policy(mut self, policy: Arc<dyn edda_admit::Admit>) -> Admission {
self.policy = Policy(policy);
self
}
pub fn admit_policy(&self) -> Arc<dyn edda_admit::Admit> {
#[cfg(feature = "admit-share")]
{
Arc::clone(&self.policy.0)
}
#[cfg(not(feature = "admit-share"))]
{
Arc::new(edda_admit::PassThrough::new())
}
}
pub fn admit_usage(
&self,
policy: &dyn edda_admit::Admit,
namespace: &str,
) -> edda_admit::Usage {
if policy.limit().is_some() {
return policy.usage(&edda_admit::Consumer::Namespace(namespace.to_owned()));
}
let window = edda_admit::DEFAULT_WINDOW_DAYS;
let bytes = match &self.flow {
Some(flow) if !namespace.is_empty() => {
flow.window_bytes(namespace, gunnar_core::now_ms(), window)
}
_ => 0,
};
edda_admit::Usage::new(bytes, window)
}
pub fn flow_seed(flow: Arc<gunnar_core::FlowLedger>) -> Arc<dyn edda_admit::Seed> {
Arc::new(
move |who: &edda_admit::Consumer, first_day: u64, today: u64| {
let edda_admit::Consumer::Namespace(namespace) = who else {
return Vec::new();
};
let days = u32::try_from(today.saturating_sub(first_day) + 1).unwrap_or(u32::MAX);
let now = gunnar_core::now_ms();
flow.moved(namespace, now, days)
.into_iter()
.filter_map(|(day, bytes)| u64::try_from(day).ok().map(|d| (d, bytes)))
.collect()
},
)
}
#[cfg(feature = "admit-share")]
fn plan(&self, principal: &str, namespace: &str, class: OpClass) -> Option<edda_admit::Plan> {
if principal == gunnar_core::ANONYMOUS_PRINCIPAL_ID {
return None;
}
let policy = &*self.policy.0;
let who = edda_admit::Consumer::Namespace(namespace.to_owned());
let usage = policy.usage(&who);
Some(policy.plan(&edda_admit::Work::new(work_kind(class), None), &who, &usage))
}
pub fn hard_capped(
max_concurrent: Option<usize>,
max_per_principal: Option<usize>,
) -> Admission {
Admission {
hard_per_principal: true,
..Admission::new(max_concurrent, max_per_principal)
}
}
pub fn with_budget(self, memory_budget: u64) -> Admission {
Admission {
governor: self
.max_concurrent
.map(|budget| Governor::new(LimitBand::for_budget(budget), memory_budget)),
..self
}
}
pub fn unlimited() -> Admission {
Admission {
anonymous_ceiling: None,
..Admission::new(None, None)
}
}
pub fn with_anonymous_ceiling(self, anonymous_ceiling: Option<usize>) -> Admission {
Admission {
anonymous_ceiling,
..self
}
}
pub fn max_anonymous(&self) -> Option<usize> {
self.anonymous_ceiling
}
pub fn keyed_reserve(&self) -> usize {
let Some(governor) = &self.governor else {
return 0;
};
let limit = governor.limit();
limit - limit.saturating_mul(CONTENTION_PERCENT).div_ceil(100) + 1
}
pub fn per_principal_is_hard(&self) -> bool {
self.hard_per_principal
}
pub fn max_concurrent(&self) -> Option<usize> {
self.max_concurrent
}
pub fn max_per_principal(&self) -> Option<usize> {
self.max_per_principal
}
pub fn governed_limit(&self) -> Option<usize> {
self.governor.as_ref().map(Governor::limit)
}
pub fn memory_budget(&self) -> Option<u64> {
self.governor.as_ref().map(|g| g.memory().budget())
}
pub fn governor(&self) -> Option<&Governor> {
self.governor.as_ref()
}
pub fn try_admit(
self: &Arc<Self>,
principal: &str,
namespace: &str,
class: OpClass,
) -> Result<AdmissionGuard, Refusal> {
self.attempt(principal, namespace, class, Count::Yes)
}
fn attempt(
self: &Arc<Self>,
principal: &str,
namespace: &str,
class: OpClass,
counting: Count,
) -> Result<AdmissionGuard, Refusal> {
let outcome = self.attempt_inner(principal, namespace, class, counting);
if counting == Count::Yes && outcome.is_err() {
self.charge_flow(namespace);
}
outcome
}
fn charge_flow(&self, namespace: &str) {
if let Some(flow) = &self.flow {
flow.refuse(
namespace,
gunnar_core::now_ms(),
gunnar_core::Refusal::Admission,
);
}
}
fn attempt_inner(
self: &Arc<Self>,
principal: &str,
namespace: &str,
class: OpClass,
counting: Count,
) -> Result<AdmissionGuard, Refusal> {
if let Some(pause) = self.paused() {
counting.bump(&self.counters.refused_paused);
return Err(Refusal::Paused { pause });
}
#[cfg(feature = "admit-share")]
let plan = self.plan(principal, namespace, class);
#[cfg(feature = "admit-share")]
let reduced = plan.as_ref().is_some_and(|p| {
matches!(
p.class,
edda_admit::Class::Reduced | edda_admit::Class::Floor
)
});
#[cfg(not(feature = "admit-share"))]
let (reduced, _) = (false, namespace);
let anonymous_ceiling = self
.anonymous_ceiling
.filter(|_| principal == gunnar_core::ANONYMOUS_PRINCIPAL_ID || reduced);
let reserve = match anonymous_ceiling {
Some(cap) if cap > 0 => self.keyed_reserve(),
_ => 0,
};
let counts = counting == Count::Yes;
let claim = match &self.governor {
Some(governor) => match governor.claim_reserving(class, reserve, counts) {
Ok(claim) => Some(claim),
Err(Shed::Reserved {
occupied,
limit,
reserve,
}) => {
#[cfg(feature = "admit-share")]
if reduced {
counting.bump(&self.counters.refused_reduced);
self.warn_reduced(plan.as_ref(), namespace);
return Err(Refusal::ReducedShare {
ceiling: anonymous_ceiling.unwrap_or(0),
busy: true,
});
}
counting.bump(&self.counters.refused_keyed_reserve);
return Err(Refusal::KeyedReserve {
occupied,
limit,
reserve,
});
}
Err(Shed::Concurrency { limit, .. }) => {
counting.bump(&self.counters.refused_at_capacity);
return Err(Refusal::AtCapacity { limit });
}
Err(Shed::LaneFull {
lane,
ceiling,
limit,
}) => {
counting.bump(&self.counters.refused_lane);
return Err(Refusal::LaneAtCapacity {
lane,
ceiling,
limit,
});
}
Err(Shed::Memory {
class,
wanted,
reserved,
budget,
}) => {
counting.bump(&self.counters.refused_memory);
return Err(Refusal::MemoryBudget {
class,
wanted,
reserved,
budget,
});
}
},
None => None,
};
if self.max_per_principal.is_some() || anonymous_ceiling.is_some() {
let mut held = self.per_principal.lock().unwrap_or_else(|e| e.into_inner());
let already_active = held.contains_key(principal);
let active = held.len() + usize::from(!already_active);
let count = held.entry(principal.to_string()).or_insert(0);
let (ceiling, anonymous) = match anonymous_ceiling {
Some(cap) => (cap, true),
None => (
self.ceiling_for(self.max_per_principal.unwrap_or(usize::MAX), active),
false,
),
};
if *count >= ceiling {
if *count == 0 {
held.remove(principal);
}
drop(held);
if let (Some(governor), Some(claim)) = (&self.governor, &claim) {
governor.release(claim, 0);
}
#[cfg(feature = "admit-share")]
if anonymous && reduced {
counting.bump(&self.counters.refused_reduced);
self.warn_reduced(plan.as_ref(), namespace);
return Err(Refusal::ReducedShare {
ceiling,
busy: false,
});
}
if anonymous {
counting.bump(&self.counters.refused_anonymous);
return Err(Refusal::AnonymousAtCapacity { limit: ceiling });
}
counting.bump(&self.counters.refused_per_principal);
return Err(Refusal::PrincipalAtCapacity {
limit: ceiling,
active_principals: active,
});
}
*count += 1;
}
self.counters.admitted.fetch_add(1, Ordering::Relaxed);
let in_flight = self.counters.in_flight.fetch_add(1, Ordering::Relaxed) + 1;
self.counters
.peak_in_flight
.fetch_max(in_flight, Ordering::Relaxed);
#[cfg(feature = "admit-share")]
let admitted = plan.and_then(|plan| self.policy.0.admit(&plan).ok());
Ok(AdmissionGuard {
#[cfg(feature = "admit-share")]
admitted,
gate: Arc::clone(self),
principal: (self.max_per_principal.is_some() || anonymous_ceiling.is_some())
.then(|| principal.to_string()),
claim,
marked: ServiceMark::default(),
wire: AtomicU64::new(0),
grown: AtomicU64::new(0),
})
}
#[cfg(feature = "admit-share")]
fn warn_reduced(&self, plan: Option<&edda_admit::Plan>, namespace: &str) {
tracing::warn!(
target: "gunnar::git",
event = "git.admission.reduced",
namespace,
why = plan.map_or("", |p| p.why.as_str()),
"a namespace over its transfer budget met a full reduced share"
);
}
pub fn with_wait(self, wait: Duration) -> Admission {
Admission { wait, ..self }
}
pub fn wait(&self) -> Duration {
self.wait
}
pub fn with_keyed_wait(self, keyed_wait: Option<Duration>) -> Admission {
Admission { keyed_wait, ..self }
}
pub fn wait_for(&self, principal: &str) -> Duration {
if self.wait.is_zero() || principal == gunnar_core::ANONYMOUS_PRINCIPAL_ID {
return self.wait;
}
self.keyed_wait.map_or(self.wait, |k| k.max(self.wait))
}
pub async fn admit_within(
self: &Arc<Self>,
principal: &str,
namespace: &str,
class: OpClass,
) -> Result<(AdmissionGuard, Duration), Refusal> {
let started = std::time::Instant::now();
let wait = self.wait_for(principal);
let mut refusal = match self.attempt(principal, namespace, class, Count::No) {
Ok(guard) => {
self.record_wait(class, Duration::ZERO);
return Ok((guard, Duration::ZERO));
}
Err(refusal) => refusal,
};
if wait.is_zero()
|| matches!(
refusal,
Refusal::AnonymousAtCapacity { .. }
| Refusal::KeyedReserve { .. }
| Refusal::Paused { .. }
)
|| is_reduced(&refusal)
{
self.count_refusal(namespace, &refusal);
return Err(refusal);
}
let _parked = Parked::new(self.governor.as_ref(), class.lane());
loop {
let remaining = match wait.checked_sub(started.elapsed()) {
Some(remaining) if !remaining.is_zero() => remaining,
_ => {
self.counters
.refused_after_wait
.fetch_add(1, Ordering::Relaxed);
self.count_refusal(namespace, &refusal);
self.record_wait(class, wait);
return Err(refusal);
}
};
let freed = self.slot_freed.notified();
match self.attempt(principal, namespace, class, Count::No) {
Ok(guard) => {
let waited = started.elapsed();
let ms = u64::try_from(waited.as_millis()).unwrap_or(u64::MAX);
self.counters
.admitted_after_wait
.fetch_add(1, Ordering::Relaxed);
self.counters
.waited_ms_total
.fetch_add(ms, Ordering::Relaxed);
self.counters.peak_wait_ms.fetch_max(ms, Ordering::Relaxed);
self.record_wait(class, waited);
return Ok((guard, waited));
}
Err(
again @ (Refusal::AnonymousAtCapacity { .. } | Refusal::KeyedReserve { .. }),
) => {
self.count_refusal(namespace, &again);
self.record_wait(class, started.elapsed());
return Err(again);
}
Err(again) if is_reduced(&again) => {
self.count_refusal(namespace, &again);
self.record_wait(class, started.elapsed());
return Err(again);
}
Err(again) => refusal = again,
}
let _ = tokio::time::timeout(remaining.min(RETRY_INTERVAL), freed).await;
}
}
fn count_refusal(&self, namespace: &str, refusal: &Refusal) {
self.charge_flow(namespace);
let counter = match refusal {
Refusal::AtCapacity { .. } => &self.counters.refused_at_capacity,
Refusal::LaneAtCapacity { .. } => &self.counters.refused_lane,
Refusal::PrincipalAtCapacity { .. } => &self.counters.refused_per_principal,
Refusal::AnonymousAtCapacity { .. } => &self.counters.refused_anonymous,
Refusal::KeyedReserve { .. } => &self.counters.refused_keyed_reserve,
Refusal::MemoryBudget { .. } => &self.counters.refused_memory,
Refusal::Paused { .. } => &self.counters.refused_paused,
#[cfg(feature = "admit-share")]
Refusal::ReducedShare { .. } => &self.counters.refused_reduced,
};
counter.fetch_add(1, Ordering::Relaxed);
}
pub fn wait_row_due(&self) -> bool {
let now = self.epoch.elapsed().as_millis() as u64;
self.next_wait_row_ms
.fetch_update(Ordering::Relaxed, Ordering::Relaxed, |due| {
(now >= due).then_some(now + WAIT_ROW_INTERVAL_MS)
})
.is_ok()
}
fn record_wait(&self, class: OpClass, waited: Duration) {
if let Some(governor) = &self.governor {
governor.record_sojourn(class, waited);
}
}
fn ceiling_for(&self, configured: usize, active: usize) -> usize {
let (Some(governor), false) = (&self.governor, self.hard_per_principal) else {
return configured;
};
let server_wide = governor.limit();
let occupied = governor.occupied();
let contended =
occupied.saturating_mul(100) >= server_wide.saturating_mul(CONTENTION_PERCENT);
if !contended {
return server_wide;
}
configured.max(server_wide / active.max(1))
}
pub fn pause(&self, pause: Pause) {
self.pause.store(1 + pause as u8, Ordering::SeqCst);
tracing::warn!(
target: "gunnar::limits",
event = "admission.paused",
reason = pause.reason(),
in_flight = self.counters.in_flight.load(Ordering::Relaxed),
"the admission gate is PAUSED: every new git transfer is refused with `{}` until the \
gate is unpaused; transfers already admitted run to completion",
pause.words()
);
}
pub fn unpause(&self) {
if self.pause.swap(0, Ordering::SeqCst) != 0 {
tracing::info!(
target: "gunnar::limits",
event = "admission.unpaused",
refused_while_paused = self.counters.refused_paused.load(Ordering::Relaxed),
"the admission gate is open again"
);
}
}
pub fn note_refused_paused(&self) {
self.counters.refused_paused.fetch_add(1, Ordering::Relaxed);
}
pub fn paused(&self) -> Option<Pause> {
match self.pause.load(Ordering::SeqCst) {
0 => None,
n if n == 1 + Pause::Growing as u8 => Some(Pause::Growing),
_ => None,
}
}
pub fn in_flight(&self) -> u64 {
self.counters.in_flight.load(Ordering::Relaxed)
}
pub fn stats(&self) -> AdmissionStats {
AdmissionStats {
admitted: self.counters.admitted.load(Ordering::Relaxed),
refused_at_capacity: self.counters.refused_at_capacity.load(Ordering::Relaxed),
refused_per_principal: self.counters.refused_per_principal.load(Ordering::Relaxed),
refused_memory: self.counters.refused_memory.load(Ordering::Relaxed),
refused_anonymous: self.counters.refused_anonymous.load(Ordering::Relaxed),
refused_keyed_reserve: self.counters.refused_keyed_reserve.load(Ordering::Relaxed),
refused_paused: self.counters.refused_paused.load(Ordering::Relaxed),
#[cfg(feature = "admit-share")]
refused_reduced: self.counters.refused_reduced.load(Ordering::Relaxed),
in_flight: self.counters.in_flight.load(Ordering::Relaxed),
peak_in_flight: self.counters.peak_in_flight.load(Ordering::Relaxed),
limit: self.governor.as_ref().map_or(0, Governor::limit),
reserved_bytes: self.governor.as_ref().map_or(0, |g| g.memory().reserved()),
unmeasured_sojourn: self.counters.unmeasured_sojourn.load(Ordering::Relaxed),
admitted_after_wait: self.counters.admitted_after_wait.load(Ordering::Relaxed),
waited_ms_total: self.counters.waited_ms_total.load(Ordering::Relaxed),
peak_wait_ms: self.counters.peak_wait_ms.load(Ordering::Relaxed),
refused_after_wait: self.counters.refused_after_wait.load(Ordering::Relaxed),
refused_lane: self.counters.refused_lane.load(Ordering::Relaxed),
overdrawn: self.counters.overdrawn.load(Ordering::Relaxed),
grown_bytes: self.counters.grown_bytes.load(Ordering::Relaxed),
handoff_ms_total: self.counters.handoff_ms_total.load(Ordering::Relaxed),
peak_handoff_ms: self.counters.peak_handoff_ms.load(Ordering::Relaxed),
}
}
fn release(&self, principal: Option<&str>) {
if let Some(principal) = principal {
let mut held = self.per_principal.lock().unwrap_or_else(|e| e.into_inner());
if let Some(count) = held.get_mut(principal) {
*count = count.saturating_sub(1);
if *count == 0 {
held.remove(principal);
}
}
}
self.counters.in_flight.fetch_sub(1, Ordering::Relaxed);
}
}
fn is_reduced(refusal: &Refusal) -> bool {
#[cfg(feature = "admit-share")]
{
matches!(refusal, Refusal::ReducedShare { .. })
}
#[cfg(not(feature = "admit-share"))]
{
let _ = refusal;
false
}
}
#[derive(Debug)]
pub struct AdmissionGuard {
gate: Arc<Admission>,
principal: Option<String>,
claim: Option<Claim>,
marked: ServiceMark,
wire: AtomicU64,
grown: AtomicU64,
#[cfg(feature = "admit-share")]
admitted: Option<edda_admit::Guard>,
}
impl AdmissionGuard {
pub fn entered_service(&self) {
let (Some(_governor), Some(claim)) = (&self.gate.governor, &self.claim) else {
return;
};
if self.marked.claim() {
let handoff = claim.enqueued.elapsed();
let ms = u64::try_from(handoff.as_millis()).unwrap_or(u64::MAX);
self.gate
.counters
.handoff_ms_total
.fetch_add(ms, Ordering::Relaxed);
self.gate
.counters
.peak_handoff_ms
.fetch_max(ms, Ordering::Relaxed);
}
}
pub fn class(&self) -> Option<OpClass> {
self.claim.as_ref().map(|claim| claim.class)
}
pub fn reserve_more(&self, bytes: u64) -> Option<Growth> {
let (Some(governor), Some(_)) = (&self.gate.governor, &self.claim) else {
return None;
};
let wire = self.wire.fetch_add(bytes, Ordering::Relaxed) + bytes;
let want = wire * WIRE_MEMORY_NUMERATOR / WIRE_MEMORY_DENOMINATOR / WIRE_RESERVE_STEP
* WIRE_RESERVE_STEP;
let held = self.grown.load(Ordering::Relaxed);
if want <= held {
return None;
}
let extra = want - held;
if self.grown.fetch_max(want, Ordering::Relaxed) >= want {
return None;
}
self.gate
.counters
.grown_bytes
.fetch_add(extra, Ordering::Relaxed);
let growth = governor.charge(want, extra);
if matches!(growth, Growth::Overdrawn { .. }) {
self.gate.counters.overdrawn.fetch_add(1, Ordering::Relaxed);
}
Some(growth)
}
}
impl Drop for AdmissionGuard {
fn drop(&mut self) {
#[cfg(feature = "admit-share")]
if let Some(admitted) = self.admitted.take() {
let wire = self.wire.load(Ordering::Relaxed);
let push = matches!(self.claim.as_ref().map(|c| c.class), Some(OpClass::Push));
self.gate.policy.0.record(
admitted,
&edda_admit::Cost {
net_in: if push { wire } else { 0 },
net_out: if push { 0 } else { wire },
..edda_admit::Cost::default()
},
);
}
if let (Some(governor), Some(claim)) = (&self.gate.governor, &self.claim) {
governor.release(claim, self.grown.load(Ordering::Relaxed));
if !self.marked.claimed() {
self.gate
.counters
.unmeasured_sojourn
.fetch_add(1, Ordering::Relaxed);
}
}
self.gate.release(self.principal.as_deref());
self.gate.slot_freed.notify_waiters();
}
}
struct Parked<'a> {
governor: Option<&'a Governor>,
lane: Lane,
}
impl<'a> Parked<'a> {
fn new(governor: Option<&'a Governor>, lane: Lane) -> Parked<'a> {
if let Some(governor) = governor {
governor.entered_wait(lane);
}
Parked { governor, lane }
}
}
impl Drop for Parked<'_> {
fn drop(&mut self) {
if let Some(governor) = self.governor {
governor.left_wait(self.lane);
}
}
}
pub struct Metered<S> {
inner: S,
guard: Arc<AdmissionGuard>,
}
impl<S> Metered<S> {
pub fn new(inner: S, guard: Arc<AdmissionGuard>) -> Metered<S> {
Metered { inner, guard }
}
fn charge(&self, bytes: usize) {
let Some(growth) = self.guard.reserve_more(bytes as u64) else {
return;
};
if let Growth::Overdrawn {
held,
reserved,
budget,
} = growth
{
tracing::warn!(
target: "gunnar::git",
event = "git.transfer.overdrawn",
held_bytes = held,
reserved_bytes = reserved,
budget_bytes = budget,
"this transfer has now charged the memory budget past its limit; it finishes, \
and NEW transfers are refused by name until it does"
);
}
}
}
impl<S: tokio::io::AsyncRead + Unpin> tokio::io::AsyncRead for Metered<S> {
fn poll_read(
mut self: std::pin::Pin<&mut Self>,
cx: &mut std::task::Context<'_>,
buf: &mut tokio::io::ReadBuf<'_>,
) -> std::task::Poll<std::io::Result<()>> {
let before = buf.filled().len();
let polled = std::pin::Pin::new(&mut self.inner).poll_read(cx, buf);
if polled.is_ready() {
let read = buf.filled().len().saturating_sub(before);
if read > 0 {
self.charge(read);
}
}
polled
}
}
impl<S: tokio::io::AsyncWrite + Unpin> tokio::io::AsyncWrite for Metered<S> {
fn poll_write(
mut self: std::pin::Pin<&mut Self>,
cx: &mut std::task::Context<'_>,
buf: &[u8],
) -> std::task::Poll<std::io::Result<usize>> {
let polled = std::pin::Pin::new(&mut self.inner).poll_write(cx, buf);
if let std::task::Poll::Ready(Ok(written)) = &polled {
let written = *written;
if written > 0 {
self.charge(written);
}
}
polled
}
fn poll_flush(
mut self: std::pin::Pin<&mut Self>,
cx: &mut std::task::Context<'_>,
) -> std::task::Poll<std::io::Result<()>> {
std::pin::Pin::new(&mut self.inner).poll_flush(cx)
}
fn poll_shutdown(
mut self: std::pin::Pin<&mut Self>,
cx: &mut std::task::Context<'_>,
) -> std::task::Poll<std::io::Result<()>> {
std::pin::Pin::new(&mut self.inner).poll_shutdown(cx)
}
}
#[cfg(test)]
mod tests {
use super::*;
#[tokio::test]
async fn a_keyed_caller_waits_out_the_keyed_bound_and_a_stranger_does_not() {
let gate = Arc::new(
Admission::new(Some(1), None)
.with_wait(Duration::from_millis(50))
.with_keyed_wait(Some(Duration::from_secs(5))),
);
assert_eq!(gate.wait_for("admin"), Duration::from_secs(5));
assert_eq!(
gate.wait_for(gunnar_core::ANONYMOUS_PRINCIPAL_ID),
Duration::from_millis(50)
);
let holder = gate
.try_admit("other", "team", OpClass::Fetch)
.expect("the one slot");
let freer = tokio::spawn(async move {
tokio::time::sleep(Duration::from_millis(400)).await;
drop(holder);
});
let started = std::time::Instant::now();
let (guard, waited) = gate
.admit_within("admin", "team", OpClass::Fetch)
.await
.expect("a keyed caller is queued for the slot, not refused after 50 ms");
assert!(
waited >= Duration::from_millis(300),
"it really waited: {waited:?}"
);
assert!(started.elapsed() < Duration::from_secs(5));
freer.await.unwrap();
let started = std::time::Instant::now();
gate.admit_within(gunnar_core::ANONYMOUS_PRINCIPAL_ID, "team", OpClass::Fetch)
.await
.expect_err("anonymous is never queued for the keyed bound");
assert!(
started.elapsed() < Duration::from_secs(1),
"{:?}",
started.elapsed()
);
drop(guard);
let instant = Admission::new(Some(1), None)
.with_wait(Duration::ZERO)
.with_keyed_wait(Some(Duration::from_secs(30)));
assert_eq!(instant.wait_for("admin"), Duration::ZERO);
let plain = Admission::new(Some(1), None).with_wait(Duration::from_millis(50));
assert_eq!(plain.wait_for("admin"), Duration::from_millis(50));
}
#[tokio::test]
async fn a_paused_gate_refuses_with_the_growing_words_and_keeps_in_flight_honest() {
let gate = Arc::new(Admission::new(Some(4), None).with_wait(Duration::from_secs(5)));
let running = gate
.try_admit("one", "team", OpClass::Push)
.expect("admitted before the pause");
assert_eq!(gate.in_flight(), 1);
assert_eq!(gate.paused(), None);
gate.pause(Pause::Growing);
assert_eq!(gate.paused(), Some(Pause::Growing));
let refused = gate
.try_admit("two", "team", OpClass::Clone)
.expect_err("paused: refused");
assert_eq!(
refused,
Refusal::Paused {
pause: Pause::Growing
}
);
assert_eq!(refused.client_message(), GROWING_MESSAGE);
assert!(
refused.client_message().contains("growing"),
"the word is load-bearing"
);
assert!(refused.client_message().contains("try again"));
assert!(
!refused.client_message().contains("gunnar.rs"),
"a test2 push was told gunnar.rs is growing: {}",
refused.client_message()
);
assert!(
refused
.client_message()
.starts_with("The server is growing its storage"),
"{}",
refused.client_message()
);
assert_eq!(refused.reason(), "paused-growing");
for other in [
"concurrent git transfers",
"per-principal limit",
"unauthenticated",
] {
assert!(
!refused.client_message().contains(other),
"shares no substring with {other}"
);
}
let started = std::time::Instant::now();
let parked = gate
.admit_within("three", "team", OpClass::Fetch)
.await
.expect_err("paused: refused, not parked");
assert!(matches!(parked, Refusal::Paused { .. }));
assert!(
started.elapsed() < Duration::from_secs(1),
"not parked for the wait"
);
let stats = gate.stats();
assert_eq!(stats.refused_paused, 2);
assert_eq!(stats.refused(), 2);
assert_eq!(
stats.in_flight, 1,
"the pre-pause transfer is still what is in flight"
);
gate.pause(Pause::Growing);
assert_eq!(gate.paused(), Some(Pause::Growing), "idempotent");
drop(running);
assert_eq!(
gate.in_flight(),
0,
"the drain sees zero once the last one finishes"
);
gate.unpause();
gate.unpause();
assert_eq!(gate.paused(), None);
let _again = gate
.try_admit("two", "team", OpClass::Clone)
.expect("unpaused: admitted");
assert_eq!(gate.in_flight(), 1);
}
#[test]
fn the_server_wide_limit_refuses_and_counts() {
let gate = Arc::new(Admission::new(Some(2), None));
let a = gate
.try_admit("one", "team", OpClass::LsRemote)
.expect("the first is admitted");
let b = gate
.try_admit("two", "team", OpClass::LsRemote)
.expect("the second is admitted");
let refused = gate
.try_admit("three", "team", OpClass::LsRemote)
.expect_err("the third is refused");
assert_eq!(refused, Refusal::AtCapacity { limit: 2 });
let stats = gate.stats();
assert_eq!(stats.admitted, 2, "two got in");
assert_eq!(stats.refused_at_capacity, 1, "the refusal must be COUNTED");
assert_eq!(stats.in_flight, 2);
assert_eq!(stats.peak_in_flight, 2);
drop(a);
assert_eq!(gate.stats().in_flight, 1);
let c = gate
.try_admit("three", "team", OpClass::LsRemote)
.expect("a freed slot is reusable");
assert_eq!(gate.stats().in_flight, 2);
assert_eq!(
gate.stats().peak_in_flight,
2,
"the peak is a high-water mark, not the current value"
);
drop((b, c));
assert_eq!(gate.stats().in_flight, 0);
}
#[test]
fn one_principal_alone_reaches_the_server_wide_limit() {
use crate::config::{
DEFAULT_MAX_CONCURRENT_TRANSFERS as CONFIGURED,
DEFAULT_MAX_TRANSFERS_PER_PRINCIPAL as FLOOR,
};
let gate = Arc::new(Admission::new(Some(CONFIGURED), Some(FLOOR)));
let governed = gate.governed_limit().expect("a governed limit");
assert_eq!(
governed, CONFIGURED,
"a fresh governor starts at the top of its band, and the top is what \
the operator configured"
);
assert!(
governed > FLOOR,
"a test where the governed limit is not above the per-principal floor cannot tell a \
floor from a wall"
);
let held: Vec<_> = (0..governed)
.map(|i| {
gate.try_admit("solo", "team", OpClass::LsRemote)
.unwrap_or_else(|refusal| {
panic!(
"transfer {i} of {governed} was refused on an EMPTY server holding \
one identity: {refusal}. The per-principal number is a floor, not a \
wall; a single-key client that stops at {FLOOR} is the old invisible ceiling, back"
)
})
})
.collect();
assert_eq!(held.len(), governed);
let stats = gate.stats();
assert_eq!(
stats.admitted, governed as u64,
"the server-wide limit in force must be reachable by ONE identity"
);
assert_eq!(
stats.refused_per_principal, 0,
"not one of these {governed} may be refused for fairness: there was nobody to be \
fair to"
);
assert_eq!(stats.peak_in_flight, governed as u64);
assert_eq!(stats.limit, governed, "the stats must report what refuses");
let full = gate
.try_admit("solo", "team", OpClass::LsRemote)
.expect_err("one past the governed limit");
assert_eq!(full, Refusal::AtCapacity { limit: governed });
drop(held);
}
#[test]
fn a_greedy_principal_is_held_to_its_share_once_the_server_is_contended() {
let gate = Arc::new(Admission::new(Some(4), Some(1)));
let _polite = gate
.try_admit("polite", "team", OpClass::LsRemote)
.expect("polite's first");
let _g1 = gate
.try_admit("greedy", "team", OpClass::LsRemote)
.expect("greedy's first");
let _g2 = gate
.try_admit("greedy", "team", OpClass::LsRemote)
.expect("greedy's second is still its share");
let refused = gate
.try_admit("greedy", "team", OpClass::LsRemote)
.expect_err("greedy's third is past its share of a contended server");
assert_eq!(
refused,
Refusal::PrincipalAtCapacity {
limit: 2,
active_principals: 2
},
"the refusal must name the SHARE in force (4/2), not the configured floor (1), or an \
operator reads a number that matches nothing they set"
);
assert!(
refused.client_message().contains("per-principal limit"),
"installtest's `refusals_in_log` tells the two refusals apart by this exact phrase; \
losing it silently zeroes the server-side witness: {}",
refused.client_message()
);
let _other = gate
.try_admit("polite", "team", OpClass::LsRemote)
.expect("the slot greedy was refused must be available to the quiet principal");
let stats = gate.stats();
assert_eq!(stats.refused_per_principal, 1);
assert_eq!(
stats.refused_at_capacity, 0,
"the server had a free slot; attributing this to capacity sends an operator to raise \
the wrong number"
);
assert_eq!(stats.in_flight, 4);
}
#[test]
fn fairness_does_not_bind_while_there_are_free_slots() {
let gate = Arc::new(Admission::new(Some(100), Some(1)));
let _quiet = gate
.try_admit("quiet", "team", OpClass::LsRemote)
.expect("one transfer from a quiet identity");
let busy: Vec<_> = (0..73)
.map(|i| {
gate.try_admit("busy", "team", OpClass::LsRemote)
.unwrap_or_else(|r| panic!("transfer {i} refused at {}% full: {r}", i + 2))
})
.collect();
assert_eq!(
busy.len(),
73,
"74 of 100 held, so nothing may be refused yet"
);
assert_eq!(gate.stats().refused_per_principal, 0);
let refused = gate
.try_admit("busy", "team", OpClass::LsRemote)
.expect_err(
"the transfer that makes the server 75% full is the first one weighed \
against the share",
);
assert_eq!(
refused,
Refusal::PrincipalAtCapacity {
limit: 50,
active_principals: 2
}
);
}
#[test]
fn a_hard_cap_binds_on_an_empty_server() {
let fair = Arc::new(Admission::new(Some(64), Some(2)));
let _a = fair
.try_admit("solo", "team", OpClass::LsRemote)
.expect("first");
let _b = fair
.try_admit("solo", "team", OpClass::LsRemote)
.expect("second");
fair.try_admit("solo", "team", OpClass::LsRemote)
.expect("a fair gate lets a lone principal past its floor on an idle server");
let hard = Arc::new(Admission::hard_capped(Some(64), Some(2)));
let _c = hard
.try_admit("solo", "team", OpClass::LsRemote)
.expect("first");
let _d = hard
.try_admit("solo", "team", OpClass::LsRemote)
.expect("second");
let refused = hard
.try_admit("solo", "team", OpClass::LsRemote)
.expect_err("a pinned cap binds whether or not anyone else is there");
assert_eq!(
refused,
Refusal::PrincipalAtCapacity {
limit: 2,
active_principals: 1
}
);
assert!(
!refused.client_message().contains("fair share"),
"a pinned cap is not a share and must not be explained as one: {}",
refused.client_message()
);
assert_eq!(hard.stats().refused_per_principal, 1);
assert_eq!(fair.stats().refused_per_principal, 0);
}
#[test]
fn a_refused_transfer_does_not_leak_a_server_wide_slot() {
let gate = Arc::new(Admission::new(Some(4), Some(1)));
let _polite = gate
.try_admit("polite", "team", OpClass::LsRemote)
.expect("polite is active");
let _g1 = gate
.try_admit("greedy", "team", OpClass::LsRemote)
.expect("greedy's share is 2");
let _g2 = gate
.try_admit("greedy", "team", OpClass::LsRemote)
.expect("greedy's share is 2");
for _ in 0..100 {
gate.try_admit("greedy", "team", OpClass::LsRemote)
.expect_err("refused every time");
}
let _other = gate
.try_admit("somebody-else", "team", OpClass::LsRemote)
.expect("the fourth slot survived 100 refusals");
assert_eq!(gate.stats().in_flight, 4);
assert_eq!(gate.stats().refused_per_principal, 100);
}
#[test]
fn a_departed_principal_stops_shrinking_everybody_elses_share() {
let gate = Arc::new(Admission::new(Some(4), Some(1)));
let leaving = gate
.try_admit("leaving", "team", OpClass::LsRemote)
.expect("two identities are active");
let _g1 = gate
.try_admit("stayer", "team", OpClass::LsRemote)
.expect("first");
let _g2 = gate
.try_admit("stayer", "team", OpClass::LsRemote)
.expect("share of 2 with 2 active");
gate.try_admit("stayer", "team", OpClass::LsRemote)
.expect_err("held to 2 while `leaving` is still there");
drop(leaving);
let _g3 = gate
.try_admit("stayer", "team", OpClass::LsRemote)
.expect("with the other identity gone the share must widen back out");
let _g4 = gate
.try_admit("stayer", "team", OpClass::LsRemote)
.expect("all four slots are reachable by the only identity left");
assert_eq!(gate.stats().in_flight, 4);
}
#[test]
fn the_map_does_not_grow_with_every_principal_ever_seen() {
let gate = Arc::new(Admission::new(Some(4), Some(1)));
for i in 0..500 {
let guard = gate
.try_admit(&format!("principal-{i}"), "team", OpClass::LsRemote)
.expect("admitted");
drop(guard);
}
let held = gate.per_principal.lock().expect("the per-principal map");
assert!(
held.is_empty(),
"the per-principal map kept {} entries for principals with nothing in flight",
held.len()
);
}
#[test]
fn the_three_refusals_do_not_borrow_each_others_words() {
let capacity = Refusal::AtCapacity { limit: 37 }.client_message();
let principal = Refusal::PrincipalAtCapacity {
limit: 4,
active_principals: 3,
}
.client_message();
let memory = Refusal::MemoryBudget {
class: OpClass::Clone,
wanted: 2048 * 1024 * 1024,
reserved: 6144 * 1024 * 1024,
budget: 8192 * 1024 * 1024,
}
.client_message();
assert!(capacity.contains("concurrent git transfers"));
assert!(!capacity.contains("per-principal limit"));
assert!(principal.contains("per-principal limit"));
assert!(!principal.contains("concurrent git transfers"));
assert!(
!memory.contains("concurrent git transfers"),
"the memory refusal contains \"concurrent git transfers\", which installtest counts \
as a CAPACITY refusal; an operator reading that raises the concurrency limit and \
makes the memory problem worse: {memory}"
);
assert!(
!memory.contains("per-principal limit"),
"the memory refusal contains \"per-principal limit\", so installtest counts a \
server-wide memory exhaustion as one identity's fairness throttle: {memory}"
);
assert!(
memory.contains("memory budget"),
"the memory refusal must name the knob to turn: {memory}"
);
for message in [&capacity, &principal, &memory] {
assert!(
message.contains("try again"),
"a transient refusal that does not say `try again` reads as permanent: {message}"
);
}
}
#[derive(Default)]
struct Work {
runs: AtomicU64,
bytes_allocated: AtomicU64,
}
impl Work {
fn perform(&self) -> Vec<u8> {
let buffer = vec![0u8; 1024 * 1024];
self.runs.fetch_add(1, Ordering::Relaxed);
self.bytes_allocated
.fetch_add(buffer.len() as u64, Ordering::Relaxed);
buffer
}
fn runs(&self) -> u64 {
self.runs.load(Ordering::Relaxed)
}
fn bytes(&self) -> u64 {
self.bytes_allocated.load(Ordering::Relaxed)
}
}
#[test]
fn a_denial_is_paid_for_by_nobody() {
let gate = Arc::new(Admission::new(Some(10), None));
let work = Work::default();
let mut held = Vec::new();
let mut refused = 0;
for _ in 0..40 {
match gate.try_admit("solo", "team", OpClass::LsRemote) {
Ok(guard) => held.push((guard, work.perform())),
Err(_) => refused += 1,
}
}
assert_eq!(refused, 30, "10 of 40 fit under the limit");
assert_eq!(
work.runs(),
10,
"{refused} operations were refused, but the work ran {} times and allocated {} \
bytes. A refusal after the allocation is not load shedding",
work.runs(),
work.bytes()
);
assert_eq!(work.bytes(), 10 * 1024 * 1024);
let clone = OpClass::Clone.admission_bytes();
let gate = Arc::new(Admission::new(None, None).with_budget(clone * 4));
assert!(
gate.memory_budget().is_none(),
"an unlimited concurrency gate has no budget to attach one to; the memory gate must \
ride on the same governor or a configuration could get one without the other"
);
let gate = Arc::new(Admission::with_memory_budget(Some(100), None, clone * 4));
let work = Work::default();
let mut held = Vec::new();
let mut refused = Vec::new();
for _ in 0..8 {
match gate.try_admit("solo", "team", OpClass::Clone) {
Ok(guard) => held.push((guard, work.perform())),
Err(refusal) => refused.push(refusal),
}
}
assert_eq!(work.runs(), 4, "a 4-clone budget admits 4 clones");
assert_eq!(
work.bytes(),
4 * 1024 * 1024,
"the 4 refused clones allocated {} bytes between them, which is 4 more than a \
refusal at admission can possibly cost",
work.bytes() - 4 * 1024 * 1024
);
assert_eq!(refused.len(), 4);
assert!(
refused.iter().all(|r| matches!(
r,
Refusal::MemoryBudget {
class: OpClass::Clone,
..
}
)),
"the refusals must name MEMORY, or an operator raises the concurrency limit and \
makes it worse: {refused:?}"
);
let stats = gate.stats();
assert_eq!(stats.refused_memory, 4);
assert_eq!(
stats.refused_at_capacity, 0,
"the limit was 100 and 4 were held"
);
assert_eq!(stats.reserved_bytes, clone * 4);
}
#[test]
fn an_operation_that_fails_still_gives_its_reservation_back() {
let clone = OpClass::Clone.admission_bytes();
let gate = Arc::new(Admission::with_memory_budget(Some(100), None, clone * 4));
for _ in 0..3 {
let outcome: Result<(), &str> = (|| {
let _guard = gate
.try_admit("solo", "team", OpClass::Clone)
.map_err(|_| "refused")?;
Err("the transport died mid-pack")
})();
assert!(outcome.is_err());
}
for _ in 0..3 {
let panicked = std::panic::catch_unwind(std::panic::AssertUnwindSafe(|| {
let _guard = gate
.try_admit("solo", "team", OpClass::Clone)
.expect("admitted before the panic");
panic!("the pack walk blew up");
}));
assert!(panicked.is_err());
}
let stats = gate.stats();
assert_eq!(
stats.reserved_bytes,
0,
"6 operations ended in an error or a panic and {} bytes are still reserved of {}; \
the 7th would be refused, so the server has wedged",
stats.reserved_bytes,
gate.memory_budget().unwrap_or(0)
);
assert_eq!(stats.in_flight, 0);
assert_eq!(
gate.governor().expect("a governor").occupied(),
0,
"the concurrency slots leaked even though the reservations did not"
);
let held: Vec<_> = (0..4)
.map(|i| {
gate.try_admit("solo", "team", OpClass::Clone)
.unwrap_or_else(|r| panic!("clone {i} after 6 failures was refused: {r}"))
})
.collect();
assert_eq!(held.len(), 4);
}
#[test]
fn a_transfer_that_never_enters_service_is_counted_as_unmeasured() {
let gate = Arc::new(Admission::new(Some(100), None));
drop(
gate.try_admit("solo", "team", OpClass::LsRemote)
.expect("admitted"),
);
assert_eq!(
gate.stats().unmeasured_sojourn,
1,
"a transfer that never said its work started filed no sojourn sample and must be \
counted; a controller starved of samples grows to its ceiling and stays there"
);
let guard = gate
.try_admit("solo", "team", OpClass::LsRemote)
.expect("admitted");
guard.entered_service();
drop(guard);
assert_eq!(
gate.stats().unmeasured_sojourn,
1,
"a transfer that DID report entering service was counted as unmeasured: {}",
gate.stats().unmeasured_sojourn
);
let governor = gate.governor().expect("a governor");
assert_eq!(governor.codel().shrinks() + governor.codel().grows(), 0);
let guard = gate
.try_admit("solo", "team", OpClass::LsRemote)
.expect("admitted");
guard.entered_service();
drop(guard);
assert_eq!(
gate.stats().unmeasured_sojourn,
1,
"the one-shot mark must not turn a second measured transfer into an unmeasured one"
);
}
#[test]
fn an_unlimited_gate_admits_everything_and_still_counts() {
let gate = Arc::new(Admission::unlimited());
let mut held = Vec::new();
for _ in 0..1000 {
held.push(
gate.try_admit("anyone", "team", OpClass::LsRemote)
.expect("no limit means no refusal"),
);
}
let stats = gate.stats();
assert_eq!(stats.admitted, 1000);
assert_eq!(stats.refused(), 0);
assert_eq!(stats.in_flight, 1000);
assert_eq!(stats.peak_in_flight, 1000);
}
}