crates/gunnar-ssh/tests/admit_share.rs
//! **The usage share** (`admit-share`): a namespace over its rolling-window
//! budget is held to the anonymous ceiling, a PassThrough policy decides
//! exactly as the gate always did, the plan says why in one sentence, and a
//! finished transfer teaches the estimate.
//!
//! The rest of this crate's admission tests (`admission.rs`,
//! `anonymous_ceiling.rs`, `keyed_reserve.rs`, the unit tests) are the other
//! half of "PassThrough is identical": they run unchanged with
//! `--features admit-share`, where every gate they build holds the default
//! PassThrough policy.
#![cfg(feature = "admit-share")]
use std::sync::Arc;
use edda_admit::{Admit, Class, Settings, WorkKind};
use gunnar_core::{ControlStore, FlowLedger, FlowStore, RedbControlStore, Transfer};
use gunnar_ssh::governor::OpClass;
use gunnar_ssh::{Admission, Refusal};
/// A flow ledger in which `heavy` moved `heavy_bytes` today and `light` 10 bytes.
fn ledger(heavy_bytes: u64) -> Arc<FlowLedger> {
let store = Arc::new(RedbControlStore::in_memory().unwrap()) as Arc<dyn ControlStore>;
let flow = Arc::new(FlowLedger::new(FlowStore::new(store).unwrap()));
let now = gunnar_core::now_ms();
for (ns, bytes) in [("heavy", heavy_bytes), ("light", 10)] {
flow.record(
ns,
now,
&Transfer {
bytes_out: bytes,
fetched: true,
..Transfer::default()
},
);
}
flow
}
/// Anonymous ceiling 2; server-wide 8; the policy as given.
fn gate(policy: Arc<dyn Admit>, flow: Arc<FlowLedger>) -> Arc<Admission> {
Arc::new(
Admission::new(Some(8), Some(1))
.with_anonymous_ceiling(Some(2))
.with_flow(flow)
.with_policy(policy),
)
}
/// A usage share with `bytes` per 30 days, SEEDED from `flow` — as
/// `GunnarSsh::with_flow` builds it.
fn budget(bytes: u64, flow: &Arc<FlowLedger>) -> Arc<dyn Admit> {
Settings {
limit: Some(bytes),
..Settings::default()
}
.build_seeded(Some(Admission::flow_seed(Arc::clone(flow))))
}
/// ★ **Over budget is the anonymous share; within budget is not.** On an idle
/// box a within-budget identity takes as many slots as it likes (the
/// per-principal share only binds under contention); an over-budget one is
/// held to the anonymous ceiling of 2, refused with its own reason, and never
/// parked.
#[tokio::test]
async fn an_over_budget_namespace_gets_the_anonymous_ceiling() {
let flow = ledger(5000);
let gate = gate(budget(1000, &flow), flow);
let heavy: Vec<_> = (0..2)
.map(|_| {
gate.try_admit("p-heavy", "heavy", OpClass::Clone)
.expect("inside the ceiling")
})
.collect();
let refused = gate
.try_admit("p-heavy", "heavy", OpClass::Clone)
.expect_err("a third transfer is past the reduced share");
assert_eq!(
refused,
Refusal::ReducedShare {
ceiling: 2,
busy: false
}
);
assert_eq!(refused.reason(), "admission-reduced-share");
let words = refused.client_message();
assert!(words.contains("over its transfer budget"), "{words}");
assert!(
!words.contains("concurrent git transfers") && !words.contains("per-principal limit"),
"installtest tells refusals apart by those substrings: {words}"
);
// Within budget, same box: three at once, no share applied.
let light: Vec<_> = (0..3)
.map(|_| {
gate.try_admit("p-light", "light", OpClass::Clone)
.expect("Full share")
})
.collect();
assert_eq!(gate.stats().refused_reduced, 1);
assert_eq!(
gate.stats().refused_anonymous,
0,
"not counted as a stranger"
);
// admit_within never parks it, even with a wait configured.
let started = std::time::Instant::now();
let parked = gate.admit_within("p-heavy", "heavy", OpClass::Clone).await;
assert!(matches!(parked, Err(Refusal::ReducedShare { .. })));
assert!(
started.elapsed() < gate.wait(),
"the reduced share was parked"
);
drop((heavy, light));
}
/// ★ **Under load the over-budget namespace is refused by the keyed reserve**
/// while within-budget identities still get in: once the box is past
/// `CONTENTION_PERCENT`, a reduced namespace — like a stranger — may not take
/// the top of it.
#[tokio::test]
async fn under_load_the_over_budget_namespace_gives_way() {
let flow = ledger(5000);
let gate = gate(budget(1000, &flow), flow);
// Load the box with within-budget work: 6 of 8 is past 75 %.
let load: Vec<_> = (0..6)
.map(|i| {
gate.try_admit(&format!("p-light-{i}"), "light", OpClass::Fetch)
.expect("the box has room")
})
.collect();
assert_eq!(
gate.try_admit("p-heavy", "heavy", OpClass::Clone)
.expect_err("reserved"),
Refusal::ReducedShare {
ceiling: 2,
busy: true
}
);
// …and a within-budget identity is still admitted into that reserve.
let keyed = gate.try_admit("p-light-x", "light", OpClass::Fetch);
assert!(keyed.is_ok(), "{keyed:?}");
drop(load);
}
/// ★ **PassThrough decides as the gate always did**: the same over-budget
/// ledger, no budget, and the heavy namespace takes the Full share —
/// indistinguishable from a gate with no policy at all.
#[test]
fn pass_through_is_the_gate_as_it_was() {
let flow = ledger(u64::MAX / 4);
for policy in [
Settings::pass_through().build(),
Arc::new(edda_admit::PassThrough::new()) as Arc<dyn Admit>,
] {
let with = gate(policy, Arc::clone(&flow));
let without = Arc::new(
Admission::new(Some(8), Some(1))
.with_anonymous_ceiling(Some(2))
.with_flow(Arc::clone(&flow)),
);
let a: Vec<_> = (0..8)
.map(|_| with.try_admit("p-heavy", "heavy", OpClass::Clone))
.collect();
let b: Vec<_> = (0..8)
.map(|_| without.try_admit("p-heavy", "heavy", OpClass::Clone))
.collect();
assert_eq!(
a.iter()
.map(|r| r.as_ref().err().copied())
.collect::<Vec<_>>(),
b.iter()
.map(|r| r.as_ref().err().copied())
.collect::<Vec<_>>(),
);
assert!(
a.iter().all(Result::is_ok),
"an idle box is the heavy namespace's to fill"
);
assert_eq!(with.stats().refused_reduced, 0);
}
}
/// ★ **The plan says why, in the design's sentence**, off the flow ledger's
/// rolling window.
#[test]
fn the_plan_text_names_the_budget_the_usage_and_the_class() {
let flow = ledger(63_000_000_000);
let gate = gate(budget(100_000_000_000, &flow), flow);
let policy = gate.admit_policy();
let usage = gate.admit_usage(&*policy, "heavy");
assert_eq!(
usage,
policy.usage(&edda_admit::Consumer::Namespace("heavy".into()))
);
assert_eq!(usage, edda_admit::Usage::new(63_000_000_000, 30));
let plan = policy.plan(
&edda_admit::Work::new(WorkKind::Clone, Some(1_200_000_000)),
&edda_admit::Consumer::Namespace("heavy".into()),
&usage,
);
assert_eq!(plan.class, Class::Full);
assert_eq!(plan.why, "100 GB/30d used 63 GB; this clone ~1.2 GB; Full");
assert_eq!(policy.policy(), "usage-share");
}
/// ★ **A finished transfer feeds the estimate and the usage**: the wire
/// total the memory gate already counts (`reserve_more`, called by the
/// transport as bytes move) reaches `record` ONCE, when the guard drops —
/// outbound for a clone, inbound for a push — and the next plan's estimate is
/// what was learnt, while the namespace's usage grows by it without the
/// ledger being read again.
#[test]
fn the_record_feeds_the_estimate() {
let flow = ledger(10);
let gate = gate(budget(1_000_000, &flow), flow);
let policy = gate.admit_policy();
let light = edda_admit::Consumer::Namespace("light".into());
assert_eq!(policy.usage(&light).bytes, 10, "seeded from the ledger");
let guard = gate.try_admit("p-light", "light", OpClass::Clone).unwrap();
guard.reserve_more(7_300);
drop(guard);
let push = gate.try_admit("p-light", "light", OpClass::Push).unwrap();
push.reserve_more(400);
drop(push);
assert_eq!(
policy.usage(&light).bytes,
10 + 7_300 + 400,
"the table is the cache"
);
let rows = policy.estimates();
assert_eq!(rows.len(), 2);
assert_eq!((rows[0].kind, rows[0].samples), (WorkKind::Clone, 1));
assert_eq!((rows[0].mean.net_in, rows[0].mean.net_out), (0, 7_300));
assert_eq!(
(rows[1].kind, rows[1].mean.net_in, rows[1].mean.net_out),
(WorkKind::Push, 400, 0)
);
let plan = gate.admit_policy().plan(
&edda_admit::Work::new(WorkKind::Clone, None),
&edda_admit::Consumer::Namespace("light".into()),
&edda_admit::Usage::none(),
);
assert_eq!(plan.estimate.usage_bytes(), 7_300);
assert!(plan.why.contains("this clone ~7.3 KB"), "{}", plan.why);
}
/// Anonymous keeps its own ceiling and is never planned against a namespace.
#[test]
fn anonymous_is_not_planned() {
let flow = ledger(5000);
let gate = gate(budget(1, &flow), flow);
let a: Vec<_> = (0..2)
.map(|_| {
gate.try_admit(gunnar_core::ANONYMOUS_PRINCIPAL_ID, "heavy", OpClass::Clone)
.unwrap()
})
.collect();
assert_eq!(
gate.try_admit(gunnar_core::ANONYMOUS_PRINCIPAL_ID, "heavy", OpClass::Clone)
.unwrap_err(),
Refusal::AnonymousAtCapacity { limit: 2 }
);
assert_eq!(
gate.stats().refused_reduced,
0,
"a stranger is not a reduced namespace"
);
drop(a);
// A stranger's transfer is not recorded against the policy either.
assert!(gate.admit_policy().estimates().is_empty());
}