gunnar-server 3.0.0

//! **`UsageBatch.ReadMany`: many namespaces in one call, only what changed —
//! and the usage ledger behind it always equals a full recount** (2026-10-06).
//!
//! The load test on the UpCloud mockup found monetize's poller asking
//! `Usage.Read` once per tenant every 300 s, ~25–30 s of 1.4 cores on main per
//! period at 1,000 tenants and growing faster than the tenant count (every read
//! listed the whole catalog and walked the whole repository root). `ReadMany`
//! answers a page of namespaces out of `gunnar_server::usage_ledger`, which
//! re-measures only what was marked as changed.
//!
//! A ledger is only worth having if it is never WRONG, so the heart of this
//! file is a property test: random pushes, new repositories and namespaces, soft
//! deletes, purges, undeletes, LFS uploads, gc and a data-set growth's
//! suspend/resume, in random order over many seeds — and after every step the
//! ledger's answer (a full `ReadMany`, and the changed-since read folded onto
//! the caller's last copy) must equal `Usage.ReadAll`'s full recount (the walk
//! that has always been the definition of the meters), meter for meter.

use std::collections::BTreeMap;
use std::sync::Arc;

use gunnar_core::{NewRepo, RepoRef, Transfer};
use gunnar_grpc::seam::DeleteOptions;
use gunnar_grpc::ControlPlane as _;
use gunnar_lfs::LfsStore as _;
use gunnar_proto::{NamespaceUsage, ReadManyUsageResponse, RepoPath};
use gunnar_server::control_plane::CoreControlPlane;
use gunnar_server::stores::ServerStores;
use gunnar_store::conformance::pack::PackBuilder;

/// A small deterministic PRNG (xorshift64*): the property test's seeds are
/// reproducible without a dependency.
struct Rng(u64);
impl Rng {
    fn next(&mut self) -> u64 {
        let mut x = self.0;
        x ^= x >> 12;
        x ^= x << 25;
        x ^= x >> 27;
        self.0 = x;
        x.wrapping_mul(0x2545_F491_4F6C_DD1D)
    }
    fn below(&mut self, n: usize) -> usize {
        (self.next() % n.max(1) as u64) as usize
    }
}

fn pack(seed: u64, blobs: usize, size: usize) -> Vec<u8> {
    let mut pb = PackBuilder::new(gix_hash::Kind::Sha256);
    for i in 0..blobs {
        let mut body = format!("{seed}:{i}\n").into_bytes();
        body.resize(size, b'a' + (i % 26) as u8);
        pb.object(gix_object::Kind::Blob, body);
    }
    pb.build().expect("pack").bytes
}

/// What a push does to the store: open, write a pack, close — and the
/// transfer the transport records at its end.
fn push(stores: &ServerStores, path: &RepoRef, seed: u64, blobs: usize, size: usize) {
    let repo = stores.catalog().get(path).expect("read").expect("row");
    let handle = stores.repo_store(&repo).expect("open");
    handle
        .serve()
        .put_pack(&pack(seed, blobs, size))
        .expect("put_pack");
    drop(handle);
    stores.close_repo(path);
    stores.flow().record(
        path.namespace(),
        gunnar_core::now_ms(),
        &Transfer {
            bytes_in: (blobs * size) as u64,
            pushed: true,
            principal: "SHA256:t".into(),
            ..Transfer::default()
        },
    );
}

/// **Let every store close.** The engine keeps a store open until the last
/// handle lets go (a background index drain may hold one a moment after
/// `close_repo`), and only a CLOSED store's bytes are final — it may trim what
/// it had reserved. The recount and the ledger are compared at rest.
fn quiesce(stores: &ServerStores) {
    // What the handle cache would do in its own time: let every handle go
    // (an undelete, a create, leave theirs cached).
    for repo in stores.catalog().list(None).expect("list") {
        stores.close_repo(&repo.path());
    }
    let deadline = std::time::Instant::now() + std::time::Duration::from_secs(30);
    while !stores.engine().open_object_counts().is_empty() {
        assert!(
            std::time::Instant::now() < deadline,
            "a store never closed: {:?}",
            stores.engine().open_object_counts()
        );
        std::thread::sleep(std::time::Duration::from_millis(20));
    }
}

fn create(stores: &ServerStores, path: &RepoRef) {
    stores
        .create_repo(NewRepo::new(path.clone()), false)
        .expect("create");
}

/// What an LFS upload does: the object lands in the repository's LFS store and
/// the LFS session records its transfer.
fn lfs_upload(stores: &ServerStores, path: &RepoRef, seed: u64, size: usize) {
    let dir = stores.repo_dir(path).expect("dir");
    let lfs = gunnar_lfs::FsLfsStore::open_or_init(&dir).expect("lfs store");
    let mut body = format!("lfs {seed}").into_bytes();
    body.resize(size, 0x4c);
    let oid = gunnar_lfs::Oid::of(&body);
    {
        use std::io::Write as _;
        let mut w = lfs.begin_write(&oid, body.len() as u64).expect("begin");
        w.write_all(&body).expect("write");
        w.commit().expect("commit");
    }
    stores.flow().record(
        path.namespace(),
        gunnar_core::now_ms(),
        &Transfer {
            lfs_bytes_in: size as u64,
            bytes_in: size as u64,
            principal: "SHA256:t".into(),
            ..Transfer::default()
        },
    );
}

fn gc(stores: &ServerStores, path: &RepoRef) {
    let repo = stores.catalog().get(path).expect("read").expect("row");
    let handle = stores.repo_store(&repo).expect("open");
    let _ = handle.serve().gc();
    drop(handle);
    stores.close_repo(path);
}

fn delete(plane: &CoreControlPlane, path: &RepoRef, purge: bool) {
    plane
        .delete_repo(
            &RepoPath::new(path.namespace(), path.name()),
            DeleteOptions {
                purge,
                retention_days: Some(30),
                principal_id: "tester".into(),
            },
        )
        .expect("delete");
}

/// The meters a ledger keeps, out of a row: everything but the clock and the
/// live open-store objects (both readers ask the live registry for those, a
/// moment apart).
fn stock(row: &NamespaceUsage) -> (u64, u64, u64, u64, u64, u64, u64, u64, u64) {
    (
        row.pack_bytes,
        row.cache_bytes,
        row.lfs_bytes,
        row.tombstoned_bytes,
        row.repos,
        row.bytes_in_today,
        row.bytes_out_today,
        row.pushes_today,
        row.fetches_today,
    )
}

type Copy = BTreeMap<String, (u64, u64, u64, u64, u64, u64, u64, u64, u64)>;

fn recount(plane: &CoreControlPlane) -> Copy {
    plane
        .usage(None)
        .expect("Usage.ReadAll")
        .iter()
        .map(|r| (r.namespace.clone(), stock(r)))
        .collect()
}

/// Every page of one `ReadMany`, following the tokens. Returns the rows, the
/// number of calls it took, and the first page's cursor and `changed_only`.
fn read_all_pages(
    plane: &CoreControlPlane,
    page_size: usize,
    since: &str,
) -> (Vec<NamespaceUsage>, usize, String, bool) {
    let mut rows = Vec::new();
    let mut token = String::new();
    let mut calls = 0;
    let mut first: Option<(String, bool)> = None;
    loop {
        let page: ReadManyUsageResponse = plane
            .usage_page(&token, page_size, since)
            .expect("ReadMany");
        calls += 1;
        if first.is_none() {
            first = Some((page.cursor.clone(), page.changed_only));
        }
        assert!(
            page.usages.len() <= page_size,
            "a page never carries more than it was asked for"
        );
        rows.extend(page.usages);
        if page.next_page_token.is_empty() {
            break;
        }
        token = page.next_page_token;
        assert!(calls < 10_000, "the pages never end");
    }
    let (cursor, changed_only) = first.expect("one page");
    (rows, calls, cursor, changed_only)
}

fn box_with(n: usize) -> (tempfile::TempDir, Arc<ServerStores>, Vec<RepoRef>) {
    let dir = tempfile::tempdir().expect("tempdir");
    let stores = Arc::new(ServerStores::open(dir.path()).expect("stores"));
    let mut repos = Vec::new();
    for i in 0..n {
        let ns = format!("t{i:04}");
        stores.catalog().create_namespace(&ns).expect("namespace");
        let path = RepoRef::new(&ns, "main").expect("legal");
        create(&stores, &path);
        push(&stores, &path, i as u64, 1, 500);
        repos.push(path);
    }
    (dir, stores, repos)
}

/// ★ **One page per `page_size` namespaces, every namespace exactly once, the
/// same numbers as the full recount — and with the cursor, ONLY the namespaces
/// that changed.** RED against the seam's default `usage_page` (which pages the
/// full recount and never honours `changed_since`); GREEN on the ledger.
#[test]
fn read_many_pages_every_namespace_once_and_a_cursor_answers_only_what_moved() {
    let (_dir, stores, repos) = box_with(60);
    let plane = CoreControlPlane::new(Arc::clone(&stores));
    quiesce(&stores);

    let (rows, calls, cursor, changed_only) = read_all_pages(&plane, 25, "");
    assert_eq!(
        calls, 3,
        "60 namespaces at 25 a page is three calls, not sixty"
    );
    assert!(!changed_only, "no cursor: every namespace");
    let got: Copy = rows
        .iter()
        .map(|r| (r.namespace.clone(), stock(r)))
        .collect();
    assert_eq!(got.len(), 60, "every namespace exactly once");
    assert_eq!(
        got,
        recount(&plane),
        "the ledger's numbers are the full recount's"
    );
    assert!(!cursor.is_empty(), "a cursor to ask with next time");

    // Nothing moved: the changed-since read answers nothing, in one call.
    let (quiet, calls, cursor, changed_only) = read_all_pages(&plane, 25, &cursor);
    // The first read MEASURED each namespace for the first time, after its
    // cursor was taken; so the first changed-since read may answer them once.
    // The second one is the quiet one.
    let _ = quiet;
    let (quiet, calls2, cursor, changed_only2) = read_all_pages(&plane, 25, &cursor);
    assert!(changed_only && changed_only2, "the cursor is honoured");
    assert_eq!(calls2, 1, "a quiet estate is one call");
    assert!(
        quiet.is_empty(),
        "nothing moved, nothing answered: {:?}",
        quiet.iter().map(|r| &r.namespace).collect::<Vec<_>>()
    );
    let _ = calls;

    // Three namespaces move: a push, an LFS upload, a soft delete.
    push(&stores, &repos[7], 1_007, 3, 4_000);
    lfs_upload(&stores, &repos[21], 2_021, 9_999);
    delete(&plane, &repos[42], false);
    quiesce(&stores);
    let (moved, _, _, changed_only) = read_all_pages(&plane, 25, &cursor);
    assert!(changed_only);
    let names: Vec<&str> = moved.iter().map(|r| r.namespace.as_str()).collect();
    assert_eq!(
        names,
        ["t0007", "t0021", "t0042"],
        "exactly the namespaces that moved"
    );
    let truth = recount(&plane);
    for row in &moved {
        assert_eq!(
            stock(row),
            truth[&row.namespace],
            "{} reads as the recount does",
            row.namespace
        );
    }
    assert_eq!(
        truth["t0042"].4, 0,
        "the soft-deleted repository left `repos`"
    );
    assert!(truth["t0042"].3 > 0, "…and its bytes are tombstoned");

    // A cursor from another ledger (a restart) or garbage answers everything.
    let (all, _, _, changed_only) = read_all_pages(&plane, 100, "not-a-cursor");
    assert!(!changed_only);
    assert_eq!(all.len(), 60);
}

/// ★ **The property: the ledger equals a full recount after any sequence of
/// pushes, new repositories and namespaces, soft deletes, purges, undeletes,
/// LFS uploads, gc and a growth's suspend/resume** — read in full, AND as a
/// caller that keeps a copy and folds each changed-since answer onto it (which
/// is what monetize's poller does).
#[test]
fn the_ledger_equals_a_full_recount_after_random_pushes_deletes_and_growths() {
    for seed in [1u64, 7, 42, 1_234, 99_991] {
        let mut rng = Rng(seed.wrapping_mul(0x9E37_79B9_7F4A_7C15) | 1);
        let (_dir, stores, mut repos) = box_with(8);
        let plane = CoreControlPlane::new(Arc::clone(&stores));
        quiesce(&stores);
        let mut deleted: Vec<RepoRef> = Vec::new();
        let (rows, _, mut cursor, _) = read_all_pages(&plane, 3, "");
        let mut copy: Copy = rows
            .iter()
            .map(|r| (r.namespace.clone(), stock(r)))
            .collect();
        let mut next_ns = 8;
        for step in 0..60u64 {
            let what = rng.below(9);
            let op = match what {
                0 | 1 => {
                    let path = repos[rng.below(repos.len())].clone();
                    push(
                        &stores,
                        &path,
                        seed * 1_000 + step,
                        1 + rng.below(4),
                        200 + rng.below(6_000),
                    );
                    format!("push {path}")
                }
                2 => {
                    let ns = format!("t{:04}", rng.below(next_ns));
                    let path = RepoRef::new(&ns, &format!("r{step}")).expect("legal");
                    if stores.catalog().namespace(&ns).expect("read").is_some() {
                        create(&stores, &path);
                        push(&stores, &path, step, 1, 300);
                        repos.push(path.clone());
                    }
                    format!("new repo {path}")
                }
                3 => {
                    let ns = format!("t{next_ns:04}");
                    next_ns += 1;
                    stores.catalog().create_namespace(&ns).expect("namespace");
                    format!("new namespace {ns}")
                }
                4 if repos.len() > 2 => {
                    let path = repos.remove(rng.below(repos.len()));
                    let purge = rng.below(3) == 0;
                    delete(&plane, &path, purge);
                    if !purge {
                        deleted.push(path.clone());
                    }
                    format!("delete {path} purge={purge}")
                }
                5 if !deleted.is_empty() => {
                    let path = deleted.remove(rng.below(deleted.len()));
                    if rng.below(2) == 0 {
                        plane
                            .undelete_repo(&RepoPath::new(path.namespace(), path.name()))
                            .expect("undelete");
                        repos.push(path.clone());
                        format!("undelete {path}")
                    } else {
                        delete(&plane, &path, true);
                        format!("purge tombstoned {path}")
                    }
                }
                6 => {
                    let path = repos[rng.below(repos.len())].clone();
                    lfs_upload(&stores, &path, seed + step, 100 + rng.below(20_000));
                    format!("lfs {path}")
                }
                7 => {
                    let path = repos[rng.below(repos.len())].clone();
                    gc(&stores, &path);
                    format!("gc {path}")
                }
                _ => {
                    // A growth: the data set goes off-line and comes back.
                    stores.suspend_repos("a test growth");
                    stores.resume_repos();
                    "growth".to_string()
                }
            };
            quiesce(&stores);
            let truth = recount(&plane);
            // The full read.
            let (rows, _, _, _) = read_all_pages(&plane, 1 + rng.below(5), "");
            let full: Copy = rows
                .iter()
                .map(|r| (r.namespace.clone(), stock(r)))
                .collect();
            assert_eq!(
                full, truth,
                "seed {seed} step {step} ({op}): a full ReadMany disagrees with the recount"
            );
            // The poller's way: fold what changed since the cursor onto the copy.
            let (moved, _, next, changed_only) = read_all_pages(&plane, 1 + rng.below(5), &cursor);
            if changed_only {
                for row in &moved {
                    copy.insert(row.namespace.clone(), stock(row));
                }
            } else {
                copy = moved
                    .iter()
                    .map(|r| (r.namespace.clone(), stock(r)))
                    .collect();
            }
            // A namespace that is gone is not in the recount; the poller drops
            // it because the tenant list no longer names it.
            copy.retain(|ns, _| truth.contains_key(ns));
            assert_eq!(
                copy, truth,
                "seed {seed} step {step} ({op}): the changed-since copy drifted from the recount"
            );
            cursor = next;
        }
        assert_eq!(
            stores
                .usage_ledger()
                .reconcile_step(&stores, std::time::Duration::from_secs(1))
                .expect("the reconcile pass runs")
                .1,
            0,
            "seed {seed}: the reconcile pass found bytes that moved without a mark"
        );
    }
}

/// **`Usage.Read` of one namespace reads that namespace**: the same numbers as
/// the recount, for a namespace among many.
#[test]
fn usage_read_of_one_namespace_is_the_recount_s_row() {
    let (_dir, stores, repos) = box_with(20);
    let plane = CoreControlPlane::new(Arc::clone(&stores));
    push(&stores, &repos[3], 77, 2, 3_000);
    quiesce(&stores);
    let truth = recount(&plane);
    for ns in ["t0000", "t0003", "t0019"] {
        let one = plane
            .usage(Some(ns))
            .expect("Usage.Read")
            .pop()
            .expect("row");
        assert_eq!(stock(&one), truth[ns], "{ns}");
        let repos = plane.repo_usage(ns).expect("repos");
        assert_eq!(
            repos.iter().map(|r| r.stored_bytes).sum::<u64>(),
            one.pack_bytes,
            "{ns}: the rows add up to the meter"
        );
    }
    assert!(matches!(
        plane.usage(Some("nobody")),
        Err(gunnar_grpc::ControlError::NotFound(_))
    ));
}