crates/gunnar-server/tests/flow_meters.rs
//! **The FLOW meters, from the ledger to the wire.**
//!
//! Every meter `usage_meters.rs` asserts is a STOCK meter: it answers *how much
//! is on the disk*, and it reads exactly the same for a tenant who pushed this
//! morning and a tenant who last touched this box in March. This file is the
//! other kind — what MOVED — and it asserts the seam that carries it:
//!
//! ```text
//! the transfer paths ──▶ gunnar_core::FlowLedger ──▶ CoreControlPlane
//! ├─▶ Usage.Read (today)
//! └─▶ Usage.Activity (the series)
//! ```
//!
//! The ledger's own arithmetic — the day boundary, the flush that copies rather
//! than adds, the retention window, the exact principal set — is proved in
//! `gunnar_core::flow`'s unit tests against a real redb. What is proved HERE is
//! everything that could still be wrong once those numbers are right: that one
//! namespace's traffic never lands on another's row, that `Usage.Read` carries
//! the live figure and not a stale flushed one, that a refusal is counted
//! WITHOUT moving the bytes, that the series comes back oldest first with the
//! reason breakdown intact, and that `Usage.Activity` takes the namespace's own
//! `Read` grant rather than a server-wide one.
//!
//! **Measure, never assert**: every "is counted" below has its "is not counted"
//! twin in the same block, so a ledger that summed every namespace into one row
//! would go red here rather than green.
//!
//! # Why the transfers are driven through `FlowLedger::record` and not a socket
//!
//! `transport::serve_blocking`'s counting point is a `Drop` guard whose whole
//! body is one `record` call fed by the two `Metered` counters that already
//! feed `ServerMetrics`. Driving a real `git push` here would re-prove
//! `Metered`, which `usage_meters.rs` and the transport's own tests already
//! own, and would not reach a single line this file is about. What this file
//! drives is the exact call that guard makes.
use std::sync::Arc;
use gunnar_core::{FlowLedger, Refusal, Transfer};
use gunnar_grpc::ControlPlane as _;
use gunnar_grpc::{Caller, EventBus, ServerConfig, ServiceContext, UsageService};
use gunnar_proto::usage_server::Usage as _;
use gunnar_proto::{ActivityRequest, ActivityResponse, NamespaceUsage, ReadUsageRequest};
use gunnar_server::control_plane::CoreControlPlane;
use gunnar_server::stores::ServerStores;
const DAY: i64 = gunnar_core::DAY_MS;
/// **Midday of the CURRENT UTC day**, and it has to be the current one.
///
/// RED with a hard-coded instant: `ControlPlane::activity` resolves its window
/// against `now`, and clamps `until` to today — because a series that answered
/// for tomorrow would be answering about a day that has not happened. So a
/// fixture pinned to a fixed date passes until the wall clock moves past it and
/// then quietly loses its last day, which is how
/// `the_day_boundary_splits_the_counts` came to assert two rows and get one.
///
/// Every timestamp in this file is derived from this and points BACKWARDS.
fn noon() -> i64 {
gunnar_core::midnight_of(gunnar_core::day_of(gunnar_core::now_ms())) + DAY / 2
}
fn push(bytes: u64, who: &str) -> Transfer {
Transfer {
bytes_in: bytes,
pushed: true,
principal: who.into(),
..Transfer::default()
}
}
fn fetch(bytes: u64, who: &str) -> Transfer {
Transfer {
bytes_out: bytes,
fetched: true,
principal: who.into(),
..Transfer::default()
}
}
/// A box with two namespaces and nothing in them. The repositories are beside
/// the point: flow is about what moved, not about what is held, and a namespace
/// with no repository can still be pushed to (the first-repository door) and
/// still be a tenant an operator wants to see.
fn box_with(namespaces: &[&str]) -> (tempfile::TempDir, Arc<ServerStores>) {
let dir = tempfile::tempdir().expect("tempdir");
let stores = Arc::new(ServerStores::open(dir.path()).expect("stores"));
for ns in namespaces {
stores.catalog().create_namespace(ns).expect("namespace");
// A principal whose id IS the namespace owns it (rule 2 of
// `AuthTable::may`), which is how the grant half below reads without a
// grant row.
stores
.auth()
.add_principal(gunnar_core::Principal {
id: (*ns).to_owned(),
name: (*ns).to_owned(),
kind: gunnar_core::PrincipalKind::User,
created: 0,
})
.expect("principal");
}
(dir, stores)
}
fn flow(stores: &ServerStores) -> &Arc<FlowLedger> {
stores.flow()
}
fn meters(stores: &Arc<ServerStores>, ns: &str) -> NamespaceUsage {
CoreControlPlane::new(Arc::clone(stores))
.usage(Some(ns))
.expect("Usage.Read")
.pop()
.expect("one row")
}
fn 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(),
))
}
fn activity_as(
service: &UsageService,
principal: &str,
namespace: &str,
since_unix_ms: i64,
until_unix_ms: i64,
) -> Result<ActivityResponse, tonic::Status> {
let mut request = tonic::Request::new(ActivityRequest {
namespace: namespace.to_owned(),
since_unix_ms,
until_unix_ms,
});
request.extensions_mut().insert(Caller {
principal_id: principal.to_owned(),
fingerprint: "SHA256:test".to_owned(),
machine: false,
authenticated: true,
});
runtime()
.block_on(service.activity(request))
.map(tonic::Response::into_inner)
}
fn read_as(
service: &UsageService,
principal: &str,
namespace: &str,
) -> Result<NamespaceUsage, 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,
});
runtime()
.block_on(service.read(request))
.map(|answer| answer.into_inner().usage.expect("the meters"))
}
fn runtime() -> tokio::runtime::Runtime {
tokio::runtime::Builder::new_current_thread()
.enable_all()
.build()
.expect("runtime")
}
// ───────────────────────────────────────────────────────────────────────────
// Usage.Read: today, beside what is held
// ───────────────────────────────────────────────────────────────────────────
/// ★ **A push of N bytes shows `bytes_in_today = N` and `pushes_today = 1` for
/// the namespace it was pushed to — and ZERO for the one beside it.**
///
/// The first thing that has to be true of a per-tenant flow meter and the one
/// thing `ServerMetrics` — a single set of counters for the whole box — could
/// never be. RED before the flow fields existed: every one of these seven
/// numbers was absent from `NamespaceUsage` and the answer to *is this customer
/// using the product* was the same for every tenant on the box.
#[test]
fn a_push_lands_on_its_own_namespace_and_the_neighbour_stays_at_zero() {
let (_dir, stores) = box_with(&["alpha", "beta"]);
flow(&stores).record("alpha", noon(), &push(4_096, "SHA256:aaa"));
let alpha = meters(&stores, "alpha");
assert_eq!(alpha.bytes_in_today, 4_096);
assert_eq!(alpha.pushes_today, 1);
assert_eq!(alpha.last_push_unix_ms, noon());
// The twin: a push serves nothing, and the neighbour moved nothing.
assert_eq!(alpha.bytes_out_today, 0, "a push serves no pack");
assert_eq!(alpha.fetches_today, 0);
assert_eq!(alpha.last_fetch_unix_ms, 0, "never fetched is 0, not now");
let beta = meters(&stores, "beta");
assert_eq!(
(beta.bytes_in_today, beta.pushes_today, beta.bytes_out_today),
(0, 0, 0),
"alpha's push reached beta's row: the ledger is not per namespace"
);
assert_eq!(beta.last_push_unix_ms, 0);
}
/// ★ **A fetch shows `bytes_out_today`; the bare advertisement does not show a
/// fetch.**
///
/// `git ls-remote` and a monitoring probe take the ref advertisement and ask
/// for no objects. Counting those as fetches would make a health check look
/// like a customer, which is the one reading that would make the SILENT list
/// useless.
#[test]
fn a_fetch_is_bytes_out_and_an_advertisement_is_bytes_without_a_fetch() {
let (_dir, stores) = box_with(&["alpha"]);
flow(&stores).record("alpha", noon(), &fetch(9_000, "SHA256:bbb"));
flow(&stores).record(
"alpha",
noon(),
&Transfer {
bytes_out: 120,
..Transfer::default()
},
);
let alpha = meters(&stores, "alpha");
assert_eq!(alpha.bytes_out_today, 9_120, "both wrote bytes to the wire");
assert_eq!(alpha.fetches_today, 1, "only one of them asked for objects");
assert_eq!(alpha.bytes_in_today, 0, "a fetch accepts nothing");
assert_eq!(alpha.last_fetch_unix_ms, noon());
}
/// ★ **A refused push shows `refused_today = 1` with its reason, and the bytes
/// do not move.**
///
/// The distinction the whole `refused` counter exists for: a tenant who is
/// being turned away has traffic that produced no work, and a meter that could
/// not tell the two apart would report a customer hammering a closed door as a
/// customer using the product.
#[test]
fn a_refusal_is_counted_by_reason_and_moves_no_bytes() {
let (_dir, stores) = box_with(&["alpha", "beta"]);
flow(&stores).record("alpha", noon(), &push(4_096, "SHA256:aaa"));
flow(&stores).refuse("alpha", noon(), Refusal::Cap);
flow(&stores).refuse("alpha", noon(), Refusal::Cap);
flow(&stores).refuse("alpha", noon(), Refusal::Admission);
let alpha = meters(&stores, "alpha");
assert_eq!(alpha.refused_today, 3);
assert_eq!(alpha.bytes_in_today, 4_096, "a refused push moved nothing");
assert_eq!(alpha.pushes_today, 1, "and it is not a push either");
assert_eq!(meters(&stores, "beta").refused_today, 0);
// The breakdown is on the series, by the three words an operator acts on
// differently. A reason with no refusals is ABSENT, never 0.
let day = day_series(&stores, "alpha");
let today = day.last().expect("today has a row");
assert_eq!(today.refused, 3);
assert_eq!(today.refused_by_reason.get("cap"), Some(&2));
assert_eq!(today.refused_by_reason.get("admission"), Some(&1));
assert_eq!(
today.refused_by_reason.get("policy"),
None,
"a reason with no refusals must be absent, not a zero row on a console"
);
}
/// `last_push_unix_ms` looks back over the WHOLE kept window, not over today.
///
/// The number an operator reads to decide a tenant has gone quiet. A tenant
/// silent for six weeks has a today of zeroes and a last-seen six weeks ago,
/// and reading the second out of the first is exactly the mistake the pair
/// exists to prevent.
#[test]
fn last_seen_reaches_back_past_today() {
let (_dir, stores) = box_with(&["alpha"]);
let six_weeks_ago = noon() - 42 * DAY;
flow(&stores).record("alpha", six_weeks_ago, &push(1_000, "SHA256:aaa"));
flow(&stores).flush(six_weeks_ago).expect("flush");
let alpha = CoreControlPlane::new(Arc::clone(&stores))
.usage(Some("alpha"))
.expect("usage")
.pop()
.expect("row");
assert_eq!(alpha.bytes_in_today, 0, "nothing moved today");
assert_eq!(alpha.pushes_today, 0);
assert_eq!(
alpha.last_push_unix_ms, six_weeks_ago,
"a today of zeroes must not erase when the tenant was last seen"
);
}
// ───────────────────────────────────────────────────────────────────────────
// Usage.Activity: the series
// ───────────────────────────────────────────────────────────────────────────
fn day_series(stores: &Arc<ServerStores>, ns: &str) -> Vec<gunnar_proto::DayActivity> {
CoreControlPlane::new(Arc::clone(stores))
.activity(ns, 0, 0)
.expect("Usage.Activity")
.0
}
/// ★ **The series comes back OLDEST FIRST, and a day with no traffic has no
/// row.**
///
/// Oldest first because every consumer of it draws a left-to-right axis, and a
/// series whose order the caller has to know to re-sort is a series two
/// consumers will draw two ways. No row for a silent day because the caller
/// already knows which days it asked for: a day that was never written and a
/// day on which nothing happened are the same fact, and zero-filling would put
/// a bar of height zero beside a bar that means "not counted".
#[test]
fn the_series_is_oldest_first_and_a_silent_day_has_no_row() {
let (_dir, stores) = box_with(&["alpha"]);
for (days_ago, bytes) in [(4i64, 100u64), (3, 200), (0, 400)] {
let at = noon() - days_ago * DAY;
flow(&stores).record("alpha", at, &push(bytes, "SHA256:aaa"));
flow(&stores).flush(at).expect("flush");
}
let series = day_series(&stores, "alpha");
assert_eq!(series.len(), 3, "three days moved something: {series:?}");
let days: Vec<i64> = series.iter().map(|d| d.day_unix_ms).collect();
let mut sorted = days.clone();
sorted.sort_unstable();
assert_eq!(days, sorted, "the series must be oldest first");
assert_eq!(
series.iter().map(|d| d.bytes_in).collect::<Vec<_>>(),
vec![100, 200, 400]
);
// The two silent days in the middle are absent, not zero rows.
let midnight =
|days_ago: i64| gunnar_core::midnight_of(gunnar_core::day_of(noon() - days_ago * DAY));
assert!(
!days.contains(&midnight(2)),
"a day with no traffic is not a row"
);
assert!(!days.contains(&midnight(1)));
// Distinct principals is exact, and it is per DAY: one key across three
// days is one key on each of them, not three.
assert!(series.iter().all(|d| d.distinct_principals == 1));
assert!(series.iter().all(|d| !d.distinct_principals_capped));
}
/// A window narrower than the kept series returns only what it names, and the
/// answer says how many days this box keeps — so an empty answer that means
/// *nothing happened* can be told from one that means *you asked about a time
/// this box no longer remembers*.
#[test]
fn the_window_is_honoured_and_the_retention_is_reported() {
let (_dir, stores) = box_with(&["alpha"]);
for days_ago in [4i64, 3, 2, 1, 0] {
let at = noon() - days_ago * DAY;
flow(&stores).record("alpha", at, &push(10, "SHA256:aaa"));
flow(&stores).flush(at).expect("flush");
}
let plane = CoreControlPlane::new(Arc::clone(&stores));
let (all, retained) = plane.activity("alpha", 0, 0).expect("whole window");
assert_eq!(all.len(), 5);
assert_eq!(retained, gunnar_core::RETENTION_DAYS);
let (narrow, _) = plane
.activity("alpha", noon() - 2 * DAY, noon() - DAY)
.expect("narrow window");
assert_eq!(narrow.len(), 2, "two days named, two days back: {narrow:?}");
// A window that runs backwards is an empty question, not an error.
let (backwards, _) = plane
.activity("alpha", noon(), noon() - 4 * DAY)
.expect("a backwards window is answered, not refused");
assert!(backwards.is_empty());
}
/// A namespace the catalog does not hold is `NotFound`, never an empty series.
///
/// The one conclusion this whole ledger exists to make reliable is *this tenant
/// has gone quiet*. If a typo answered "no activity" it would be
/// indistinguishable from the real thing.
#[test]
fn an_unknown_namespace_is_not_found_rather_than_silent() {
let (_dir, stores) = box_with(&["alpha"]);
let err = CoreControlPlane::new(Arc::clone(&stores))
.activity("alhpa", 0, 0)
.expect_err("a typo must not read as a silent tenant");
assert!(
format!("{err:?}").contains("alhpa"),
"the refusal must name what was not found: {err:?}"
);
}
/// ★ **`Usage.Activity` takes `Read` on the NAMESPACE — the same grant
/// `Usage.Read` takes, and not the server-wide one `ReadAll` takes.**
///
/// What a tenant pushed and fetched is that tenant's own business. Putting it
/// behind an operator credential would mean a console had to read everyone's
/// traffic to show one person theirs, which is the shape of gate that ends with
/// one tenant's numbers on another tenant's page.
#[test]
fn activity_is_the_namespaces_own_read_grant() {
let (_dir, stores) = box_with(&["alpha", "beta"]);
flow(&stores).record("alpha", noon(), &push(4_096, "SHA256:aaa"));
flow(&stores).flush(noon()).expect("flush");
let service = service(&stores);
let mine = activity_as(&service, "alpha", "alpha", 0, 0).expect("the owner reads her own");
assert_eq!(mine.days.len(), 1);
assert_eq!(mine.days[0].bytes_in, 4_096);
assert_eq!(mine.retained_days, gunnar_core::RETENTION_DAYS);
let theirs = activity_as(&service, "beta", "alpha", 0, 0)
.expect_err("beta must not read alpha's traffic");
assert_eq!(theirs.code(), tonic::Code::PermissionDenied, "{theirs:?}");
// And the same principal is refused the same way on `Read`, so the two
// calls cannot drift apart on the grant they take.
assert_eq!(
read_as(&service, "beta", "alpha")
.expect_err("beta must not read alpha's meters either")
.code(),
tonic::Code::PermissionDenied
);
// An empty namespace is an argument error and not a whole-server read.
assert_eq!(
activity_as(&service, "alpha", "", 0, 0)
.expect_err("no namespace, no answer")
.code(),
tonic::Code::InvalidArgument
);
}
/// ★ **A restart keeps yesterday and drops nothing of today.**
///
/// The flush copies rather than adds and a new slot is seeded from the store,
/// so a second `ServerStores` over the same data root goes on counting today
/// from where the first left it. Without the seed every restart would look like
/// a quiet morning; without the copy every restart would double the day.
#[test]
fn a_restart_keeps_yesterday_and_resumes_today() {
let dir = tempfile::tempdir().expect("tempdir");
let yesterday = noon() - DAY;
{
let stores = Arc::new(ServerStores::open(dir.path()).expect("stores"));
stores
.catalog()
.create_namespace("alpha")
.expect("namespace");
flow(&stores).record("alpha", yesterday, &push(1_000, "SHA256:aaa"));
flow(&stores).flush(yesterday).expect("flush");
flow(&stores).record("alpha", noon(), &push(500, "SHA256:aaa"));
flow(&stores).flush(noon()).expect("flush");
}
// The server restarts on the same data root.
let stores = Arc::new(ServerStores::open(dir.path()).expect("reopen"));
assert_eq!(
meters(&stores, "alpha").bytes_in_today,
500,
"today did not survive the restart"
);
flow(&stores).record("alpha", noon() + 60_000, &push(1, "SHA256:aaa"));
let after = meters(&stores, "alpha");
assert_eq!(
after.bytes_in_today, 501,
"counting resumed from zero instead of from the stored figure"
);
assert_eq!(after.pushes_today, 2);
let series = day_series(&stores, "alpha");
assert_eq!(series.len(), 2, "yesterday and today: {series:?}");
assert_eq!(series[0].bytes_in, 1_000, "yesterday is untouched");
}
/// ★ **A day rollover splits the counts**, and the completed day keeps its own.
///
/// The boundary crossed is YESTERDAY into TODAY, never today into tomorrow:
/// `Usage.Activity` clamps its window to today, so a fixture that pushed on
/// both sides of the NEXT midnight would have its second day clipped off and
/// this would assert one row where it means two. See [`noon`].
#[test]
fn the_day_boundary_splits_the_counts() {
let (_dir, stores) = box_with(&["alpha"]);
let last_second = gunnar_core::midnight_of(gunnar_core::day_of(noon())) - 1;
let first_second = last_second + 2;
flow(&stores).record("alpha", last_second, &push(1_000, "SHA256:aaa"));
flow(&stores).record("alpha", first_second, &push(7, "SHA256:aaa"));
flow(&stores).flush(first_second).expect("flush");
let series = day_series(&stores, "alpha");
assert_eq!(
series.len(),
2,
"one push each side of midnight: {series:?}"
);
assert_eq!(series[0].bytes_in, 1_000);
assert_eq!(series[0].pushes, 1);
assert_eq!(series[1].bytes_in, 7);
assert_eq!(series[1].pushes, 1);
}
/// **Deleting a namespace forgets its activity**, including which key
/// fingerprints pushed to it — and leaves every other tenant's alone.
#[test]
fn deleting_a_namespace_forgets_its_activity() {
let (_dir, stores) = box_with(&["alpha", "beta"]);
flow(&stores).record("alpha", noon(), &push(4_096, "SHA256:aaa"));
flow(&stores).record("beta", noon(), &push(8, "SHA256:bbb"));
flow(&stores).flush(noon()).expect("flush");
let plane = CoreControlPlane::new(Arc::clone(&stores));
plane.delete_namespace("alpha", true).expect("delete");
assert!(
flow(&stores)
.store()
.series("alpha", i64::MIN / 2, i64::MAX / 2)
.expect("series")
.is_empty(),
"a deleted tenant must not leave a record of who worked on it"
);
assert_eq!(
day_series(&stores, "beta").len(),
1,
"the neighbour's series was taken with it"
);
}
/// The box's own fold, which the dashboard band draws: what moved today across
/// every namespace, and how many tenants have gone quiet.
#[test]
fn the_fleet_fold_counts_the_silent_tenants() {
let (_dir, stores) = box_with(&["busy", "quiet", "never"]);
flow(&stores).record("busy", noon(), &push(4_096, "SHA256:aaa"));
flow(&stores).record("busy", noon(), &fetch(1_024, "SHA256:aaa"));
// Seen once, long ago.
let long_ago = noon() - 30 * DAY;
flow(&stores).record("quiet", long_ago, &push(1, "SHA256:bbb"));
flow(&stores).flush(long_ago).expect("flush");
// `never` has no traffic at all and no row, so the fold has to find it
// through the live map or not at all — which is why it is here.
flow(&stores).record("never", noon(), &Transfer::default());
let fleet = flow(&stores).fleet(noon(), 14);
assert_eq!(fleet.bytes_in_today, 4_096);
assert_eq!(fleet.bytes_out_today, 1_024);
assert_eq!(
fleet.silent, 1,
"only `quiet` has a last-seen older than the window; `busy` moved today and `never` \
has no row at all so the fold cannot see it: {fleet:?}"
);
assert!(fleet.namespaces >= 2, "{fleet:?}");
}
/// Flow is on EVERY build. `Usage` is not behind `metered` and neither is this:
/// a self-hoster reading their own disk wants to know what moved through it as
/// much as anybody billing from it does.
#[test]
fn the_repositories_are_beside_the_point() {
let (_dir, stores) = box_with(&["alpha"]);
// No repository is ever created. The first-repository door means a
// namespace can be pushed to before it holds anything, and a meter that
// needed a catalog row would report that push as nobody's.
assert!(
stores
.catalog()
.list(Some("alpha"))
.expect("list")
.is_empty(),
"the fixture must hold no repository, or this proves nothing"
);
flow(&stores).record("alpha", noon(), &push(4_096, "SHA256:aaa"));
let alpha = meters(&stores, "alpha");
assert_eq!(alpha.pack_bytes, 0, "nothing is held");
assert_eq!(alpha.bytes_in_today, 4_096, "and something moved anyway");
}
/// **The hot path allocates nothing once a namespace has a slot.**
///
/// The flow counters sit on the transfer path, so the rule is the doctrine's:
/// preallocate and reuse, never allocate per request. A slot is made and seeded
/// on the FIRST transfer of a namespace-day — one `Arc`, one map entry, one
/// store read — and every transfer after it is a read lock, a hash lookup and a
/// handful of relaxed `fetch_add`s.
///
/// Counted with a counting global allocator rather than argued: the assertion
/// is that the second thousand transfers allocate strictly less than the first
/// one did, and in fact allocate only what `Transfer::principal`'s own `String`
/// costs — which the caller built, not the ledger.
///
/// **The count is PER THREAD**, and it has to be — see [`counting_alloc`]. A
/// process-wide counter measured the other thirteen tests in this file as well,
/// because cargo runs them concurrently in this same binary.
#[test]
fn the_counting_path_does_not_allocate_per_transfer() {
let (_dir, stores) = box_with(&["alpha"]);
let ledger = flow(&stores);
// Warm the slot: the first transfer of a namespace-day makes it.
ledger.record("alpha", noon(), &push(1, "SHA256:aaa"));
let transfer = push(4_096, "SHA256:aaa");
let before = counting_alloc::allocations();
for _ in 0..1_000 {
ledger.record("alpha", noon(), &transfer);
}
let per_transfer = (counting_alloc::allocations() - before) as f64 / 1_000.0;
assert!(
per_transfer <= 1.0,
"the counting path allocates {per_transfer} times per transfer; it must reuse the \
namespace's slot rather than build one"
);
assert_eq!(
meters(&stores, "alpha").pushes_today,
1_001,
"and it counted every one of them"
);
}
/// A counting global allocator, for the assertion above and nothing else.
///
/// It is `#[cfg(test)]`-only by construction — this is a test target — so the
/// shipped binary carries no counting wrapper.
///
/// # The counter is PER THREAD, and a process-wide one was a false red
///
/// This counted into a single `AtomicUsize` until 2026-09-10, which measured
/// **every thread in the process**. A global allocator sees the whole binary,
/// and cargo runs the fourteen tests in this file CONCURRENTLY in it — so the
/// thousand-transfer loop was charged for whatever the other thirteen happened
/// to allocate while it ran. MEASURED, in `cargo test --workspace` on a loaded
/// box:
///
/// ```text
/// the counting path allocates 3.949 times per transfer; it must reuse the
/// namespace's slot rather than build one
/// ```
///
/// The ledger had not allocated 3.949 times per transfer. It had allocated
/// approximately nothing, and the neighbours' `tempfile`, redb and `String`
/// traffic made up the rest. The same test passed when the file was run on its
/// own, and passed in `cargo test -p gunnar-server --test flow_meters` — which
/// is the worst shape a test can have: **green when you check it, red in CI**,
/// and pointing at innocent code when it goes.
///
/// A thread-local `Cell` fixes it at the root: the loop is charged for its own
/// thread and nothing else, and the number stops depending on how busy the box
/// is. `const`-initialised so the TLS slot itself never heap-allocates (a
/// lazily-initialised thread-local inside a global allocator recurses), and
/// `try_with` so an allocation during TLS teardown returns `Err` instead of
/// panicking inside `alloc`.
mod counting_alloc {
use std::alloc::{GlobalAlloc, Layout, System};
use std::cell::Cell;
thread_local! {
static ALLOCATIONS: Cell<usize> = const { Cell::new(0) };
}
/// Charge one allocation to the CURRENT thread, and never fail: during
/// thread-local destruction the slot is gone and `try_with` says so rather
/// than allocating to report it.
fn bump() {
let _ = ALLOCATIONS.try_with(|n| n.set(n.get().wrapping_add(1)));
}
pub struct Counting;
unsafe impl GlobalAlloc for Counting {
unsafe fn alloc(&self, layout: Layout) -> *mut u8 {
bump();
unsafe { System.alloc(layout) }
}
unsafe fn dealloc(&self, ptr: *mut u8, layout: Layout) {
unsafe { System.dealloc(ptr, layout) }
}
unsafe fn realloc(&self, ptr: *mut u8, layout: Layout, new_size: usize) -> *mut u8 {
bump();
unsafe { System.realloc(ptr, layout, new_size) }
}
}
#[global_allocator]
static ALLOC: Counting = Counting;
/// Allocations charged to THIS thread so far. Both reads in a measurement
/// must happen on the same thread, which is the thread the test body runs
/// on.
pub fn allocations() -> usize {
ALLOCATIONS.with(Cell::get)
}
}