crates/gunnar-server/src/usage_ledger.rs
//! **The usage ledger: every namespace's stock meters, kept current by
//! re-measuring only what changed** (2026-10-06).
//!
//! # Why
//!
//! monetize's poller asked `Usage.Read` once per tenant every 300 s, and every
//! one of those reads listed the WHOLE catalog (`RepoCatalog::list` is
//! O(repositories)) and walked the WHOLE repository root for tombstones
//! (`ServerStores::tombstones`). One poll was therefore O(tenants²) disk and
//! catalog work: MEASURED on the UpCloud mockup's load test at ~25–30 s of 1.4
//! cores on main per period at 1,000 tenants, and on oden's debug bench
//! (`tests/usage_poll_scale.rs`) 8.5 ms a call at 1,000 tenants — at ~10,000
//! tenants one poll took the whole period.
//!
//! # What it keeps
//!
//! Per namespace: its repositories (from the catalog, re-indexed only when the
//! catalog document's VERSION changes — one pointer read per settle), each
//! repository's last measured [`RepoBytes`], the namespace's tombstoned bytes,
//! the indexed objects its OPEN stores hold, and the sequence number of the last
//! time any of that — or its flow — changed. A read of a namespace nothing has
//! touched is a copy out of memory.
//!
//! # What makes it re-measure (the marks)
//!
//! A repository's bytes can only change while its store is OPEN, or by a
//! store-level operation on its directory (soft delete, purge, reclaim, a
//! growth's remount). So:
//!
//! * **opening a store** marks the repository ([`ServerStores::repo_store`]'s
//! one open site), and **every store that is open, or was open at the last
//! look, is re-measured** at the next settle (the open set is bounded by the
//! handle cache, 256) — a push, a fetch that grows the exploded cache, a
//! background index drain, a gc: all happen on an open store;
//! * **a transfer, a refusal or an auth failure** marks the namespace through
//! the flow ledger's observer (the `*_today` meters moved; LFS bytes may have);
//! * **the catalog's version** changing re-indexes: a namespace or repository
//! created or deleted is a change;
//! * **tombstone operations** mark their namespace; a **growth's resume** marks
//! everything;
//! * and the **reconcile pass** ([`UsageLedger::reconcile_step`], on the flow
//! flush's one-minute tick) re-measures a slice of namespaces every tick, so
//! anything that moved bytes without a mark is corrected within one cycle and
//! COUNTED ([`UsageLedger::drift_corrections`]) rather than silently absorbed.
//!
//! # The cursor
//!
//! `<epoch>.<utc day>.<seq>`. `epoch` is this ledger's (a restart is a new
//! ledger and every old cursor reads as "everything"); the day is in it because
//! the `*_today` meters reset at midnight UTC without any event; `seq` is the
//! ledger's change counter, taken BEFORE the read settles so a change that lands
//! during the read is answered again next time rather than lost.
use std::collections::{BTreeMap, BTreeSet, HashMap, HashSet};
use std::ops::Bound;
use std::path::PathBuf;
use std::sync::atomic::{AtomicU64, Ordering};
use std::sync::{Mutex, MutexGuard};
use std::time::Duration;
use gunnar_core::RepoRef;
use crate::control_plane::{repo_bytes, RepoBytes};
use crate::stores::{dir_bytes, ServerStores, TombstoneReason};
/// The reconcile pass's cycle: every namespace re-measured at least this often.
pub const RECONCILE_CYCLE: Duration = Duration::from_secs(30 * 60);
/// Marks taken on hot paths: a set insert under a short lock, never I/O.
#[derive(Default)]
struct Marks {
repos: HashSet<(String, String)>,
namespaces: HashSet<String>,
flow: HashSet<String>,
all: bool,
}
#[derive(Debug, Clone, Default)]
struct RepoRow {
/// `None` when the catalogued path did not resolve (an illegal or escaping
/// path): such a repository weighs nothing here, as it did in the walk.
dir: Option<PathBuf>,
/// `None` until first measured.
bytes: Option<RepoBytes>,
}
#[derive(Debug, Default)]
struct Ns {
repos: BTreeMap<String, RepoRow>,
/// `None` until first measured.
tombstoned: Option<u64>,
/// Indexed objects of this namespace's OPEN stores, as of the last sweep.
objects: u64,
/// The ledger sequence at which this row last changed.
changed_seq: u64,
}
impl Ns {
fn unmeasured(&self) -> bool {
self.tombstoned.is_none()
|| self
.repos
.values()
.any(|r| r.dir.is_some() && r.bytes.is_none())
}
}
#[derive(Default)]
struct State {
/// The catalog version the index was built from; `None` = never built.
catalog: Option<Option<gunnar_core::BlobId>>,
ns: BTreeMap<String, Ns>,
/// Git dir → (namespace, repository), for the open-store set.
by_dir: HashMap<PathBuf, (String, String)>,
open_last: HashSet<PathBuf>,
/// Where the reconcile pass resumes.
reconcile_after: String,
}
/// Work a settle found to do, carried out with no lock held.
#[derive(Default)]
struct Work {
repos: Vec<(String, String, PathBuf)>,
tombstones: BTreeSet<String>,
}
/// [`UsageLedger::page`]'s answer: the rows (namespace, what it holds), whether
/// there are more, and whether a `changed_since` was honoured.
pub type Page = (Vec<(String, Held)>, bool, bool);
/// [`UsageLedger::exact`]'s answer: the namespace's total and its repositories'
/// bytes (name, bytes, store directory), sorted by name.
pub type Exact = (Held, Vec<(String, RepoBytes, Option<PathBuf>)>);
/// One namespace's stock meters as the ledger holds them.
#[derive(Debug, Clone, Default, PartialEq, Eq)]
pub struct Held {
pub bytes: RepoBytes,
pub repos: u64,
pub tombstoned_bytes: u64,
pub objects: u64,
}
/// See the module doc.
pub struct UsageLedger {
epoch: String,
seq: AtomicU64,
marks: Mutex<Marks>,
state: Mutex<State>,
/// One settle at a time: two would walk the same repositories twice.
settling: Mutex<()>,
drift: AtomicU64,
}
impl std::fmt::Debug for UsageLedger {
fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
f.debug_struct("UsageLedger")
.field("epoch", &self.epoch)
.field("seq", &self.seq.load(Ordering::Relaxed))
.finish_non_exhaustive()
}
}
fn lock<T>(m: &Mutex<T>) -> MutexGuard<'_, T> {
// A cache: whatever a panicking holder left is still a set of readings,
// and the next settle re-measures what it needs.
m.lock().unwrap_or_else(|p| p.into_inner())
}
/// Bytes still held by this namespace's soft-deleted repositories — the
/// tombstones directly in its own directory (namespaces nest, and a nested
/// namespace's tombstones are its own). The walk `ServerStores::tombstones`
/// makes over the whole root, narrowed to one directory.
pub fn tombstoned_bytes(stores: &ServerStores, namespace: &str) -> u64 {
let dir = stores.repos_root().join(namespace);
let Ok(entries) = std::fs::read_dir(&dir) else {
return 0;
};
let mut total = 0;
for entry in entries.flatten() {
let name = entry.file_name().to_string_lossy().into_owned();
let Some(store) = name.strip_suffix(".tombstone") else {
continue;
};
if !store.ends_with(".git") {
continue;
}
let Some(record) = ServerStores::read_tombstone(&entry.path()) else {
continue;
};
if record.reason != TombstoneReason::Deleted {
continue;
}
total += dir_bytes(&dir.join(store));
}
total
}
/// `<epoch>.<day>.<seq>`, or `None` for anything else.
fn parse_cursor(cursor: &str) -> Option<(&str, i64, u64)> {
let mut parts = cursor.split('.');
let (epoch, day, seq) = (parts.next()?, parts.next()?, parts.next()?);
if parts.next().is_some() || epoch.is_empty() {
return None;
}
Some((epoch, day.parse().ok()?, seq.parse().ok()?))
}
impl Default for UsageLedger {
fn default() -> Self {
Self::new()
}
}
impl UsageLedger {
pub fn new() -> Self {
static LEDGERS: AtomicU64 = AtomicU64::new(0);
let n = LEDGERS.fetch_add(1, Ordering::Relaxed);
let epoch = format!("{:x}{:x}{n:x}", gunnar_core::now_ms(), std::process::id());
UsageLedger {
epoch,
seq: AtomicU64::new(0),
marks: Mutex::new(Marks::default()),
state: Mutex::new(State::default()),
settling: Mutex::new(()),
drift: AtomicU64::new(0),
}
}
/// This ledger's epoch: a cursor from another is "everything".
pub fn epoch(&self) -> &str {
&self.epoch
}
/// How many re-measurements by the reconcile pass found a namespace
/// different from what the ledger held — a change no mark reported. Zero
/// on a box whose marks are complete.
pub fn drift_corrections(&self) -> u64 {
self.drift.load(Ordering::Relaxed)
}
// ── the marks: cheap, from any path ─────────────────────────────────────
/// This repository's bytes may have changed.
pub fn mark_repo(&self, namespace: &str, name: &str) {
lock(&self.marks)
.repos
.insert((namespace.to_owned(), name.to_owned()));
}
/// Everything about this namespace may have changed: its repositories'
/// bytes and its tombstones.
pub fn mark_namespace(&self, namespace: &str) {
lock(&self.marks).namespaces.insert(namespace.to_owned());
}
/// This namespace's flow moved (the flow ledger's observer).
pub fn mark_flow(&self, namespace: &str) {
lock(&self.marks).flow.insert(namespace.to_owned());
}
/// Everything may have changed (a growth's remount, a role change).
pub fn mark_all(&self) {
lock(&self.marks).all = true;
}
fn bump(&self, ns: &mut Ns) {
ns.changed_seq = self.seq.fetch_add(1, Ordering::SeqCst) + 1;
}
/// The cursor as of now.
pub fn cursor_now(&self, now_unix_ms: i64) -> String {
format!(
"{}.{}.{}",
self.epoch,
gunnar_core::day_of(now_unix_ms),
self.seq.load(Ordering::SeqCst)
)
}
/// The sequence a cursor names, when this ledger can honour it today.
fn honour(&self, cursor: &str, now_unix_ms: i64) -> Option<u64> {
let (epoch, day, seq) = parse_cursor(cursor)?;
(epoch == self.epoch && day == gunnar_core::day_of(now_unix_ms)).then_some(seq)
}
// ── the index ───────────────────────────────────────────────────────────
/// Rebuild the namespace → repositories index when the catalog changed.
fn refresh_index(
&self,
stores: &ServerStores,
st: &mut State,
marks: &mut Marks,
) -> gunnar_core::Result<()> {
let catalog = stores.catalog();
let version = catalog.version()?;
if st.catalog == Some(version) {
return Ok(());
}
let initial = st.catalog.is_none();
let mut want: BTreeMap<String, BTreeSet<String>> = catalog
.namespaces()?
.into_iter()
.map(|n| (n.name, BTreeSet::new()))
.collect();
for repo in catalog.list(None)? {
want.entry(repo.namespace.clone())
.or_default()
.insert(repo.name.clone());
}
let gone: Vec<String> = st
.ns
.keys()
.filter(|n| !want.contains_key(*n))
.cloned()
.collect();
for name in gone {
if let Some(ns) = st.ns.remove(&name) {
for row in ns.repos.values() {
if let Some(dir) = &row.dir {
st.by_dir.remove(dir);
}
}
}
}
for (name, set) in want {
let fresh = !st.ns.contains_key(&name);
let mut ns = st.ns.remove(&name).unwrap_or_default();
let old: BTreeSet<String> = ns.repos.keys().cloned().collect();
if old != set {
for removed in old.difference(&set) {
if let Some(row) = ns.repos.remove(removed) {
if let Some(dir) = row.dir {
st.by_dir.remove(&dir);
}
}
}
for added in set.difference(&old) {
let dir = RepoRef::new(&name, added)
.ok()
.and_then(|path| stores.repo_dir(&path).ok());
if let Some(dir) = &dir {
st.by_dir.insert(dir.clone(), (name.clone(), added.clone()));
}
ns.repos.insert(added.clone(), RepoRow { dir, bytes: None });
}
// A deleted repository leaves a tombstone; a created one may
// replace one.
marks.namespaces.insert(name.clone());
if !initial {
self.bump(&mut ns);
}
} else if fresh && !initial {
self.bump(&mut ns);
}
st.ns.insert(name, ns);
}
st.catalog = Some(version);
Ok(())
}
/// What the marks (and, when due, the open-store set) say to re-measure.
/// Flow marks bump here: their meters are live and need no walk.
fn collect(&self, stores: &ServerStores, st: &mut State, marks: Marks, sweep: bool) -> Work {
let mut work = Work::default();
let mut repos: BTreeSet<(String, String)> = BTreeSet::new();
let mut whole: BTreeSet<String> = BTreeSet::new();
if marks.all {
whole.extend(st.ns.keys().cloned());
}
whole.extend(marks.namespaces);
for name in marks.flow {
if let Some(ns) = st.ns.get_mut(&name) {
self.bump(ns);
whole.insert(name);
}
}
repos.extend(marks.repos);
// **The open-store set, every settle** — it is the registry's live
// list, a few hundred entries at most and no disk. A store that OPENED
// or CLOSED since the last look is re-measured (a store's final bytes
// are on the disk only once the engine has let it go: it may trim what
// it had reserved), and every open store is re-measured when the caller
// asks for a full look (`sweep`: the first page of a read — the later
// pages of the same read follow it back to back and need no second).
let now_open: HashMap<PathBuf, u64> =
stores.engine().open_object_counts().into_iter().collect();
let full = sweep;
let mut objects: HashMap<String, u64> = HashMap::new();
for (dir, count) in &now_open {
if let Some((ns, repo)) = st.by_dir.get(dir) {
*objects.entry(ns.clone()).or_default() += count;
if full || !st.open_last.contains(dir) {
repos.insert((ns.clone(), repo.clone()));
}
}
}
for dir in st.open_last.iter().filter(|d| !now_open.contains_key(*d)) {
if let Some((ns, repo)) = st.by_dir.get(dir) {
repos.insert((ns.clone(), repo.clone()));
}
}
let names: Vec<String> = st
.ns
.iter()
.filter(|(_, ns)| ns.objects > 0)
.map(|(n, _)| n.clone())
.chain(objects.keys().cloned())
.collect();
for name in names {
let now = objects.get(&name).copied().unwrap_or(0);
if let Some(ns) = st.ns.get_mut(&name) {
if ns.objects != now {
ns.objects = now;
self.bump(ns);
}
}
}
st.open_last = now_open.into_keys().collect();
for name in &whole {
if let Some(ns) = st.ns.get(name) {
for (repo, row) in &ns.repos {
if let Some(dir) = &row.dir {
work.repos.push((name.clone(), repo.clone(), dir.clone()));
}
}
work.tombstones.insert(name.clone());
}
}
for (name, repo) in repos {
if whole.contains(&name) {
continue;
}
if let Some(dir) = st
.ns
.get(&name)
.and_then(|ns| ns.repos.get(&repo))
.and_then(|row| row.dir.clone())
{
work.repos.push((name, repo, dir));
}
}
work
}
/// Measure `work` with no lock held, then fold it in; a row that changed
/// (or was measured for the first time) gets a new sequence. Returns the
/// namespaces whose row changed.
fn carry_out(&self, stores: &ServerStores, work: Work) -> BTreeSet<String> {
let repos: Vec<(String, String, RepoBytes)> = work
.repos
.into_iter()
.map(|(ns, repo, dir)| (ns, repo, repo_bytes(&dir)))
.collect();
let tombstones: Vec<(String, u64)> = work
.tombstones
.into_iter()
.map(|ns| {
let t = tombstoned_bytes(stores, &ns);
(ns, t)
})
.collect();
let mut changed = BTreeSet::new();
let mut st = lock(&self.state);
for (name, repo, bytes) in repos {
let Some(ns) = st.ns.get_mut(&name) else {
continue;
};
let Some(row) = ns.repos.get_mut(&repo) else {
continue;
};
if row.bytes != Some(bytes) {
row.bytes = Some(bytes);
self.bump(ns);
changed.insert(name);
}
}
for (name, bytes) in tombstones {
let Some(ns) = st.ns.get_mut(&name) else {
continue;
};
if ns.tombstoned != Some(bytes) {
ns.tombstoned = Some(bytes);
self.bump(ns);
changed.insert(name);
}
}
changed
}
/// **Bring the ledger up to date with every mark and with the open-store
/// set** — every store that opened or closed since the last settle, and,
/// with `sweep`, every store that is open.
pub fn settle(&self, stores: &ServerStores, sweep: bool) -> gunnar_core::Result<()> {
let _one = lock(&self.settling);
let marks = std::mem::take(&mut *lock(&self.marks));
let work = {
let mut st = lock(&self.state);
let mut marks = marks;
if let Err(err) = self.refresh_index(stores, &mut st, &mut marks) {
// Put the marks back: nothing they asked for was done.
let mut back = lock(&self.marks);
back.repos.extend(marks.repos);
back.namespaces.extend(marks.namespaces);
back.flow.extend(marks.flow);
back.all |= marks.all;
return Err(err);
}
self.collect(stores, &mut st, marks, sweep)
};
self.carry_out(stores, work);
Ok(())
}
/// Measure whichever of `names` has never been measured.
fn measure_unmeasured(&self, stores: &ServerStores, names: &[String]) {
let work = {
let st = lock(&self.state);
let mut work = Work::default();
for name in names {
let Some(ns) = st.ns.get(name) else { continue };
if !ns.unmeasured() {
continue;
}
for (repo, row) in &ns.repos {
if let (Some(dir), None) = (&row.dir, &row.bytes) {
work.repos.push((name.clone(), repo.clone(), dir.clone()));
}
}
if ns.tombstoned.is_none() {
work.tombstones.insert(name.clone());
}
}
work
};
self.carry_out(stores, work);
}
fn held_of(ns: &Ns) -> Held {
let mut held = Held {
tombstoned_bytes: ns.tombstoned.unwrap_or(0),
objects: ns.objects,
..Held::default()
};
for row in ns.repos.values() {
held.repos += 1;
if let Some(b) = row.bytes {
held.bytes.pack += b.pack;
held.bytes.cache += b.cache;
held.bytes.lfs += b.lfs;
}
}
held
}
// ── the reads ───────────────────────────────────────────────────────────
/// **One page**, in namespace order after `after`, of at most `page_size`
/// namespaces — every one, or with `since` only those that changed after
/// it. Settles first. Returns the rows' namespaces with what is held, and
/// whether there are more.
pub fn page(
&self,
stores: &ServerStores,
after: &str,
page_size: usize,
changed_since: &str,
now_unix_ms: i64,
) -> gunnar_core::Result<Page> {
self.settle(stores, after.is_empty())?;
let since = self.honour(changed_since, now_unix_ms);
let page_size = page_size.max(1);
let (picked, more) = {
let st = lock(&self.state);
let range = if after.is_empty() {
st.ns.range::<str, _>((Bound::Unbounded, Bound::Unbounded))
} else {
st.ns
.range::<str, _>((Bound::Excluded(after), Bound::Unbounded))
};
let mut picked = Vec::new();
let mut more = false;
for (name, ns) in range {
if since.is_some_and(|s| ns.changed_seq <= s) {
continue;
}
if picked.len() == page_size {
more = true;
break;
}
picked.push(name.clone());
}
(picked, more)
};
self.measure_unmeasured(stores, &picked);
let st = lock(&self.state);
let rows = picked
.into_iter()
.filter_map(|name| st.ns.get(&name).map(|ns| (name.clone(), Self::held_of(ns))))
.collect();
Ok((rows, more, since.is_some()))
}
/// **One namespace, measured now** — what `Usage.Read` answers: exact, its
/// own repositories and tombstones walked, nothing else (the index answers
/// which repositories are its). `None` for a namespace the catalog does not
/// hold. Returns the per-repository bytes too, sorted by name.
pub fn exact(
&self,
stores: &ServerStores,
namespace: &str,
) -> gunnar_core::Result<Option<Exact>> {
// The index first (cheap unless the catalog changed), without taking
// the marks: they stay for the next settle.
{
let _one = lock(&self.settling);
let mut st = lock(&self.state);
let mut marks = Marks::default();
self.refresh_index(stores, &mut st, &mut marks)?;
// A changed repository SET is honoured by the walk below; the
// tombstone marks it raised are folded back for the next settle.
if !marks.namespaces.is_empty() {
lock(&self.marks).namespaces.extend(marks.namespaces);
}
}
let work = {
let st = lock(&self.state);
let Some(ns) = st.ns.get(namespace) else {
return Ok(None);
};
let mut work = Work::default();
for (repo, row) in &ns.repos {
if let Some(dir) = &row.dir {
work.repos
.push((namespace.to_owned(), repo.clone(), dir.clone()));
}
}
work.tombstones.insert(namespace.to_owned());
work
};
self.carry_out(stores, work);
let st = lock(&self.state);
let Some(ns) = st.ns.get(namespace) else {
return Ok(None);
};
let rows = ns
.repos
.iter()
.map(|(name, row)| (name.clone(), row.bytes.unwrap_or_default(), row.dir.clone()))
.collect();
Ok(Some((Self::held_of(ns), rows)))
}
/// **The reconcile pass**: re-measure the next slice of namespaces (sized so
/// the whole set is covered every [`RECONCILE_CYCLE`] at a `tick` cadence),
/// as if each had been marked. A difference found here is a change no mark
/// reported: it is corrected (and answered to the next changed-since read)
/// and counted. Returns `(namespaces looked at, corrected)`.
///
/// **A pass that cannot run is an `Err`, never a quiet `(0, 0)`** (A28 M17): the settle
/// before the slice reads the catalog, and when it cannot be read nothing is re-measured.
/// Its caller ([`crate::stores::reconcile_tick`]) counts the miss
/// (`usage_reconcile_missed` on `/metrics` and in the heartbeat) and warns.
pub fn reconcile_step(
&self,
stores: &ServerStores,
tick: Duration,
) -> gunnar_core::Result<(usize, usize)> {
self.settle(stores, false)?;
let slice = {
let mut st = lock(&self.state);
let total = st.ns.len();
if total == 0 {
return Ok((0, 0));
}
let ticks = (RECONCILE_CYCLE.as_secs() / tick.as_secs().max(1)).max(1) as usize;
let want = total.div_ceil(ticks).max(1);
let after = st.reconcile_after.clone();
let mut slice: Vec<String> = st
.ns
.range::<str, _>((
if after.is_empty() {
Bound::Unbounded
} else {
Bound::Excluded(after.as_str())
},
Bound::Unbounded,
))
.take(want)
.map(|(n, _)| n.clone())
.collect();
if slice.len() < want {
// Wrap around to the start.
let have: BTreeSet<String> = slice.iter().cloned().collect();
slice.extend(
st.ns
.keys()
.filter(|n| !have.contains(*n))
.take(want - slice.len())
.cloned(),
);
}
st.reconcile_after = slice.last().cloned().unwrap_or_default();
slice
};
let work = {
let st = lock(&self.state);
let mut work = Work::default();
for name in &slice {
let Some(ns) = st.ns.get(name) else { continue };
if ns.unmeasured() {
// Never measured: nothing held to be wrong about yet.
continue;
}
for (repo, row) in &ns.repos {
if let Some(dir) = &row.dir {
work.repos.push((name.clone(), repo.clone(), dir.clone()));
}
}
work.tombstones.insert(name.clone());
}
work
};
let looked = slice.len();
let corrected = self.carry_out(stores, work);
if !corrected.is_empty() {
self.drift
.fetch_add(corrected.len() as u64, Ordering::Relaxed);
tracing::info!(
target: gunnar_core::observe::TARGET_STORE,
event = "usage.reconciled",
corrected = corrected.len(),
first = %corrected.iter().next().cloned().unwrap_or_default(),
"the usage ledger's reconcile pass found {} namespace(s) whose bytes had moved without a mark; corrected",
corrected.len()
);
}
Ok((looked, corrected.len()))
}
}
#[cfg(test)]
mod tests {
use super::*;
#[test]
fn a_cursor_is_honoured_only_by_its_own_ledger_on_its_own_day() {
let ledger = UsageLedger::new();
let other = UsageLedger::new();
let noon = gunnar_core::midnight_of(20_000) + 12 * 3_600_000;
let cursor = ledger.cursor_now(noon);
assert_eq!(ledger.honour(&cursor, noon), Some(0));
assert_eq!(
other.honour(&cursor, noon),
None,
"another ledger (a restart) answers everything"
);
assert_eq!(
ledger.honour(&cursor, noon + 86_400_000),
None,
"the day turned: the *_today meters reset"
);
assert_eq!(ledger.honour("", noon), None);
assert_eq!(ledger.honour("x.y.z", noon), None);
assert_ne!(ledger.epoch(), other.epoch());
}
}