gunnar-server 3.0.0

//! **`Usage.Read` / `Usage.ReadAll`: the meters a bill is made of, summed per
//! namespace from the same directories `Admin.StorageInfo` walks** — and
//! attributed per file, which `StorageInfo` never did (its `hot_bytes` walks
//! `<git_dir>/objects`, a directory the znippy engine does not have, so it
//! reported 0 for every appliance repository; MEASURED 2026-09-04 while
//! writing this file).
//!
//! What each meter counts is written file by file in
//! `gunnar-proto/proto/gunnar/v1/usage.proto`. This file asserts the SUMS, on a
//! real `ServerStores` with real packs pushed through the real engine:
//!
//! * three repositories in two namespaces, and each namespace's `pack_bytes`
//!   is exactly the sum of its own repositories' on-disk pack bytes — measured
//!   here by an independent walk, not by asking the server twice;
//! * a soft-deleted repository leaves `pack_bytes` and `repos` and lands in
//!   `tombstoned_bytes` — a customer who deletes a repository and sees no
//!   change in their bill calls the bill wrong, and one who sees the bytes
//!   vanish while the tombstone still holds them is being lied to;
//! * LFS bytes are their own meter and are NOT inside `pack_bytes`;
//! * `objects` is what is in RAM: the open store's indexed objects, 0 once it
//!   is closed.
//!
//! Measure, never assert: every "is counted" has its "is not counted" twin in the same
//! assertion block, so a meter that summed everything into one number would go
//! red here rather than green.

use std::sync::Arc;

use gunnar_core::{NewRepo, RepoRef};
use gunnar_grpc::seam::DeleteOptions;
use gunnar_grpc::ControlPlane as _;
use gunnar_grpc::{Caller, EventBus, ServerConfig, ServiceContext, UsageService};
use gunnar_lfs::LfsStore as _;
use gunnar_proto::usage_server::Usage as _;
use gunnar_proto::{Allowance, NamespaceUsage, ReadUsageRequest, ReadUsageResponse, RepoPath};
use gunnar_server::control_plane::CoreControlPlane;
use gunnar_server::stores::{dir_bytes, ServerStores};
use gunnar_store::conformance::pack::PackBuilder;
use prost::Message as _;

/// Push one pack of `blobs` blobs of `size` bytes each into `path`, and return
/// its git dir. The store is CLOSED on return (the handle is dropped and the
/// LRU emptied), so a later `objects` reading of 0 is the closed truth.
fn seed(stores: &ServerStores, path: &RepoRef, blobs: usize, size: usize) -> std::path::PathBuf {
    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);
    for i in 0..blobs {
        // Distinct content: identical blobs would be one object.
        let mut body = vec![b'a' + (i % 26) as u8; size];
        body[0] = i as u8;
        pb.object(gix_object::Kind::Blob, body);
    }
    let fixture = pb.build().expect("pack");
    handle.serve().put_pack(&fixture.bytes).expect("put_pack");
    let git_dir = handle.git_dir().to_path_buf();
    drop(handle);
    stores.close_repo(path);
    git_dir
}

/// The independent measurement: the whole git dir, minus its LFS objects and
/// minus the exploded cache — the definition `usage.proto` gives for
/// `pack_bytes`, computed here without any of the server's own attribution
/// code so the two can disagree.
fn expected_pack_bytes(git_dir: &std::path::Path) -> u64 {
    dir_bytes(git_dir) - lfs_bytes(git_dir) - exploded_bytes(git_dir)
}

fn lfs_bytes(git_dir: &std::path::Path) -> u64 {
    dir_bytes(&git_dir.join("lfs"))
}

fn exploded_bytes(git_dir: &std::path::Path) -> u64 {
    let store = git_dir.join("gunnar-git.znippy");
    let Ok(entries) = std::fs::read_dir(&store) else {
        return 0;
    };
    entries
        .flatten()
        .filter(|e| {
            e.file_name()
                .to_string_lossy()
                .starts_with("objects.exploded")
        })
        .map(|e| e.metadata().map(|m| m.len()).unwrap_or(0))
        .sum()
}

fn row<'a>(rows: &'a [NamespaceUsage], ns: &str) -> &'a NamespaceUsage {
    rows.iter()
        .find(|r| r.namespace == ns)
        .unwrap_or_else(|| panic!("no usage row for {ns}: {rows:?}"))
}

#[test]
fn meters_are_summed_per_namespace_and_a_soft_delete_moves_bytes_to_tombstoned() {
    let dir = tempfile::tempdir().expect("tempdir");
    let stores = Arc::new(ServerStores::open(dir.path()).expect("stores"));
    for ns in ["alpha", "beta", "empty"] {
        stores.catalog().create_namespace(ns).expect("namespace");
    }
    let a1 = RepoRef::new("alpha", "one").expect("legal");
    let a2 = RepoRef::new("alpha", "two").expect("legal");
    let b1 = RepoRef::new("beta", "solo").expect("legal");
    let a1_dir = seed(&stores, &a1, 3, 20_000);
    let a2_dir = seed(&stores, &a2, 5, 40_000);
    let b1_dir = seed(&stores, &b1, 2, 70_000);
    // An LFS object in alpha/two, written where `FsLfsStore` puts them.
    let lfs = gunnar_lfs::FsLfsStore::open_or_init(&a2_dir).expect("lfs store");
    let lfs_body = vec![0x4cu8; 123_456];
    let oid = gunnar_lfs::Oid::of(&lfs_body);
    {
        use std::io::Write as _;
        let mut w = lfs.begin_write(&oid, lfs_body.len() as u64).expect("begin");
        w.write_all(&lfs_body).expect("write");
        w.commit().expect("commit");
    }
    assert_eq!(
        lfs_bytes(&a2_dir),
        123_456,
        "the fixture's LFS object is on disk"
    );

    let plane = CoreControlPlane::new(Arc::clone(&stores));
    let all = plane.usage(None).expect("Usage.ReadAll");
    assert_eq!(
        all.len(),
        3,
        "every catalogued namespace is a row, the empty one included: {all:?}"
    );

    let alpha = row(&all, "alpha");
    let beta = row(&all, "beta");
    let empty = row(&all, "empty");
    // ── the sums, against the independent walk ─────────────────────────────
    assert_eq!(alpha.repos, 2);
    assert_eq!(beta.repos, 1);
    assert_eq!(empty.repos, 0);
    assert_eq!(
        alpha.pack_bytes,
        expected_pack_bytes(&a1_dir) + expected_pack_bytes(&a2_dir),
        "alpha's pack bytes are its two repositories' and nothing else"
    );
    assert_eq!(beta.pack_bytes, expected_pack_bytes(&b1_dir));
    assert!(
        beta.pack_bytes > 2 * 70_000,
        "a 140 kB push cannot weigh {} bytes on disk — the meter is not reading the pack",
        beta.pack_bytes
    );
    assert_ne!(
        alpha.pack_bytes, beta.pack_bytes,
        "two namespaces with different pushes must not report one number"
    );
    assert_eq!(empty.pack_bytes, 0);
    // ── LFS is its own meter, and NOT inside pack_bytes ────────────────────
    assert_eq!(alpha.lfs_bytes, 123_456);
    assert_eq!(beta.lfs_bytes, 0);
    assert!(
        alpha.pack_bytes < dir_bytes(&a1_dir) + dir_bytes(&a2_dir),
        "pack_bytes must exclude the LFS object: {} is the whole directory",
        alpha.pack_bytes
    );
    // ── nothing is tombstoned yet; nothing is open, so nothing is in RAM ───
    assert_eq!(alpha.tombstoned_bytes, 0);
    assert_eq!(beta.tombstoned_bytes, 0);
    assert_eq!(
        alpha.objects, 0,
        "closed stores hold no indexed objects in RAM"
    );
    assert_eq!(
        alpha.cold_bytes, 0,
        "one archive per repository: no cold tier to report"
    );
    assert!(alpha.measured_at_unix_ms > 0);

    // ── `Read` answers one namespace, with the same numbers ────────────────
    let one = plane.usage(Some("alpha")).expect("Usage.Read");
    assert_eq!(one.len(), 1);
    assert_eq!(one[0].pack_bytes, alpha.pack_bytes);
    assert_eq!(one[0].lfs_bytes, alpha.lfs_bytes);
    let missing = plane
        .usage(Some("nobody"))
        .expect_err("an unknown namespace is refused");
    assert!(
        matches!(missing, gunnar_grpc::ControlError::NotFound(_)),
        "not NotFound: {missing:?}"
    );

    // ── an OPEN store reports its indexed objects ──────────────────────────
    let repo = stores.catalog().get(&b1).expect("read").expect("row");
    let held = stores.repo_store(&repo).expect("open");
    held.serve().refs().expect("touch the store");
    // The index is filled by the store's background drain, so the count
    // arrives shortly after the open rather than with it. Bounded: a store
    // that never indexes two objects is a red, not a slow green.
    let deadline = std::time::Instant::now() + std::time::Duration::from_secs(20);
    let open = loop {
        let open = plane.usage(Some("beta")).expect("Usage.Read");
        if open[0].objects == 2 || std::time::Instant::now() > deadline {
            break open;
        }
        std::thread::sleep(std::time::Duration::from_millis(50));
    };
    assert_eq!(
        open[0].objects, 2,
        "beta/solo is open and holds two indexed objects: {:?}",
        open[0]
    );
    let alpha_still_closed = plane.usage(Some("alpha")).expect("Usage.Read");
    assert_eq!(
        alpha_still_closed[0].objects, 0,
        "opening beta must not charge alpha"
    );
    drop(held);
    stores.close_repo(&b1);

    // ── the soft delete: out of pack_bytes and repos, into tombstoned_bytes ─
    let bytes_of_a1 = dir_bytes(&a1_dir);
    plane
        .delete_repo(
            &RepoPath::new("alpha", "one"),
            DeleteOptions {
                purge: false,
                retention_days: Some(30),
                principal_id: "tester".into(),
            },
        )
        .expect("soft delete");
    let after = plane.usage(Some("alpha")).expect("Usage.Read");
    let alpha_after = &after[0];
    assert_eq!(
        alpha_after.repos, 1,
        "the tombstoned repository is no longer live"
    );
    assert_eq!(
        alpha_after.pack_bytes,
        expected_pack_bytes(&a2_dir),
        "the deleted repository's pack bytes left pack_bytes"
    );
    assert_eq!(
        alpha_after.tombstoned_bytes, bytes_of_a1,
        "…and every byte its directory still holds is in tombstoned_bytes"
    );
    assert!(
        alpha_after.tombstoned_bytes > 0,
        "the tombstone holds real bytes"
    );
    let beta_after = plane.usage(Some("beta")).expect("Usage.Read");
    assert_eq!(
        beta_after[0].tombstoned_bytes, 0,
        "alpha's tombstone is not charged to beta"
    );
}

/// **`lfs_bytes` is `gunnar-lfs`'s actual store, per namespace — including the
/// uploads still in flight.**
///
/// The test above asserts one committed object. This one asserts the two things
/// that make the meter a *store* reading rather than a directory reading, and
/// both of them are contract, written in `usage.proto`:
///
/// * the number is the store's own accounting — the sum of
///   [`gunnar_lfs::LfsStore::size_of`] over what the store holds, asked of the
///   store rather than of the filesystem, so a meter that had drifted from the
///   layout `FsLfsStore` writes would go red here rather than agree with itself;
/// * an upload **in flight** (`lfs/incomplete/`) is charged. It occupies the
///   tenant's disk for as long as it is open, and a bill that only counted
///   committed objects would report a repository holding gigabytes as holding
///   none. `RepoBytes::lfs`'s own doc says "committed and in flight"; nothing
///   asserted the second half.
///
/// And the twin that keeps it honest: when the abandoned upload is dropped,
/// `FsPendingWrite::drop` sweeps it and the meter comes back down. A meter that
/// counted a directory it never re-read would stay high.
///
/// `pack_bytes` is asserted unchanged across all four readings, because "LFS is
/// its own meter" is a claim about what `pack_bytes` does NOT contain.
///
/// Red first, MEASURED 2026-09-05: with `control_plane::repo_bytes` pointed at
/// `lfs/objects` instead of the whole `lfs/` directory — the shape the meter
/// would have if it counted committed objects only — this fails at
/// *"an upload in flight is on the tenant's disk and on their bill"*:
///
/// ```text
///   left: (3690015, 123456)
///  right: (3690015, 223456)
/// ```
///
/// `meters_are_summed_per_namespace_and_a_soft_delete_moves_bytes_to_tombstoned`
/// stayed green through that same change, which is exactly why this test
/// exists: one committed object cannot tell the two meters apart.
#[test]
fn lfs_bytes_is_the_store_itself_and_charges_uploads_in_flight() {
    use std::io::Write as _;

    let dir = tempfile::tempdir().expect("tempdir");
    let stores = Arc::new(ServerStores::open(dir.path()).expect("stores"));
    stores
        .catalog()
        .create_namespace("team")
        .expect("namespace");
    let path = RepoRef::new("team", "one").expect("legal");
    let git_dir = seed(&stores, &path, 3, 20_000);
    let plane = CoreControlPlane::new(Arc::clone(&stores));

    let packed = plane.usage(Some("team")).expect("usage")[0].pack_bytes;
    assert!(packed > 0, "the fixture pushed a pack");
    let read = |plane: &CoreControlPlane| {
        let row = plane
            .usage(Some("team"))
            .expect("usage")
            .pop()
            .expect("row");
        (row.pack_bytes, row.lfs_bytes)
    };
    assert_eq!(read(&plane), (packed, 0), "no LFS yet");

    // ── one committed object ──────────────────────────────────────────────
    let lfs = gunnar_lfs::FsLfsStore::open_or_init(&git_dir).expect("lfs store");
    let committed = vec![0x4cu8; 123_456];
    let oid = gunnar_lfs::Oid::of(&committed);
    {
        let mut w = lfs
            .begin_write(&oid, committed.len() as u64)
            .expect("begin");
        w.write_all(&committed).expect("write");
        w.commit().expect("commit");
    }
    // The store's own accounting, asked of the store. Not `dir_bytes`: that
    // would be the meter's own arithmetic written twice and agreeing with
    // itself.
    assert_eq!(
        lfs.size_of(&oid)
            .expect("stat")
            .expect("the object is there"),
        123_456
    );
    assert_eq!(
        read(&plane),
        (packed, 123_456),
        "the meter is the LFS store's bytes, and they are not in pack_bytes"
    );

    // ── one upload in flight ──────────────────────────────────────────────
    let in_flight = vec![0x5du8; 100_000];
    let pending_oid = gunnar_lfs::Oid::of(&in_flight);
    let mut pending = lfs
        .begin_write(&pending_oid, in_flight.len() as u64)
        .expect("begin");
    pending.write_all(&in_flight).expect("write");
    pending.flush().expect("flush");
    assert_eq!(
        lfs.size_of(&pending_oid).expect("stat"),
        None,
        "an uncommitted upload resolves to no oid — that is the store's whole safety property"
    );
    assert_eq!(
        read(&plane),
        (packed, 223_456),
        "an upload in flight is on the tenant's disk and on their bill"
    );

    // ── and it goes when the upload is abandoned ──────────────────────────
    drop(pending);
    assert_eq!(
        read(&plane),
        (packed, 123_456),
        "the abandoned upload's bytes are still being charged after Drop swept the file"
    );
}

// ───────────────────────────────────────────────────────────────────────────
// The allowance: what a namespace is ALLOWED, beside what it holds
// ───────────────────────────────────────────────────────────────────────────
//
// `Usage.Read` carries an `Allowance` since 2026-09-05 — per meter, the cap
// this box holds for the namespace, ABSENT when there is none. This file is
// compiled without `metered`, so it is the community appliance's half: every
// cap absent for everyone, spelled as absence and never as 0, on the same
// `Read` grant the meters already take. The metered half — a fact for alice
// puts her cap on the wire and leaves bob's absent — is in
// `tests/entitlement.rs`, beside the facts.

/// The real `Usage` service over the real plane and the real auth table.
fn usage_service(stores: &Arc<ServerStores>) -> UsageService {
    let plane: Arc<dyn gunnar_grpc::ControlPlane> =
        Arc::new(CoreControlPlane::new(Arc::clone(stores)));
    UsageService::new(ServiceContext::new(
        plane,
        Arc::new(EventBus::new(16)),
        ServerConfig::default(),
    ))
}

/// `Usage.Read` as `principal`, the way the auth gate would hand it in.
fn read_as(
    service: &UsageService,
    principal: &str,
    namespace: &str,
) -> Result<ReadUsageResponse, tonic::Status> {
    let mut request = tonic::Request::new(ReadUsageRequest {
        namespace: namespace.to_owned(),
    });
    request.extensions_mut().insert(Caller {
        principal_id: principal.to_owned(),
        fingerprint: "SHA256:test".to_owned(),
        machine: false,
        authenticated: true,
    });
    tokio::runtime::Builder::new_current_thread()
        .enable_all()
        .build()
        .expect("runtime")
        .block_on(service.read(request))
        .map(tonic::Response::into_inner)
}

/// A principal whose id IS the namespace: its owner by rule 2 of
/// `AuthTable::may`, so `Read` on the namespace is theirs without a grant row.
fn owner(stores: &ServerStores, ns: &str) {
    stores
        .auth()
        .add_principal(gunnar_core::Principal {
            id: ns.to_owned(),
            name: ns.to_owned(),
            kind: gunnar_core::PrincipalKind::User,
            created: 0,
        })
        .expect("principal");
}

/// **The community appliance caps nobody, and says so as ABSENCE.**
///
/// Every one of the seven caps is `None` on a namespace that really holds
/// bytes — a `Some(0)` would read as "may hold nothing", which is the
/// opposite of the truth. The meters beside it are non-zero, so a response
/// built from `Default::default()` could not pass. Seen red with the handler
/// filling `pack_bytes: Some(0)`: *"pack_bytes cap must be ABSENT on the
/// community appliance, not Some(0)"*; and with `allowance: None`: *"the
/// allowance message itself is always present"*.
#[test]
fn the_community_appliance_reports_every_cap_absent_never_zero() {
    let dir = tempfile::tempdir().expect("tempdir");
    let stores = Arc::new(ServerStores::open(dir.path()).expect("stores"));
    stores
        .catalog()
        .create_namespace("alice")
        .expect("namespace");
    owner(&stores, "alice");
    seed(
        &stores,
        &RepoRef::new("alice", "held").expect("legal"),
        4,
        30_000,
    );

    let service = usage_service(&stores);
    let answer = read_as(&service, "alice", "alice").expect("the owner reads her own meters");
    let usage = answer.usage.expect("the meters");
    assert!(
        usage.pack_bytes > 0,
        "the fixture must hold bytes, or absence proves nothing"
    );

    let allowance = answer.allowance.expect(
        "the allowance message itself is always present; only the caps inside are optional",
    );
    for (meter, cap) in [
        ("pack_bytes", allowance.pack_bytes),
        ("cache_bytes", allowance.cache_bytes),
        ("cold_bytes", allowance.cold_bytes),
        ("lfs_bytes", allowance.lfs_bytes),
        ("tombstoned_bytes", allowance.tombstoned_bytes),
        ("repos", allowance.repos),
        ("objects", allowance.objects),
    ] {
        assert!(
            cap.is_none(),
            "{meter} cap must be ABSENT on the community appliance, not {cap:?}"
        );
        assert_ne!(cap, Some(0), "{meter}: absent is never spelled 0");
    }
    // And nothing else in the message either — the exhaustive half, so a field
    // added to `Allowance` and left populated here fails this test rather than
    // being reported to a community tenant as a limit.
    //
    // The expectation is NOT `Allowance::default()` any more, and the
    // difference is the point: since `Allowance.enforcement`, an uncapped
    // answer still carries the seven words for what a cap WOULD do
    // (`service::meters`), because an empty map means "this build did not say"
    // and this build did. Absent caps and a full label table are both true at
    // once here, and the loop above is what keeps "absent, never zero" honest —
    // the label table cannot satisfy it.
    assert_eq!(
        allowance,
        Allowance {
            enforcement: gunnar_grpc::service::meters::enforcement(),
            ..Allowance::default()
        },
        "the community appliance caps nobody and still says what its three \
         doors are"
    );
}

/// ★ **A tenant sees where their room went, repository by repository, on
/// their own `Read`** (2026-10-04, the thin layer: the console showed "My room"
/// by reading the WHOLE BOX's `Admin.StorageInfo` with its own key and
/// filtering it). Each row is measured by the same walk that sums the meters,
/// so the rows add up to `pack_bytes` exactly; another account's repository is
/// never among them.
#[test]
fn a_tenant_sees_where_their_room_went_repository_by_repository() {
    let dir = tempfile::tempdir().expect("tempdir");
    let stores = Arc::new(ServerStores::open(dir.path()).expect("stores"));
    for ns in ["alice", "bob"] {
        stores.catalog().create_namespace(ns).expect("namespace");
        owner(&stores, ns);
    }
    let big = seed(
        &stores,
        &RepoRef::new("alice", "big").expect("legal"),
        6,
        40_000,
    );
    let small = seed(
        &stores,
        &RepoRef::new("alice", "small").expect("legal"),
        2,
        10_000,
    );
    seed(
        &stores,
        &RepoRef::new("bob", "theirs").expect("legal"),
        3,
        20_000,
    );

    let answer = read_as(&usage_service(&stores), "alice", "alice").expect("alice reads alice");
    let usage = answer.usage.expect("meters");
    let mut rows: Vec<(String, u64)> = answer
        .repos
        .iter()
        .map(|r| {
            let path = r.repo.clone().expect("a path");
            (format!("{}/{}", path.namespace, path.repo), r.stored_bytes)
        })
        .collect();
    rows.sort();
    assert_eq!(
        rows,
        vec![
            ("alice/big".to_owned(), expected_pack_bytes(&big)),
            ("alice/small".to_owned(), expected_pack_bytes(&small)),
        ],
        "alice's two repositories, each measured as pack_bytes is — and nothing of bob's"
    );
    assert_eq!(
        rows.iter().map(|(_, b)| *b).sum::<u64>(),
        usage.pack_bytes,
        "the rows add up to the meter"
    );
}

/// The community appliance caps nobody, so nothing is "remaining": the
/// message is present and every meter in it is absent — never a number.
#[test]
fn the_community_appliance_has_nothing_remaining_because_nothing_is_capped() {
    let dir = tempfile::tempdir().expect("tempdir");
    let stores = Arc::new(ServerStores::open(dir.path()).expect("stores"));
    stores
        .catalog()
        .create_namespace("alice")
        .expect("namespace");
    owner(&stores, "alice");
    seed(
        &stores,
        &RepoRef::new("alice", "held").expect("legal"),
        2,
        20_000,
    );
    let answer = read_as(&usage_service(&stores), "alice", "alice").expect("read");
    assert_eq!(
        answer.remaining,
        Some(gunnar_proto::Remaining::default()),
        "present, and every meter absent"
    );
}

/// **`Read` on the namespace is enough, and a stranger is refused** — the
/// allowance rides on the meters' own grant, not on `Entitlement.Get`'s
/// server-wide one. bob is a registered principal with no grant on alice:
/// refused, and refused with no allowance in the answer (there is no answer).
#[test]
fn the_allowance_takes_read_on_the_namespace_and_a_stranger_gets_nothing() {
    let dir = tempfile::tempdir().expect("tempdir");
    let stores = Arc::new(ServerStores::open(dir.path()).expect("stores"));
    for ns in ["alice", "bob"] {
        stores.catalog().create_namespace(ns).expect("namespace");
        owner(&stores, ns);
    }
    let service = usage_service(&stores);

    // The control: alice, on her own.
    let hers = read_as(&service, "alice", "alice").expect("the owner reads");
    assert!(hers.allowance.is_some());
    // bob, on his own — also fine, so the refusal below is about the grant
    // and not about bob.
    assert!(read_as(&service, "bob", "bob")
        .expect("bob reads bob")
        .allowance
        .is_some());
    // bob, on alice: refused by the same rule the meters are refused under.
    let refused = read_as(&service, "bob", "alice").expect_err("a stranger is refused");
    assert_eq!(refused.code(), tonic::Code::PermissionDenied, "{refused:?}");
    // And the caller the gate hands in for NO credential — the reserved
    // anonymous principal — is refused too, not shown as "uncapped". (That the
    // gate refuses a missing credential before any handler runs is
    // `gunnar-grpc/tests/auth.rs`'s sweep; this is the rule below it.)
    let err = read_as(&service, gunnar_grpc::ANONYMOUS_PRINCIPAL_ID, "alice")
        .expect_err("the anonymous principal reads nobody's allowance");
    assert_eq!(err.code(), tonic::Code::PermissionDenied, "{err:?}");
}

/// `ReadUsageResponse` **as an older client compiled it**: the meters and
/// nothing after them. Decoding today's bytes into this is what a gunnar-ui
/// or a monetize built before the allowance does.
#[derive(Clone, PartialEq, prost::Message)]
struct OlderReadUsageResponse {
    #[prost(message, optional, tag = "1")]
    usage: Option<NamespaceUsage>,
}

/// **An older client gets the same meters, byte for byte.** Today's response,
/// encoded, decodes in the pre-allowance message to the identical
/// `NamespaceUsage`; and today's message with the allowance stripped encodes
/// to exactly what the older client re-encodes — the meters' half of the wire
/// did not move. Seen red by moving the allowance to tag 1 in the proto: the
/// older decoder then reads the allowance as a `NamespaceUsage` and the
/// meters compare unequal.
#[test]
fn an_older_client_decodes_the_same_meters_from_todays_bytes() {
    let dir = tempfile::tempdir().expect("tempdir");
    let stores = Arc::new(ServerStores::open(dir.path()).expect("stores"));
    stores
        .catalog()
        .create_namespace("alice")
        .expect("namespace");
    owner(&stores, "alice");
    seed(
        &stores,
        &RepoRef::new("alice", "held").expect("legal"),
        2,
        50_000,
    );

    let today = read_as(&usage_service(&stores), "alice", "alice").expect("read");
    assert!(
        today.allowance.is_some(),
        "this test needs the new field PRESENT, or it proves nothing"
    );
    let bytes = today.encode_to_vec();
    let older = OlderReadUsageResponse::decode(bytes.as_slice()).expect("an older client decodes");
    assert_eq!(older.usage, today.usage);
    assert!(
        older.usage.as_ref().is_some_and(|u| u.pack_bytes > 0),
        "{older:?}"
    );

    // Everything after the meters stripped — the allowance, and since
    // 2026-10-04 `remaining` and the per-repository rows too.
    let stripped = ReadUsageResponse {
        usage: today.usage.clone(),
        ..ReadUsageResponse::default()
    };
    assert_eq!(stripped.encode_to_vec(), older.encode_to_vec());
    assert!(
        bytes.len() > older.encoded_len(),
        "today's bytes carry more than the older message — else the allowance was absent"
    );
}

/// **The same RPC, not a new one.** The allowance rides on `Usage.Read`; the
/// deny-by-default census in `gunnar-grpc/src/auth.rs` gains nothing for it.
///
/// `Usage.Activity` joined the list on 2026-09-11 and is NOT a counter-example:
/// it is an RPC `usage.proto` has always declared and the census had simply
/// never listed (GUN-023). `Usage.AdmitPlan` joined on 2026-09-28 as a verb of
/// its own (the admission plan, not a cap), and is not one either. The property
/// here is about the ALLOWANCE — that it
/// is a field on `ReadUsageResponse` and not a verb of its own — so the
/// assertion names the family the protos declare and then says, separately,
/// that nothing in it is named for the allowance.
#[test]
fn the_allowance_rides_on_usage_read_and_adds_no_rpc_to_the_census() {
    let usage: Vec<&str> = gunnar_grpc::auth::METHODS
        .iter()
        .copied()
        .filter(|m| m.starts_with("Usage."))
        .collect();
    assert_eq!(
        usage,
        [
            "Usage.Activity",
            "Usage.AdmitPlan",
            "Usage.Read",
            "Usage.ReadAll"
        ],
        "the Usage family changed; the allowance was meant to ride on Usage.Read"
    );
    assert!(
        !gunnar_grpc::auth::METHODS
            .iter()
            .any(|m| m.contains("Allowance")),
        "no RPC is named for the allowance"
    );
}