nordisk/gunnar

crates/gunnar-ssh/tests/keyed_queue.rs

//! **A CI burst on one key is queued and served; a stranger's burst is held to
//! its ceiling and told to retry.**
//!
//! gunnar.rs, 2026-10-25: every lane pushes and fetches as ONE key from a few
//! hosts, many at once, and a git-lfs push opens up to eight pure-SSH
//! `git-lfs-transfer` sessions of its own. On a 4-core main the governed limit
//! moves down under load, and a keyed session refused after the old 2 s wait
//! cannot be retried by git-lfs: the push fails. So:
//!
//! * 40 parallel KEYED transfers against a gate of 8 are all admitted — the
//!   ones past the limit are queued up to the keyed wait, not refused;
//! * 40 parallel ANONYMOUS transfers against the anonymous ceiling of 5 get 5
//!   slots, and the other 35 are refused at once with the transient words
//!   (EX_TEMPFAIL on the wire, `try again` in the sentence — the SSH door's
//!   429 + Retry-After) and counted under their own reason;
//! * the control arm: the same keyed burst WITHOUT the keyed bound loses
//!   transfers, which is the behaviour this replaced.
//!
//! This is the gate every git and LFS transfer over SSH takes (`admit_within`,
//! called from the exec handler), so it is tested at the gate, where the burst
//! is deterministic.

use std::sync::atomic::{AtomicUsize, Ordering};
use std::sync::Arc;
use std::time::Duration;

use gunnar_ssh::governor::OpClass;
use gunnar_ssh::gunnar_core::ANONYMOUS_PRINCIPAL_ID;
use gunnar_ssh::{Admission, Refusal};

fn gate(limit: usize, keyed_wait: Option<Duration>) -> Arc<Admission> {
    Arc::new(
        Admission::new(Some(limit), Some(8))
            .with_anonymous_ceiling(Some(5))
            .with_wait(Duration::from_millis(50))
            .with_keyed_wait(keyed_wait),
    )
}

/// Run `n` transfers for `principal` at once, each holding its slot for `hold`.
/// Answers (admitted, refusals).
async fn burst(
    gate: &Arc<Admission>,
    principal: &'static str,
    n: usize,
    hold: Duration,
) -> (usize, Vec<Refusal>) {
    let admitted = Arc::new(AtomicUsize::new(0));
    let mut tasks = tokio::task::JoinSet::new();
    for _ in 0..n {
        let gate = Arc::clone(gate);
        let admitted = Arc::clone(&admitted);
        tasks.spawn(async move {
            match gate
                .admit_within(principal, "nordisk", OpClass::Fetch)
                .await
            {
                Ok((guard, _waited)) => {
                    admitted.fetch_add(1, Ordering::SeqCst);
                    tokio::time::sleep(hold).await;
                    drop(guard);
                    None
                }
                Err(refusal) => Some(refusal),
            }
        });
    }
    let mut refusals = Vec::new();
    while let Some(done) = tasks.join_next().await {
        if let Some(r) = done.expect("a transfer task") {
            refusals.push(r);
        }
    }
    (admitted.load(Ordering::SeqCst), refusals)
}

#[tokio::test(flavor = "multi_thread", worker_threads = 8)]
async fn forty_keyed_transfers_are_all_served_and_forty_strangers_get_their_ceiling() {
    // ── 40 keyed, one key, a gate of 8: all served ─────────────────────────
    let keyed = gate(8, Some(Duration::from_secs(20)));
    let (admitted, refusals) = burst(&keyed, "admin", 40, Duration::from_millis(150)).await;
    assert!(
        refusals.is_empty(),
        "a keyed CI burst was refused although it may queue for 20 s and every slot frees in \
         150 ms: {} refused, first {:?}",
        refusals.len(),
        refusals.first()
    );
    assert_eq!(admitted, 40);
    assert_eq!(keyed.stats().refused(), 0);

    // ── 40 anonymous: 5 slots, 35 told to retry, at once ───────────────────
    // A roomy box (64), so what refuses is the anonymous ceiling itself and
    // not the keyed reserve of a full one.
    let open = gate(64, Some(Duration::from_secs(20)));
    let started = std::time::Instant::now();
    let (admitted, refusals) = burst(
        &open,
        ANONYMOUS_PRINCIPAL_ID,
        40,
        Duration::from_millis(1500),
    )
    .await;
    assert_eq!(
        admitted, 5,
        "a stranger's ceiling is 5 however idle the box is"
    );
    assert_eq!(refusals.len(), 35);
    for r in &refusals {
        assert!(
            matches!(r, Refusal::AnonymousAtCapacity { limit: 5 }),
            "{r:?}"
        );
        assert!(
            r.client_message().contains("try again"),
            "transient words: {}",
            r.client_message()
        );
    }
    assert_eq!(
        open.stats().refused_anonymous,
        35,
        "counted under its own reason"
    );
    // The refusals came at once (never parked for the keyed bound); only the 5
    // admitted ones held for 1.5 s.
    assert!(
        started.elapsed() < Duration::from_secs(5),
        "{:?}",
        started.elapsed()
    );

    // ── control: the same keyed burst without the keyed bound loses work ───
    let old = gate(8, None);
    let (admitted, refusals) = burst(&old, "admin", 40, Duration::from_millis(150)).await;
    assert!(
        !refusals.is_empty() && admitted < 40,
        "the control arm (2026-10-25 behaviour: the 50 ms wait for keys too) must refuse some of \
         the burst, or this test proves nothing: admitted {admitted}"
    );
}