gunnar_server/

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());
    }
}