crates/gunnar-server/tests/grpc_admission.rs
//! **The gRPC control plane's one genuine transfer takes the git fronts' own
//! admission permit, and it REFUSES over budget** (the gRPC half).
//!
//! # The gap this closes
//!
//! The SSH and smart-HTTP front doors put every git transfer through ONE gate,
//! `gunnar_ssh::Admission`: a server-wide concurrency ceiling, a max-min-fair
//! per-principal share, and a memory ledger. The gRPC control plane on :50051
//! took **none** of it. Most control RPCs are cheap metadata and correctly stay
//! ungated — the control plane runs on a runtime of its own, so their
//! walks starve nothing but themselves. But `Repos.ArchiveOut` streams a whole
//! repository's stored bytes down the control channel: a transfer in everything
//! but name, and nothing budgeted it. Now it takes a permit from the SAME gate a
//! `git-upload-pack` does, through the `gunnar_grpc::TransferGate` seam
//! `gunnar-server::control_transfer_gate` implements over the real
//! `Arc<Admission>`.
//!
//! # The red before the green
//!
//! A test that fires an `ArchiveOut` and sees it succeed proves nothing: that is
//! exactly the green an ungated server produces. So the three tests here are a
//! matched set, and two of them are the red:
//!
//! * [`archive_out_streams_when_the_gate_is_unwired`] — a gate is full, but the
//! service was built with **no** gate (the ungated state). The archive
//! streams anyway, and the gate records no refusal. This is the hollow guard.
//! * [`archive_out_streams_under_an_unlimited_gate`] — a gate IS wired, but with
//! no ceiling. Many permits are held; the archive still streams. So the
//! refusal in the green is the ceiling's doing, not the wiring's.
//! * [`archive_out_is_refused_when_the_gate_is_full`] — the green. A gate with a
//! server-wide ceiling of `K`, `K` permits held, and the archive is **refused
//! with `RESOURCE_EXHAUSTED`**. Asserted on the applied gRPC status code AND
//! on the gate's own counters — `admitted == K`, `refused == 1`, and the `K`
//! held permits still in flight — never on a permit object existing.
//!
//! The `K` held permits are real `try_admit`s against the very gate the service
//! consults, so they are `K` in-flight transfers by construction; the archive is
//! then the `(K+1)`th, which is what makes "exactly `K` admitted, the rest
//! refused" the thing being asserted.
use std::sync::Arc;
use gunnar_core::{HashKind, NewRepo, Principal, PrincipalKind, RepoRef};
use gunnar_grpc::{Caller, EventBus, ServerConfig, ServiceContext, TransferGate};
use gunnar_proto::repos_server::Repos as _;
use gunnar_proto::{ArchiveOutRequest, RepoPath};
use gunnar_server::stores::ServerStores;
use gunnar_server::{control_transfer_gate, CoreControlPlane};
use gunnar_ssh::governor::OpClass;
use gunnar_ssh::{Admission, AdmissionGuard};
use tonic::{Code, Request};
const NS: &str = "alice";
const NAME: &str = "backup";
/// The server-wide ceiling under test. Small so a test can hold it full, and
/// pinned exactly: `LimitBand::for_budget(3)` is the band `[3, 3]`, a governor
/// that cannot move, so "over the ceiling" is deterministic.
const K: usize = 3;
/// A memory budget so large the ledger never binds first. The archive permit is
/// `OpClass::Clone`, which reserves a few MiB; this test is about the
/// **concurrency** ceiling, and a memory refusal here would be the right status
/// for the wrong reason (the identity-value trap in another guise).
const HUGE_BUDGET: u64 = 64 * 1024 * 1024 * 1024;
/// A repository owned by `alice`, in the `alice` namespace, whose principal
/// is intrinsically that namespace's admin — so `ArchiveOut`'s `Read` check
/// passes and the gate is what refuses, not the grant.
struct Rig {
_dir: tempfile::TempDir,
repos: gunnar_grpc::ReposService,
rt: tokio::runtime::Runtime,
}
impl Rig {
fn new(gate: Option<Arc<dyn TransferGate>>) -> Rig {
let dir = tempfile::tempdir().expect("tempdir");
let stores = Arc::new(ServerStores::open(dir.path()).expect("stores"));
stores.catalog().create_namespace(NS).expect("namespace");
stores
.auth()
.add_principal(Principal::new(NS, NS, PrincipalKind::User).expect("id"))
.expect("principal");
stores
.create_repo(
NewRepo::new(RepoRef::new(NS, NAME).expect("legal")).hash_kind(HashKind::Sha1),
false,
)
.expect("the repository to archive");
let plane: Arc<dyn gunnar_grpc::ControlPlane> =
Arc::new(CoreControlPlane::new(Arc::clone(&stores)));
let mut ctx =
ServiceContext::new(plane, Arc::new(EventBus::new(16)), ServerConfig::default());
if let Some(gate) = gate {
ctx = ctx.with_transfer_gate(gate);
}
Rig {
_dir: dir,
repos: gunnar_grpc::ReposService::new(ctx),
// A current-thread runtime is enough: the handler decides admission
// synchronously and returns before the cut's `spawn_blocking` runs,
// which is exactly the point at which the refusal is or is not made.
rt: tokio::runtime::Builder::new_current_thread()
.enable_all()
.build()
.expect("runtime"),
}
}
/// Drive `Repos.ArchiveOut` as `principal` and hand back the applied
/// outcome — the real gRPC status code off the handler, or `Ok` for an
/// accepted stream.
fn archive_out(&self, principal: &str) -> Result<(), Code> {
let mut request = Request::new(ArchiveOutRequest {
path: Some(RepoPath {
namespace: NS.to_owned(),
repo: NAME.to_owned(),
}),
from_generation: 0,
});
request.extensions_mut().insert(Caller {
principal_id: principal.to_owned(),
fingerprint: "SHA256:test".to_owned(),
machine: false,
authenticated: true,
});
self.rt
.block_on(self.repos.archive_out(request))
.map(|_| ())
.map_err(|status| status.code())
}
}
/// A gate with a fixed server-wide ceiling of `K`, and `K` permits already held
/// against it — a gate that is genuinely full.
fn full_gate() -> (Arc<Admission>, Vec<AdmissionGuard>) {
let admission = Arc::new(Admission::with_memory_budget(Some(K), None, HUGE_BUDGET));
let held: Vec<AdmissionGuard> = (0..K)
.map(|i| {
admission
.try_admit(&format!("filler-{i}"), "team", OpClass::Clone)
.expect("the first K transfers are admitted")
})
.collect();
let stats = admission.stats();
assert_eq!(stats.admitted, K as u64, "exactly K admitted");
assert_eq!(stats.in_flight, K as u64, "and all K are in flight");
assert_eq!(stats.refused(), 0, "nothing refused yet");
(admission, held)
}
/// **THE GREEN.** A full gate, wired into the service, refuses `ArchiveOut` with
/// `RESOURCE_EXHAUSTED` — and the gate's own counters say so.
#[test]
fn archive_out_is_refused_when_the_gate_is_full() {
let (admission, held) = full_gate();
let rig = Rig::new(Some(control_transfer_gate(Arc::clone(&admission))));
let outcome = rig.archive_out(NS);
assert_eq!(
outcome,
Err(Code::ResourceExhausted),
"an ArchiveOut over the server-wide ceiling must be refused with RESOURCE_EXHAUSTED, \
the gRPC idiom for 429 — not queued, not served"
);
let stats = admission.stats();
assert_eq!(
stats.refused(),
1,
"the refusal is the archive's, counted on the gate itself (not a permit object)"
);
assert_eq!(
stats.admitted, K as u64,
"the refused archive took NO slot: still exactly K admitted"
);
assert_eq!(
stats.in_flight, K as u64,
"and the K real transfers are still in flight — the archive never entered"
);
drop(held);
}
/// **THE RED.** The same full gate, but the service was built with no gate at
/// all: the archive streams, and the gate records no refusal. This is the
/// ungated state — a green here would prove the guard is hollow.
#[test]
fn archive_out_streams_when_the_gate_is_unwired() {
let (admission, held) = full_gate();
// The gate exists and is full, but it is NOT wired into the service.
let rig = Rig::new(None);
let outcome = rig.archive_out(NS);
assert_eq!(
outcome,
Ok(()),
"with no gate wired in, the archive streams however full the gate is — this is the \
hollow-guard state the green must be distinguished from"
);
assert_eq!(
admission.stats().refused(),
0,
"an unwired handler never consulted the gate, so it refused nothing"
);
drop(held);
}
/// **THE RED, second arm.** A gate IS wired, but with no ceiling. Even holding
/// K+5 permits, an unlimited gate admits the archive — so the green's refusal is
/// the *ceiling's* doing, not merely the wiring's.
#[test]
fn archive_out_streams_under_an_unlimited_gate() {
let admission = Arc::new(Admission::unlimited());
let held: Vec<AdmissionGuard> = (0..K + 5)
.map(|i| {
admission
.try_admit(&format!("filler-{i}"), "team", OpClass::Clone)
.expect("an unlimited gate admits every transfer")
})
.collect();
let rig = Rig::new(Some(control_transfer_gate(Arc::clone(&admission))));
let outcome = rig.archive_out(NS);
assert_eq!(
outcome,
Ok(()),
"an unlimited gate admits the archive no matter how many transfers are in flight"
);
assert_eq!(
admission.stats().refused(),
0,
"nothing to refuse: the ceiling, not the wiring, is what makes the green red"
);
drop(held);
}