gunnar-ssh 3.0.0

//! **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());
}