crates/gunnar-ssh/tests/admission.rs
//! Saturation must be visible.
//!
//! The bug was not "there is no limit". It was that the limit existed
//! accidentally (tokio's default 512 blocking threads), was never chosen, and
//! produced *no signal at all* when it was reached. That last part is the one
//! that costs an operator a day: transfers queued in silence and the symptom was
//! "git is slow".
//!
//! So these tests assert three separate things, and all three have to hold for
//! the fix to be worth anything:
//!
//! 1. the limit is **reached**: a transfer past it does not run;
//! 2. the refusal is **told to the client**, in git's own framing, with an exit
//! status that says *transient* rather than *denied*;
//! 3. the refusal is **counted**, so a server that is refusing all day says so.
//!
//! Every test here holds its transfers genuinely in flight
//! ([`BannerTransport::holding`]) rather than relying on scheduling luck to make
//! two instantaneous transfers overlap. A concurrency limit tested against a
//! transport that returns immediately is a test of the scheduler.
//!
//! # Seen red
//!
//! Measured 2026-08-03 by replacing the gate in `GunnarSsh::new` with
//! `Admission::unlimited()` and re-running this file: the third transfer was
//! admitted, `stdout` carried the banner instead of an `ERR`, no exit status
//! arrived, and `refused_at_capacity` stayed at 0. That is exactly the pre-fix
//! behaviour, and it is what `the_limit_is_the_thing_being_tested` pins from the
//! other side without needing the source edited.
mod common;
use std::time::Duration;
use common::*;
use gunnar_ssh::seam::BannerTransport;
use gunnar_ssh::{Access, SshConfig, EXIT_BUSY};
use russh::keys::ssh_key::PrivateKey;
/// Everything one exec request produced, without draining a channel that is
/// deliberately still running.
struct Started {
/// The channel, kept alive so the transfer stays in flight. Never read;
/// **holding** it is the whole job, and dropping it would end the transfer
/// and free the slot the next assertion is about.
#[allow(dead_code)]
channel: russh::Channel<russh::client::Msg>,
/// The first data band that arrived: the banner for an admitted transfer,
/// the `ERR` packet for a refused one.
first_data: String,
/// The exit status, if one had already arrived. An admitted-and-held
/// transfer has none; a refused one does.
exit_status: Option<u32>,
/// Whether the channel request itself was accepted.
accepted: bool,
}
/// Exec `command` and read until the first data band (or the channel ends).
///
/// **Does not drain.** A held transfer never ends, so draining would hang; the
/// channel is returned so the caller decides when the transfer stops.
async fn start_exec(session: &russh::client::Handle<TrustingClient>, command: &str) -> Started {
let mut channel = session
.channel_open_session()
.await
.expect("open a session channel");
channel.exec(true, command).await.expect("send the exec");
let mut first_data = Vec::new();
let mut exit_status = None;
let mut accepted = false;
while let Some(msg) = channel.wait().await {
match msg {
russh::ChannelMsg::Success => accepted = true,
russh::ChannelMsg::Failure => break,
russh::ChannelMsg::Data { ref data } => {
first_data.extend_from_slice(data);
// A refused transfer sends its ERR and then an exit status, so
// keep reading for a moment to pick the status up; an admitted
// one has written its banner and is now holding, so this loop
// must not wait for anything more.
break;
}
russh::ChannelMsg::ExitStatus { exit_status: code } => exit_status = Some(code),
_ => {}
}
}
if !first_data.is_empty() && exit_status.is_none() {
// Give a refusal's exit-status message the one extra turn it needs,
// bounded so an admitted transfer does not stall the test.
if let Ok(Some(russh::ChannelMsg::ExitStatus { exit_status: code })) =
tokio::time::timeout(Duration::from_millis(250), channel.wait()).await
{
exit_status = Some(code);
}
}
Started {
channel,
first_data: String::from_utf8_lossy(&first_data).into_owned(),
exit_status,
accepted,
}
}
fn config_with(max_concurrent: Option<usize>, max_per_principal: Option<usize>) -> SshConfig {
SshConfig {
max_concurrent_transfers: max_concurrent,
max_transfers_per_principal: max_per_principal,
..SshConfig::ephemeral_loopback()
}
}
const UPLOAD: &str = "git-upload-pack 'nordisk/osm-katana.git'";
/// Wait for `predicate` to hold, or give up. Used only to let a *released*
/// transfer finish; nothing here waits for a refusal, which is synchronous.
async fn eventually(mut predicate: impl FnMut() -> bool) -> bool {
for _ in 0..200 {
if predicate() {
return true;
}
tokio::time::sleep(Duration::from_millis(10)).await;
}
predicate()
}
#[tokio::test(flavor = "multi_thread")]
async fn the_transfer_past_the_limit_is_refused_told_and_counted() {
let key = mint_key();
let (transport, hold) = BannerTransport::holding();
let fixture = start_with_config(
plane_with_key(&key, Access::Read).control,
config_with(Some(2), None),
transport,
)
.await;
let addr = fixture.server.local_addr();
let session = authenticate(addr, &key, "git")
.await
.expect("connect")
.expect("the key is registered");
let first = start_exec(&session, UPLOAD).await;
let second = start_exec(&session, UPLOAD).await;
assert!(
first.first_data.contains("git-upload-pack")
&& second.first_data.contains("git-upload-pack"),
"the two transfers under the limit must actually be served: {:?} / {:?}",
first.first_data,
second.first_data
);
assert_eq!(
(first.exit_status, second.exit_status),
(None, None),
"both are still in flight, so neither has reported a status yet"
);
// The one past the limit.
let third = start_exec(&session, UPLOAD).await;
assert!(
third.accepted,
"the channel request itself is accepted. The refusal is spoken in git's own protocol, \
because a bare channel failure makes git print `Could not read from remote repository` \
and the operator learns nothing"
);
assert!(
third.first_data.starts_with("00")
&& third.first_data.contains("ERR ")
&& third.first_data.contains("concurrent git transfers"),
"the refused client must be TOLD, as a pkt-line ERR, and told what happened; got {:?}",
third.first_data
);
assert!(
third.first_data.contains("try again"),
"a transient refusal that does not say `try again` reads as permanent: {:?}",
third.first_data
);
assert_eq!(
third.exit_status,
Some(EXIT_BUSY),
"EX_TEMPFAIL (75), not the 128 that means `you may not`. A retry script has to be able \
to tell a busy server from a permission error"
);
assert!(
!third.first_data.contains("git-upload-pack "),
"the refused transfer must not have been SERVED as well as refused: {:?}",
third.first_data
);
// …and it is counted. This is the assertion the ceiling is really about: the
// pre-fix server also failed to serve past its ceiling. It just never said
// so anywhere an operator could read.
let stats = fixture.server.admission();
assert_eq!(stats.admitted, 2, "two transfers were let through");
assert_eq!(
stats.refused_at_capacity, 1,
"the refusal MUST move a counter; a silent ceiling is the bug"
);
assert_eq!(
stats.refused_per_principal, 0,
"no per-principal limit was set"
);
assert_eq!(stats.in_flight, 2);
assert_eq!(
stats.peak_in_flight, 2,
"peak_in_flight is how an operator sees they were AT the ceiling"
);
assert_eq!(
fixture.transport.served().len(),
2,
"the transport is the applied output: it must have been handed exactly the two admitted \
transfers, not three"
);
// A slot that is released is reusable: a limit, not a budget.
hold.release();
drop(first);
drop(second);
drop(third);
assert!(
eventually(|| fixture.server.admission().in_flight == 0).await,
"released transfers must give their slots back; still {} in flight",
fixture.server.admission().in_flight
);
// The hold is released, so this one is served and finishes on its own.
let fourth = start_exec(&session, UPLOAD).await;
assert!(
fourth.first_data.contains("git-upload-pack"),
"a freed slot must admit the next transfer; got {:?}",
fourth.first_data
);
assert_eq!(
fixture.server.admission().admitted,
3,
"the fourth transfer was served without moving the admitted counter, so the gate is no \
longer metering what it lets through"
);
drop(fourth);
fixture.server.shutdown("done").await.expect("shutdown");
}
/// The other half: a server-wide limit alone still lets one client
/// take the whole server.
///
/// **Direction B of the max-min bar, over the real wire.** The greedy principal
/// is held to its *share* of a contended server (`4 / 2 = 2`), not to the
/// configured floor of 1, and the slot it does not get is there for the quiet
/// one. `polite` transfers first so that there is a second active identity —
/// fairness has nothing to divide between one claimant, which is
/// `a_lone_principal_is_not_throttled_against_an_empty_server` below.
#[tokio::test(flavor = "multi_thread")]
async fn one_principal_cannot_starve_the_others() {
let greedy = mint_key();
let polite = mint_key();
let plane = Plane::empty()
.with_the_repo()
.with_principal("greedy")
.with_key("greedy", &greedy)
.with_grant("greedy", Access::Read)
.with_principal("polite")
.with_key("polite", &polite)
.with_grant("polite", Access::Read);
let (transport, hold) = BannerTransport::holding();
// Room for four on the server, a guaranteed floor of one each. Three in
// flight is 75% and puts the server into contention; the share is then two.
let fixture = start_with_config(plane.control, config_with(Some(4), Some(1)), transport).await;
let addr = fixture.server.local_addr();
let greedy_session = authenticate(addr, &greedy, "git")
.await
.expect("connect")
.expect("registered");
let polite_session = authenticate(addr, &polite, "git")
.await
.expect("connect")
.expect("registered");
let quiet = start_exec(&polite_session, UPLOAD).await;
assert!(
quiet.first_data.contains("git-upload-pack"),
"the quiet principal's first transfer must be served: {:?}",
quiet.first_data
);
let held = start_exec(&greedy_session, UPLOAD).await;
assert!(held.first_data.contains("git-upload-pack"));
let held2 = start_exec(&greedy_session, UPLOAD).await;
assert!(
held2.first_data.contains("git-upload-pack"),
"the second transfer is still within greedy's share of a two-identity server (4/2), and \
refusing it here would be the old wall: {:?}",
held2.first_data
);
let refused = start_exec(&greedy_session, UPLOAD).await;
assert!(
refused.first_data.contains("per-principal limit"),
"the transfer past the share must be refused for the PER-PRINCIPAL reason, not the \
server-wide one; got {:?}",
refused.first_data
);
assert!(
refused.first_data.contains("2 identities"),
"the refusal must say it is a SHARE of a contended server and how many identities it is \
divided between; `you have hit the limit` with no number to reconcile against the \
configured 1 is what sends an operator to the wrong knob: {:?}",
refused.first_data
);
assert_eq!(refused.exit_status, Some(EXIT_BUSY));
// The whole point: the slot greedy did not get is still there for somebody
// else, and the server was never full.
let other = start_exec(&polite_session, UPLOAD).await;
assert!(
other.first_data.contains("git-upload-pack"),
"a different principal must still be served while the greedy one is at its share; got \
{:?}",
other.first_data
);
let stats = fixture.server.admission();
assert_eq!(stats.refused_per_principal, 1);
assert_eq!(
stats.refused_at_capacity, 0,
"the server-wide limit was 4 and only 3 were in flight, so attributing this refusal to \
capacity would send an operator to raise the wrong number"
);
assert_eq!(stats.admitted, 4);
assert_eq!(stats.in_flight, 4);
assert_eq!(
fixture.transport.served().len(),
4,
"the transport is the applied output: four transfers reached it, not three and not five"
);
hold.release();
drop((quiet, held, held2, refused, other));
fixture.server.shutdown("done").await.expect("shutdown");
}
/// **Direction A of the max-min bar, over the real wire.** One key, an empty
/// server, and every slot the server advertises is reachable.
///
/// This is the measured bug: on the reference desktop with the shipped defaults, one principal
/// driving 64 clients was served 32 and refused 32 while 224 slots sat idle,
/// because `max_transfers_per_principal` bound with nobody to be fair to. Here
/// the floor is **one** and eight transfers are served, so a gate that still
/// treated the floor as a wall would refuse seven of them.
///
/// The server-wide limit is asserted from the other side in the same test, so
/// this cannot pass on a gate that simply lost both limits.
#[tokio::test(flavor = "multi_thread")]
async fn a_lone_principal_is_not_throttled_against_an_empty_server() {
const SERVER_WIDE: usize = 8;
const FLOOR: usize = 1;
let key = mint_key();
let (transport, hold) = BannerTransport::holding();
let fixture = start_with_config(
plane_with_key(&key, Access::Read).control,
config_with(Some(SERVER_WIDE), Some(FLOOR)),
transport,
)
.await;
let addr = fixture.server.local_addr();
let session = authenticate(addr, &key, "git")
.await
.expect("connect")
.expect("registered");
let mut started = Vec::new();
for i in 0..SERVER_WIDE {
let s = start_exec(&session, UPLOAD).await;
assert!(
s.first_data.contains("git-upload-pack"),
"transfer {i} of {SERVER_WIDE} was refused on an EMPTY server holding ONE identity, \
with a per-principal floor of {FLOOR}: {:?}. That is the 32-of-256 bug at a smaller \
scale — the floor is a guarantee, not a ceiling",
s.first_data
);
assert_eq!(
s.exit_status, None,
"transfer {i} ended when it should be held"
);
started.push(s);
}
let stats = fixture.server.admission();
assert_eq!(stats.admitted, SERVER_WIDE as u64);
assert_eq!(
stats.refused_per_principal, 0,
"not one of these may be refused for fairness: there was nobody to be fair to"
);
assert_eq!(stats.peak_in_flight, SERVER_WIDE as u64);
assert_eq!(
fixture.transport.served().len(),
SERVER_WIDE,
"the transport is the applied output: all {SERVER_WIDE} reached it"
);
// The server-wide gate is still a gate, and it is the one that speaks now.
let past = start_exec(&session, UPLOAD).await;
assert!(
past.first_data.contains("concurrent git transfers"),
"past the server-wide limit the refusal must be the SERVER-WIDE one; a per-principal \
refusal here would mean the share was doing the refusing after all: {:?}",
past.first_data
);
assert_eq!(past.exit_status, Some(EXIT_BUSY));
assert_eq!(fixture.server.admission().refused_at_capacity, 1);
assert_eq!(fixture.server.admission().refused_per_principal, 0);
hold.release();
drop(started);
drop(past);
fixture.server.shutdown("done").await.expect("shutdown");
}
/// The pinned mode, over the wire: an operator who needs the old unconditional
/// wall still has it, and it binds on an empty server.
///
/// Without this the new `per_principal_limit_is_hard` field could be dead —
/// present in the struct, threaded nowhere, and nothing would say so.
#[tokio::test(flavor = "multi_thread")]
async fn a_pinned_per_principal_cap_still_binds_on_an_idle_server() {
let key = mint_key();
let (transport, hold) = BannerTransport::holding();
let config = SshConfig {
per_principal_limit_is_hard: true,
..config_with(Some(8), Some(1))
};
let fixture = start_with_config(
plane_with_key(&key, Access::Read).control,
config,
transport,
)
.await;
let addr = fixture.server.local_addr();
let session = authenticate(addr, &key, "git")
.await
.expect("connect")
.expect("registered");
let held = start_exec(&session, UPLOAD).await;
assert!(held.first_data.contains("git-upload-pack"));
let refused = start_exec(&session, UPLOAD).await;
assert!(
refused.first_data.contains("per-principal limit"),
"a pinned cap must refuse the second transfer even though seven slots are free; got {:?}",
refused.first_data
);
assert!(
!refused.first_data.contains("identities are using it"),
"a pinned cap is not a share and must not be explained as one: {:?}",
refused.first_data
);
assert_eq!(refused.exit_status, Some(EXIT_BUSY));
assert_eq!(fixture.server.admission().refused_per_principal, 1);
assert_eq!(
fixture.transport.served().len(),
1,
"exactly one transfer reached the transport"
);
hold.release();
drop((held, refused));
fixture.server.shutdown("done").await.expect("shutdown");
}
/// **The guard's own red.** With the limit switched off, the very same probe
/// reports the unlimited behaviour: the third transfer is served and nothing
/// is counted.
///
/// Without this, `the_transfer_past_the_limit_is_refused_told_and_counted` could
/// pass for a reason that has nothing to do with the gate; a server that
/// refused the third channel for any reason at all would satisfy it. This pins
/// that the difference is the *limit*.
#[tokio::test(flavor = "multi_thread")]
async fn the_limit_is_the_thing_being_tested() {
let key = mint_key();
let (transport, hold) = BannerTransport::holding();
let fixture = start_with_config(
plane_with_key(&key, Access::Read).control,
config_with(None, None),
transport,
)
.await;
let addr = fixture.server.local_addr();
let session = authenticate(addr, &key, "git")
.await
.expect("connect")
.expect("registered");
let mut started = Vec::new();
for _ in 0..3 {
started.push(start_exec(&session, UPLOAD).await);
}
for (i, s) in started.iter().enumerate() {
assert!(
s.first_data.contains("git-upload-pack"),
"transfer {i} was refused by a server with NO limit configured: {:?}",
s.first_data
);
assert_eq!(
s.exit_status, None,
"transfer {i} ended when it should be held"
);
}
let stats = fixture.server.admission();
assert_eq!(stats.admitted, 3);
assert_eq!(
stats.refused(),
0,
"an unlimited gate must refuse nothing, or the limit test above is measuring something \
other than the limit"
);
assert_eq!(stats.in_flight, 3);
hold.release();
drop(started);
fixture.server.shutdown("done").await.expect("shutdown");
}
/// A refused transfer must not leave the slot it was refused for held.
///
/// The server-wide permit is taken before the per-principal check; if the
/// refusal path forgot to return it, a client hammering its own limit would
/// drain the server one refusal at a time, a denial of service *caused by* the
/// gate that was meant to prevent one.
#[tokio::test(flavor = "multi_thread")]
async fn a_refusal_does_not_consume_the_slot_it_was_refused_for() {
let greedy = mint_key();
let polite = mint_key();
let plane = Plane::empty()
.with_the_repo()
.with_principal("greedy")
.with_key("greedy", &greedy)
.with_grant("greedy", Access::Read)
.with_principal("polite")
.with_key("polite", &polite)
.with_grant("polite", Access::Read);
let (transport, hold) = BannerTransport::holding();
// Two identities, so the refusals below are the SHIPPED fair-share path and
// not a pinned cap: a leak test that only covers the escape hatch would
// leave the default path untested.
let fixture = start_with_config(plane.control, config_with(Some(4), Some(1)), transport).await;
let addr = fixture.server.local_addr();
let greedy_session = authenticate(addr, &greedy, "git")
.await
.expect("connect")
.expect("registered");
let polite_session = authenticate(addr, &polite, "git")
.await
.expect("connect")
.expect("registered");
let quiet = start_exec(&polite_session, UPLOAD).await;
assert!(quiet.first_data.contains("git-upload-pack"));
let held = start_exec(&greedy_session, UPLOAD).await;
let held2 = start_exec(&greedy_session, UPLOAD).await;
assert!(held.first_data.contains("git-upload-pack"));
assert!(held2.first_data.contains("git-upload-pack"));
let mut refusals = Vec::new();
for _ in 0..12 {
refusals.push(start_exec(&greedy_session, UPLOAD).await);
}
for r in &refusals {
assert_eq!(r.exit_status, Some(EXIT_BUSY));
assert!(
r.first_data.contains("per-principal limit"),
"refused for the wrong reason: {:?}",
r.first_data
);
}
let stats = fixture.server.admission();
assert_eq!(stats.refused_per_principal, 12);
assert_eq!(
stats.in_flight, 3,
"twelve refusals must leave exactly the three real transfers in flight, not fifteen"
);
// …and the slot they were refused for is genuinely free for somebody else.
let survivor = start_exec(&polite_session, UPLOAD).await;
assert!(
survivor.first_data.contains("git-upload-pack"),
"the fourth slot did not survive twelve refusals; the gate is eating what it refuses: \
{:?}",
survivor.first_data
);
hold.release();
drop((quiet, held, held2, survivor));
drop(refusals);
fixture.server.shutdown("done").await.expect("shutdown");
}
/// Authorisation is decided **before** admission, and the log/counter reflect
/// that.
///
/// If the gate ran first, a full server would answer "busy" to a principal that
/// may not touch the repository at all, which is both a leak (the repository
/// exists) and a lie (their next attempt will not work either). It would also
/// inflate the refusal counter with requests that were never going to be served,
/// making the one number an operator uses to size the limit useless.
#[tokio::test(flavor = "multi_thread")]
async fn a_denied_request_is_denied_not_reported_busy() {
let key: PrivateKey = mint_key();
let stranger = mint_key();
let plane = Plane::empty()
.with_the_repo()
.with_principal("principal-1")
.with_key("principal-1", &key)
.with_grant("principal-1", Access::Read)
.with_principal("stranger")
.with_key("stranger", &stranger);
let (transport, hold) = BannerTransport::holding();
let fixture = start_with_config(plane.control, config_with(Some(1), None), transport).await;
let addr = fixture.server.local_addr();
let session = authenticate(addr, &key, "git")
.await
.expect("connect")
.expect("registered");
let held = start_exec(&session, UPLOAD).await;
assert!(held.first_data.contains("git-upload-pack"));
// The server is now FULL, which is the interesting condition.
let stranger_session = authenticate(addr, &stranger, "git")
.await
.expect("connect")
.expect("registered");
let denied = start_exec(&stranger_session, UPLOAD).await;
assert!(
denied
.first_data
.contains("repository not found, or you do not have access"),
"a principal with no grant must be DENIED even when the server is full; got {:?}",
denied.first_data
);
assert_eq!(
fixture.server.admission().refused_at_capacity,
0,
"an unauthorised request must not be counted as a capacity refusal. That number is what \
an operator sizes the limit from"
);
hold.release();
drop((held, denied));
fixture.server.shutdown("done").await.expect("shutdown");
}
/// **The governor is actually wired to the real transfer path**, and it is the
/// sojourn — not a flag — that proves it.
///
/// The CoDel controller in `gunnar_ssh::governor` can only see what
/// `AdmissionGuard::entered_service` files for it, and that call lives in
/// `server.rs`, one line inside the spawned transfer task. Delete it and every
/// unit test in `governor.rs` still passes, every other test in this file still
/// passes, and the shipped server has a delay governor that never receives a
/// single sample: it grows to the top of its band on start-up and stays there
/// forever. A gate that looks like it works and does not is the exact failure
/// the unbounded listener was about, one level up.
///
/// So this asserts the **applied output** of the wiring: transfers that really
/// went through russh, a real exec dispatch and a real transport must leave
/// `unmeasured_sojourn` at zero.
///
/// # Seen red
///
/// Measured 2026-08-14 by commenting out `admitted.entered_service();` in
/// `Handler::exec_request` and re-running this file:
///
/// ```text
/// 3 of 3 transfers reached the transport without ever filing a sojourn
/// sample, so the CoDel governor is receiving nothing and its limit is 16
/// and cannot move
/// ```
#[tokio::test(flavor = "multi_thread")]
async fn every_served_transfer_files_a_sojourn_sample() {
let key = mint_key();
let (transport, hold) = BannerTransport::holding();
let fixture = start_with_config(
plane_with_key(&key, Access::Read).control,
config_with(Some(16), None),
transport,
)
.await;
let session = authenticate(fixture.server.local_addr(), &key, "git")
.await
.expect("connect")
.expect("the key is registered");
let started = vec![
start_exec(&session, UPLOAD).await,
start_exec(&session, UPLOAD).await,
start_exec(&session, UPLOAD).await,
];
assert!(
started
.iter()
.all(|s| s.first_data.contains("git-upload-pack")),
"all three must actually be served, or this proves nothing about the wiring"
);
assert_eq!(fixture.server.admission().admitted, 3);
hold.release();
drop(started);
assert!(
eventually(|| fixture.server.admission().in_flight == 0).await,
"the transfers must finish so their guards drop and any unmeasured one is counted"
);
let stats = fixture.server.admission();
assert_eq!(
stats.unmeasured_sojourn, 0,
"{} of 3 transfers reached the transport without ever filing a sojourn sample, so the \
CoDel governor is receiving nothing and its limit is {} and cannot move",
stats.unmeasured_sojourn, stats.limit
);
assert_eq!(
stats.limit, 16,
"the governed limit is the configured ceiling here (band [10, 16], starting at the top)"
);
assert_eq!(
stats.reserved_bytes, 0,
"every finished transfer must have given its memory reservation back"
);
fixture.server.shutdown("done").await.expect("shutdown");
}