gunnar-server 3.0.0

//! **What one usage poll costs main, at N tenants** (2026-10-06, the load test's
//! finding: monetize's poller asks `Usage.Read` once per tenant every 300 s, and at
//! 1,000 tenants that was ~25–30 s of 1.4 cores on main per period).
//!
//! A bench, not a gate: `#[ignore]`d, run by hand with
//! `GUNNAR_USAGE_BENCH_N=1000 cargo test -p gunnar-server --test usage_poll_scale -- --ignored --nocapture`.
//! It seeds N namespaces with one small repository each on a real `ServerStores`
//! and times the per-tenant path the poller drives today — what `Usage.Read`
//! does on the plane (`usage(Some(ns))` + `repo_usage(ns)`) — and prints the wall
//! and process CPU time of one full pass. The gated twin of this file, the one
//! that goes red and green, is `usage_batch.rs`.

use std::sync::Arc;
use std::time::Instant;

use gunnar_core::{NewRepo, RepoRef};
use gunnar_grpc::ControlPlane as _;
use gunnar_server::control_plane::CoreControlPlane;
use gunnar_server::stores::ServerStores;
use gunnar_store::conformance::pack::PackBuilder;

/// User + system CPU of this process so far, in milliseconds (`/proc/self/stat`).
pub fn cpu_ms() -> u64 {
    let stat = std::fs::read_to_string("/proc/self/stat").unwrap_or_default();
    // Fields after the `)` of the command name: state is field 3, utime 14, stime 15.
    let rest = stat.rsplit_once(')').map(|(_, r)| r).unwrap_or("");
    let f: Vec<&str> = rest.split_whitespace().collect();
    let ticks = |i: usize| f.get(i).and_then(|s| s.parse::<u64>().ok()).unwrap_or(0);
    // `rest` starts at field 3, so field k is f[k - 3].
    (ticks(14 - 3) + ticks(15 - 3)) * 10
}

/// One namespace with one repository holding one small pack, store closed.
pub fn seed_tenant(stores: &ServerStores, ns: &str) {
    stores.catalog().create_namespace(ns).expect("namespace");
    let path = RepoRef::new(ns, "main").expect("legal");
    let repo = stores
        .create_repo(NewRepo::new(path.clone()), false)
        .expect("create");
    let handle = stores.repo_store(&repo).expect("open");
    let mut pb = PackBuilder::new(gix_hash::Kind::Sha256);
    let mut body = format!("tenant {ns}\n").into_bytes();
    body.resize(2_000, b'x');
    pb.object(gix_object::Kind::Blob, body);
    let fixture = pb.build().expect("pack");
    handle.serve().put_pack(&fixture.bytes).expect("put_pack");
    drop(handle);
    stores.close_repo(&path);
}

#[test]
#[ignore = "a bench: run by hand with GUNNAR_USAGE_BENCH_N"]
fn one_poll_of_per_tenant_usage_reads() {
    let n: usize = std::env::var("GUNNAR_USAGE_BENCH_N")
        .ok()
        .and_then(|v| v.parse().ok())
        .unwrap_or(1_000);
    let dir = tempfile::tempdir().expect("tempdir");
    let stores = Arc::new(ServerStores::open(dir.path()).expect("stores"));
    let t = Instant::now();
    let names: Vec<String> = (0..n).map(|i| format!("t{i:05}")).collect();
    for ns in &names {
        seed_tenant(&stores, ns);
    }
    eprintln!("seeded {n} tenants in {:?}", t.elapsed());
    let plane = CoreControlPlane::new(Arc::clone(&stores));
    // Warm once (the catalog document, the flow document).
    let _ = plane.usage(Some(&names[0])).expect("warm");

    let (wall, cpu0) = (Instant::now(), cpu_ms());
    for ns in &names {
        let row = plane
            .usage(Some(ns))
            .expect("Usage.Read")
            .pop()
            .expect("row");
        assert_eq!(row.repos, 1);
        let _ = plane.repo_usage(ns).expect("Usage.Read repos");
    }
    let (wall, cpu) = (wall.elapsed(), cpu_ms() - cpu0);
    eprintln!(
        "BENCH per-tenant Usage.Read N={n}: calls={n} wall={:.1} ms cpu={cpu} ms per-call={:.3} ms",
        wall.as_secs_f64() * 1e3,
        wall.as_secs_f64() * 1e3 / n as f64
    );

    // ── the batched read: UsageBatch.ReadMany, 500 a page ─────────────────
    let pass = |since: &str| {
        let (wall, cpu0) = (Instant::now(), cpu_ms());
        let (mut token, mut calls, mut rows, mut cursor) =
            (String::new(), 0usize, 0usize, String::new());
        loop {
            let page = plane.usage_page(&token, 500, since).expect("ReadMany");
            calls += 1;
            rows += page.usages.len();
            if cursor.is_empty() {
                cursor = page.cursor.clone();
            }
            if page.next_page_token.is_empty() {
                break;
            }
            token = page.next_page_token;
        }
        (calls, rows, cursor, wall.elapsed(), cpu_ms() - cpu0)
    };
    let (calls, rows, cursor, wall, cpu) = pass("");
    eprintln!("BENCH ReadMany full (cold ledger) N={n}: calls={calls} rows={rows} wall={:.1} ms cpu={cpu} ms", wall.as_secs_f64() * 1e3);
    let (calls, rows, cursor, wall, cpu) = {
        let _ = pass(&cursor);
        pass(&cursor)
    };
    eprintln!("BENCH ReadMany changed-since, nothing moved N={n}: calls={calls} rows={rows} wall={:.1} ms cpu={cpu} ms", wall.as_secs_f64() * 1e3);
    // One percent of the tenants push.
    for ns in names.iter().step_by(100) {
        let path = RepoRef::new(ns, "main").expect("legal");
        let repo = stores.catalog().get(&path).expect("read").expect("row");
        let handle = stores.repo_store(&repo).expect("open");
        let mut pb = PackBuilder::new(gix_hash::Kind::Sha256);
        pb.object(gix_object::Kind::Blob, format!("more {ns}").into_bytes());
        handle
            .serve()
            .put_pack(&pb.build().expect("pack").bytes)
            .expect("put_pack");
        drop(handle);
        stores.close_repo(&path);
        stores.flow().record(
            ns,
            gunnar_core::now_ms(),
            &gunnar_core::Transfer {
                bytes_in: 100,
                pushed: true,
                ..Default::default()
            },
        );
    }
    let (calls, rows, _, wall, cpu) = pass(&cursor);
    eprintln!("BENCH ReadMany changed-since, 1% moved N={n}: calls={calls} rows={rows} wall={:.1} ms cpu={cpu} ms", wall.as_secs_f64() * 1e3);
    let (calls, rows, _, wall, cpu) = pass("");
    eprintln!("BENCH ReadMany full (warm ledger) N={n}: calls={calls} rows={rows} wall={:.1} ms cpu={cpu} ms", wall.as_secs_f64() * 1e3);
}