gunnar-core 3.0.0

//! **The flow ledger: what MOVED through a namespace, per UTC day.**
//!
//! # The hole this fills
//!
//! Every meter gunnar had before this file was a **stock** meter: `pack_bytes`,
//! `cache_bytes`, `cold_bytes`, `lfs_bytes`, `tombstoned_bytes`, `objects` and
//! `repos` all answer *how much is sitting on the disk right now*. They are the
//! right numbers to bill from and the wrong ones to run a service on, because
//! every one of them reads the same for a tenant who pushed this morning and a
//! tenant who last touched the server in March. A silent customer and a busy
//! one are indistinguishable in a stock meter, and so are a customer who is
//! working and a customer who is being refused at the door.
//!
//! This module counts the other kind: **flow** — bytes in, bytes out, pushes,
//! fetches, refusals, and when the tenant was last seen. Nothing here is ever a
//! wall. Every counter is `Enforcement::Measured` in monetize's vocabulary
//! (`monetize_product::Enforcement`, which gunnar took onto the wire in
//! `usage.proto` on purpose): they are counted and reported, and nothing on this
//! box refuses because of one. Promoting any of them to `Enforced` would need a
//! refusal to exist first, and there is none.
//!
//! # Two halves, and why
//!
//! * [`FlowLedger`] is the LIVE half: pre-sized atomics, one slot per namespace
//!   that has been touched since boot. A transfer bumps a handful of
//!   `fetch_add`s under a read lock and allocates nothing. This is what the
//!   transfer paths call.
//! * [`FlowStore`] is the KEPT half: one document per namespace in the control
//!   store (redb, the same store family the catalog, the auth table and the
//!   entitlement facts live in), holding up to [`RETENTION_DAYS`] days of
//!   [`FlowDay`] rows.
//!
//! A periodic flush copies the live slot into the store. **It copies, it does
//! not add**: the live slot holds the whole of today, and the stored row for
//! today is a snapshot of it, so a flush that runs twice writes the same number
//! twice rather than doubling it. That is the property that makes a restart
//! safe in both directions — nothing is counted twice, and nothing older than
//! one flush interval is lost, because a slot opened after a restart is SEEDED
//! from the stored row for today rather than from zero.
//!
//! # The day boundary
//!
//! Days are UTC, numbered as whole days since the Unix epoch ([`day_of`]). A
//! transfer is charged to the day its *record* is taken, which is the day it
//! finished. A transfer that starts at 23:59 and ends at 00:01 lands wholly in
//! the second day; splitting one transfer across two days would need the byte
//! counters to be sampled on a clock, and a meter that is sampled is a meter
//! that can be read while it is being written.
//!
//! # What is NOT here
//!
//! **A key that resolves to nobody has no namespace**, and this ledger is per
//! namespace. `ssh.auth.refused` with `reason = "unknown-key"` happens before
//! any repository is named — there is no exec line yet — so it cannot be
//! charged to a tenant and is not counted here. It is counted server-wide by
//! `ServerMetrics::add_connections_refused`, where it belongs.
//! [`FlowDay::auth_failures`] is the OTHER half of the same sentence: a
//! principal that authenticated and then held no grant on the repository it
//! asked for. That one names a namespace, so that one is a tenant's number.

use std::collections::{BTreeMap, BTreeSet, HashMap};
use std::sync::atomic::{AtomicI64, AtomicU64, Ordering};
use std::sync::{Arc, Mutex, RwLock};

use crate::control::ControlStore;
use crate::doc::{impl_doc, DocRef, ShardedMap};
use crate::error::Result;

/// **The name the flow document is stored under** in the control store.
pub const FLOW_REF: &str = "refs/gunnar/flow";

/// Milliseconds in a day. UTC, no leap seconds: the wire carries Unix
/// milliseconds and a Unix millisecond has never counted one.
pub const DAY_MS: i64 = 86_400_000;

/// **How many days of flow are kept per namespace: 400.**
///
/// The same window `monetize_server::state::UsageStore::RETENTION_MS` keeps its
/// readings for, and for the same reason: a shade over a year, so a question
/// about "this month last year" can still be answered, and bounded, so a table
/// written on every transfer is not a disk-fill bug with a schedule.
pub const RETENTION_DAYS: i64 = 400;

/// **The most principals one namespace-day counts by name.**
///
/// [`FlowDay::principals`] is an EXACT set rather than a sketch. A HyperLogLog
/// would answer "roughly how many" in a fixed 1–2 KiB; a `BTreeSet` of
/// fingerprints answers "exactly how many, and which" in about 50 B each. At
/// the scale a git server has — a namespace is a team, and a team with ten
/// thousand distinct keys pushing in one day is not a team — exact costs less
/// than the sketch and answers a strictly harder question, which is the whole
/// argument. The ceiling exists so that the pathological case is bounded rather
/// than unbounded, and when it is hit [`FlowDay::principals_capped`] says so,
/// because a count that stopped growing must never be read as a total.
pub const PRINCIPAL_CEILING: usize = 10_000;

/// The UTC day a millisecond timestamp falls in, as whole days since the epoch.
/// Euclidean, so a timestamp before 1970 floors rather than truncating toward
/// zero and cannot share a day number with one after it.
pub fn day_of(unix_ms: i64) -> i64 {
    unix_ms.div_euclid(DAY_MS)
}

/// Midnight UTC of a day number, in Unix milliseconds — the value the wire
/// carries as `DayActivity.day_unix_ms`.
pub fn midnight_of(day: i64) -> i64 {
    day.saturating_mul(DAY_MS)
}

/// **Unix milliseconds → `YYYY-MM-DDTHH:MM:SSZ`, in std alone.**
///
/// Eleven lines of arithmetic rather than a date crate in every consumer's
/// graph, and the ONE runtime copy of it: `gunnar control usage` and `gunnar
/// control activity` both print timestamps off this, so the two cannot render
/// the same instant two ways. (`gunnar-common/build.rs` has the same
/// arithmetic and must: a build script cannot link the crate it is building.)
///
/// The civil date is Howard Hinnant's `civil_from_days`, exact for every
/// representable day and needing no table. Pre-1970 falls out correctly because
/// the era arithmetic floors.
pub fn iso8601_utc(unix_ms: i64) -> String {
    let secs = unix_ms.div_euclid(1000);
    let days = secs.div_euclid(86_400);
    let rem = secs.rem_euclid(86_400);
    let (hh, mm, ss) = (rem / 3600, (rem % 3600) / 60, rem % 60);
    // Shift the epoch to 0000-03-01 so leap days land at the end of the cycle.
    let z = days + 719_468;
    let era = z.div_euclid(146_097);
    let doe = z.rem_euclid(146_097);
    let yoe = (doe - doe / 1460 + doe / 36_524 - doe / 146_096) / 365;
    let y = yoe + era * 400;
    let doy = doe - (365 * yoe + yoe / 4 - yoe / 100);
    let mp = (5 * doy + 2) / 153;
    let d = doy - (153 * mp + 2) / 5 + 1;
    let m = if mp < 10 { mp + 3 } else { mp - 9 };
    let y = if m <= 2 { y + 1 } else { y };
    format!("{y:04}-{m:02}-{d:02}T{hh:02}:{mm:02}:{ss:02}Z")
}

/// **Why a transfer was turned away**, in the three words an operator can act
/// on differently.
///
/// Deliberately three and not the gate's six. `gunnar_ssh::Refusal` has six
/// variants because the admission gate has six ceilings, and which of them was
/// hit is an operator's question about the BOX. A tenant's question is a
/// different one — *is the server busy, am I over my plan, or am I not allowed*
/// — and those are three answers. The six-way detail stays where it is, on the
/// gate's own `AdmissionStats` and in the `admission-*` reason on the log line.
#[derive(Clone, Copy, PartialEq, Eq, PartialOrd, Ord, Hash, Debug)]
pub enum Refusal {
    /// **The box was full.** Every `gunnar_ssh::Refusal` variant: the gate is at
    /// capacity, the principal's share is spent, a lane is full, the anonymous
    /// ceiling is reached, the keyed reserve is held, the memory budget is
    /// spent. Nothing about the tenant is wrong; the server said "not now".
    Admission,
    /// **The plan said no before a byte.** `PlanGate::pack_quota` refusing a
    /// namespace already at its `pack_bytes` cap, and the LFS allowance
    /// refusing an upload past `lfs_bytes`. This is the one an invoice can fix.
    Cap,
    /// **The push gate said no.** A protected ref, a required review, a
    /// signature that did not verify — `gunnar_policy`'s verdict. Nothing an
    /// invoice can fix and nothing the operator should page about.
    Policy,
}

impl Refusal {
    /// `admission` | `cap` | `policy` — the wire spelling, one writer. The same
    /// idiom as `Scope::name` and `Growth::name` on monetize's side: a word,
    /// never an enum's ordinal, because an ordinal is an encoding of the
    /// meaning rather than the meaning.
    pub fn name(self) -> &'static str {
        match self {
            Refusal::Admission => "admission",
            Refusal::Cap => "cap",
            Refusal::Policy => "policy",
        }
    }

    /// Every reason, in the order a breakdown is rendered in.
    pub const ALL: [Refusal; 3] = [Refusal::Admission, Refusal::Cap, Refusal::Policy];
}

/// **One namespace's flow for one UTC day.**
///
/// Every field is a count of something that HAPPENED. There is no field here
/// whose absence could be mistaken for a reading that was never taken: a day
/// with no row is a day with no traffic, and a row of zeroes is the same thing
/// said twice. That is the opposite of the stock meters, where a missing
/// reading and a zero mean different things — and the difference is that a
/// transfer either occurred or did not.
#[derive(Clone, Default, PartialEq, Eq, Debug, serde::Serialize, serde::Deserialize)]
#[serde(deny_unknown_fields)]
pub struct FlowDay {
    /// Bytes the server ACCEPTED off the wire: receive-pack, and the client
    /// half of every other conversation. Measured by the transport's own
    /// `Metered` reader, which is the one place a byte arriving is counted.
    #[serde(default)]
    pub bytes_in: u64,
    /// Bytes the server SERVED: upload-pack, the advertisement, and the archive
    /// stream. Measured by the transport's own `Metered` writer.
    #[serde(default)]
    pub bytes_out: u64,
    /// receive-pack sessions that finished without an error.
    #[serde(default)]
    pub pushes: u64,
    /// upload-pack sessions, clones included. The bare advertisement
    /// (`GET /info/refs`, `git ls-remote`) is NOT one: it reads the ref list and
    /// asks for no objects, and counting it would make a monitoring probe look
    /// like a customer.
    #[serde(default)]
    pub fetches: u64,
    /// LFS object bytes received. A subset of nothing: LFS rides its own
    /// conversation, so these bytes are also in [`FlowDay::bytes_in`] as wire
    /// bytes, and this is the applied figure the session itself counted.
    #[serde(default)]
    pub lfs_bytes_in: u64,
    /// LFS object bytes sent. See [`FlowDay::lfs_bytes_in`].
    #[serde(default)]
    pub lfs_bytes_out: u64,
    /// [`Refusal::Admission`] — the box was full.
    #[serde(default)]
    pub refused_admission: u64,
    /// [`Refusal::Cap`] — the plan refused.
    #[serde(default)]
    pub refused_cap: u64,
    /// [`Refusal::Policy`] — the push gate refused.
    #[serde(default)]
    pub refused_policy: u64,
    /// **A principal that authenticated and then held no grant here.** See the
    /// module doc for the half that is deliberately absent: a key that resolves
    /// to nobody names no namespace and is not a tenant's number.
    #[serde(default)]
    pub auth_failures: u64,
    /// The last transfer that put bytes IN, in Unix milliseconds; `0` for a day
    /// with no push. Kept per day rather than only in the latest row so a
    /// series answers *when in that week did they stop* and not only *have they
    /// stopped*.
    #[serde(default)]
    pub last_push_unix_ms: i64,
    /// The last transfer that took bytes OUT. `0` for a day with no fetch.
    #[serde(default)]
    pub last_fetch_unix_ms: i64,
    /// **Which principals were seen**, by the fingerprint the auth table knows
    /// them as. Exact, bounded by [`PRINCIPAL_CEILING`]; see there for why a
    /// set and not a sketch.
    #[serde(default)]
    pub principals: BTreeSet<String>,
    /// **The principal set stopped growing at [`PRINCIPAL_CEILING`]**, so
    /// `principals.len()` is a FLOOR and not a total. It has never been true on
    /// any real namespace; it exists so that if it ever is, the number says so
    /// instead of quietly flattening.
    #[serde(default)]
    pub principals_capped: bool,
}

impl FlowDay {
    /// Every refusal, whatever the reason.
    pub fn refused(&self) -> u64 {
        self.refused_admission
            .saturating_add(self.refused_cap)
            .saturating_add(self.refused_policy)
    }

    /// The count for one reason.
    pub fn refused_for(&self, reason: Refusal) -> u64 {
        match reason {
            Refusal::Admission => self.refused_admission,
            Refusal::Cap => self.refused_cap,
            Refusal::Policy => self.refused_policy,
        }
    }

    /// **Distinct principals seen that day** — a floor when
    /// [`FlowDay::principals_capped`].
    pub fn distinct_principals(&self) -> u64 {
        self.principals.len() as u64
    }

    /// **Nothing happened**: no bytes, no sessions, no refusals, nobody seen. A
    /// row like this is never written — a day with no traffic has no row — and
    /// this is what the flush asks before writing one.
    pub fn is_empty(&self) -> bool {
        *self == FlowDay::default()
    }

    /// The last time anything moved in either direction, or `0`.
    pub fn last_seen_unix_ms(&self) -> i64 {
        self.last_push_unix_ms.max(self.last_fetch_unix_ms)
    }

    /// **Add `other` into this day.** Used to fold a series into a window
    /// total, never by the flush — see the module doc for why the flush copies
    /// rather than adds.
    pub fn add(&mut self, other: &FlowDay) {
        self.bytes_in = self.bytes_in.saturating_add(other.bytes_in);
        self.bytes_out = self.bytes_out.saturating_add(other.bytes_out);
        self.pushes = self.pushes.saturating_add(other.pushes);
        self.fetches = self.fetches.saturating_add(other.fetches);
        self.lfs_bytes_in = self.lfs_bytes_in.saturating_add(other.lfs_bytes_in);
        self.lfs_bytes_out = self.lfs_bytes_out.saturating_add(other.lfs_bytes_out);
        self.refused_admission = self
            .refused_admission
            .saturating_add(other.refused_admission);
        self.refused_cap = self.refused_cap.saturating_add(other.refused_cap);
        self.refused_policy = self.refused_policy.saturating_add(other.refused_policy);
        self.auth_failures = self.auth_failures.saturating_add(other.auth_failures);
        self.last_push_unix_ms = self.last_push_unix_ms.max(other.last_push_unix_ms);
        self.last_fetch_unix_ms = self.last_fetch_unix_ms.max(other.last_fetch_unix_ms);
        for who in &other.principals {
            if self.principals.len() >= PRINCIPAL_CEILING {
                self.principals_capped = true;
                break;
            }
            self.principals.insert(who.clone());
        }
        self.principals_capped |= other.principals_capped;
    }
}

/// One namespace's kept days, newest last. Absent days are days with no
/// traffic.
#[derive(Clone, Default, PartialEq, Eq, Debug, serde::Serialize, serde::Deserialize)]
#[serde(deny_unknown_fields)]
pub struct NamespaceFlow {
    /// Keyed by the day number [`day_of`] returns.
    #[serde(default)]
    pub days: BTreeMap<i64, FlowDay>,
}

/// The document as it is stored.
#[derive(Default, Clone, serde::Serialize, serde::Deserialize)]
#[serde(deny_unknown_fields)]
struct FlowDoc {
    /// Keyed by namespace name. One entry per namespace, holding at most
    /// [`RETENTION_DAYS`] rows — a few tens of kilobytes at the top end, which
    /// is one blob and one read rather than a scan over a day-keyed table.
    #[serde(default)]
    flows: ShardedMap<NamespaceFlow>,
}

impl_doc!(FlowDoc { flows });

/// **The kept half**: read and write a namespace's days. Cheap to clone; every
/// clone reads and writes the same document.
#[derive(Clone, Debug)]
pub struct FlowStore {
    doc: DocRef<FlowDoc>,
}

impl FlowStore {
    /// Open the flow document held in `store` (the server's control store). A
    /// store that has never held one reads as empty, which is the truth for
    /// every server that predates this file: no flow was counted, so no flow is
    /// reported — never a zero that would read as *nothing happened*.
    pub fn new(store: Arc<dyn ControlStore>) -> Result<Self> {
        Ok(FlowStore {
            doc: DocRef::new(store, FLOW_REF)?,
        })
    }

    /// The name the document is stored under.
    pub fn ref_name(&self) -> &str {
        self.doc.ref_name()
    }

    /// One namespace's row for one day, or `None` when that day has no traffic.
    pub fn day(&self, namespace: &str, day: i64) -> Result<Option<FlowDay>> {
        Ok(self
            .doc
            .load()?
            .flows
            .get(namespace)?
            .and_then(|flow| flow.days.get(&day).cloned()))
    }

    /// **The series for one namespace, oldest first**, over the inclusive day
    /// range. Days with no traffic are absent rather than zero-filled: the
    /// caller knows the range it asked for, and a row that was never written is
    /// not a row of zeroes that was.
    pub fn series(
        &self,
        namespace: &str,
        from_day: i64,
        to_day: i64,
    ) -> Result<Vec<(i64, FlowDay)>> {
        let Some(flow) = self.doc.load()?.flows.get(namespace)? else {
            return Ok(Vec::new());
        };
        Ok(flow
            .days
            .range(from_day..=to_day)
            .map(|(day, row)| (*day, row.clone()))
            .collect())
    }

    /// Every namespace that has ever had a row. Whole-map, O(entries) — the
    /// fleet view and the flush, never a transfer.
    pub fn namespaces(&self) -> Result<Vec<String>> {
        let mut names: Vec<String> = self.doc.load()?.flows.keys()?;
        names.sort();
        Ok(names)
    }

    /// **Write these rows, replacing whatever each day held, and drop anything
    /// older than [`RETENTION_DAYS`] before `today`.**
    ///
    /// Replace and not add, because the live slot holds the whole of the day it
    /// is writing: see the module doc. An empty row REMOVES the day rather than
    /// storing zeroes — a namespace that had a row and then had a day of
    /// silence must not grow a row saying so.
    ///
    /// One document update for the whole batch, so a flush touching forty
    /// namespaces is one compare-and-swap and not forty.
    pub fn put(&self, rows: &[(String, i64, FlowDay)], today: i64) -> Result<()> {
        if rows.is_empty() {
            return Ok(());
        }
        let oldest = today.saturating_sub(RETENTION_DAYS);
        let rows = rows.to_vec();
        self.doc.update(|doc| {
            for (namespace, day, row) in &rows {
                let mut flow = doc.flows.get(namespace)?.unwrap_or_default();
                if row.is_empty() {
                    flow.days.remove(day);
                } else {
                    flow.days.insert(*day, row.clone());
                }
                flow.days.retain(|day, _| *day > oldest);
                if flow.days.is_empty() {
                    doc.flows.remove(namespace)?;
                } else {
                    doc.flows.insert(namespace.clone(), flow)?;
                }
            }
            Ok(())
        })
    }

    /// Forget one namespace's whole series, reporting whether there was one.
    /// The flow half of deleting a namespace — a tenant that is gone must not
    /// leave a record of who pushed to it.
    pub fn forget(&self, namespace: &str) -> Result<bool> {
        let key = namespace.to_owned();
        self.doc.update(|doc| Ok(doc.flows.remove(&key)?.is_some()))
    }
}

/// **One namespace's flow at a glance**: today, and when it was last seen.
///
/// The shape `Usage.Read` carries. `last_*` deliberately look back over the
/// whole kept window rather than over `today` — see [`FlowLedger::glance`].
#[derive(Clone, Default, PartialEq, Eq, Debug)]
pub struct Glance {
    /// Today's counts. All zero for a day with no traffic, which for a FLOW
    /// meter is a fact and not a gap: a transfer either occurred or it did not.
    pub today: FlowDay,
    /// The last push anywhere in the kept window, or `0` for a namespace that
    /// has never been pushed to inside it.
    pub last_push_unix_ms: i64,
    /// The last fetch anywhere in the kept window, or `0`.
    pub last_fetch_unix_ms: i64,
}

impl Glance {
    /// The last time anything moved in either direction, or `0`.
    pub fn last_seen_unix_ms(&self) -> i64 {
        self.last_push_unix_ms.max(self.last_fetch_unix_ms)
    }
}

/// **The whole box's flow at a glance**, folded over every namespace. See
/// [`FlowLedger::fleet`].
#[derive(Clone, Copy, Default, PartialEq, Eq, Debug)]
pub struct Fleet {
    /// Wire bytes accepted today, across every namespace.
    pub bytes_in_today: u64,
    /// Wire bytes served today, across every namespace.
    pub bytes_out_today: u64,
    /// Transfers turned away today, across every namespace.
    pub refused_today: u64,
    /// How many namespaces this fold looked at.
    pub namespaces: u64,
    /// **How many have neither pushed nor fetched inside the caller's window.**
    /// A tenant that has never been seen at all counts here: it is the quietest
    /// one there is, and rounding it to "not silent" would hide exactly the
    /// account an operator is looking for.
    pub silent: u64,
}

/// The live counters for one namespace's current day. Every field is an atomic
/// so the hot path takes no lock; `roll` is taken only when the day turns and
/// when the flush snapshots.
#[derive(Debug)]
struct Slot {
    day: AtomicI64,
    bytes_in: AtomicU64,
    bytes_out: AtomicU64,
    pushes: AtomicU64,
    fetches: AtomicU64,
    lfs_bytes_in: AtomicU64,
    lfs_bytes_out: AtomicU64,
    refused_admission: AtomicU64,
    refused_cap: AtomicU64,
    refused_policy: AtomicU64,
    auth_failures: AtomicU64,
    last_push_unix_ms: AtomicI64,
    last_fetch_unix_ms: AtomicI64,
    /// The one thing that is not an atomic. A `BTreeSet` insert per transfer —
    /// once per SESSION, never per byte — is not on the byte path.
    principals: Mutex<(BTreeSet<String>, bool)>,
    /// Held while the day turns and while the flush reads, so the two cannot
    /// interleave and a completed day cannot be written twice.
    roll: Mutex<()>,
}

impl Slot {
    fn seeded(day: i64, from: FlowDay) -> Slot {
        Slot {
            day: AtomicI64::new(day),
            bytes_in: AtomicU64::new(from.bytes_in),
            bytes_out: AtomicU64::new(from.bytes_out),
            pushes: AtomicU64::new(from.pushes),
            fetches: AtomicU64::new(from.fetches),
            lfs_bytes_in: AtomicU64::new(from.lfs_bytes_in),
            lfs_bytes_out: AtomicU64::new(from.lfs_bytes_out),
            refused_admission: AtomicU64::new(from.refused_admission),
            refused_cap: AtomicU64::new(from.refused_cap),
            refused_policy: AtomicU64::new(from.refused_policy),
            auth_failures: AtomicU64::new(from.auth_failures),
            last_push_unix_ms: AtomicI64::new(from.last_push_unix_ms),
            last_fetch_unix_ms: AtomicI64::new(from.last_fetch_unix_ms),
            principals: Mutex::new((from.principals, from.principals_capped)),
            roll: Mutex::new(()),
        }
    }

    /// A snapshot, without disturbing the counters.
    fn read(&self) -> FlowDay {
        let (principals, principals_capped) = {
            let held = self.principals.lock().expect("flow principals");
            (held.0.clone(), held.1)
        };
        FlowDay {
            bytes_in: self.bytes_in.load(Ordering::Relaxed),
            bytes_out: self.bytes_out.load(Ordering::Relaxed),
            pushes: self.pushes.load(Ordering::Relaxed),
            fetches: self.fetches.load(Ordering::Relaxed),
            lfs_bytes_in: self.lfs_bytes_in.load(Ordering::Relaxed),
            lfs_bytes_out: self.lfs_bytes_out.load(Ordering::Relaxed),
            refused_admission: self.refused_admission.load(Ordering::Relaxed),
            refused_cap: self.refused_cap.load(Ordering::Relaxed),
            refused_policy: self.refused_policy.load(Ordering::Relaxed),
            auth_failures: self.auth_failures.load(Ordering::Relaxed),
            last_push_unix_ms: self.last_push_unix_ms.load(Ordering::Relaxed),
            last_fetch_unix_ms: self.last_fetch_unix_ms.load(Ordering::Relaxed),
            principals,
            principals_capped,
        }
    }

    /// Zero everything and open a new day.
    fn reset_to(&self, day: i64) {
        for counter in [
            &self.bytes_in,
            &self.bytes_out,
            &self.pushes,
            &self.fetches,
            &self.lfs_bytes_in,
            &self.lfs_bytes_out,
            &self.refused_admission,
            &self.refused_cap,
            &self.refused_policy,
            &self.auth_failures,
        ] {
            counter.store(0, Ordering::Relaxed);
        }
        self.last_push_unix_ms.store(0, Ordering::Relaxed);
        self.last_fetch_unix_ms.store(0, Ordering::Relaxed);
        {
            let mut held = self.principals.lock().expect("flow principals");
            held.0.clear();
            held.1 = false;
        }
        self.day.store(day, Ordering::Release);
    }

    fn saw(&self, principal: &str) {
        if principal.is_empty() {
            return;
        }
        let mut held = self.principals.lock().expect("flow principals");
        if held.0.contains(principal) {
            return;
        }
        if held.0.len() >= PRINCIPAL_CEILING {
            held.1 = true;
            return;
        }
        held.0.insert(principal.to_owned());
    }
}

/// **What one finished transfer moved.** Built by the counting point, applied
/// once.
///
/// One struct rather than eight `record_*` methods, because a transfer is ONE
/// event and its numbers must land on one day together: eight calls could
/// straddle midnight and split a single push across two rows.
#[derive(Clone, Default, PartialEq, Eq, Debug)]
pub struct Transfer {
    /// Wire bytes accepted.
    pub bytes_in: u64,
    /// Wire bytes served.
    pub bytes_out: u64,
    /// A receive-pack session that finished.
    pub pushed: bool,
    /// An upload-pack session that was asked for objects — never the bare
    /// advertisement.
    pub fetched: bool,
    /// LFS object bytes received, as the LFS session counted them.
    pub lfs_bytes_in: u64,
    /// LFS object bytes sent.
    pub lfs_bytes_out: u64,
    /// The principal, by the fingerprint the auth table knows. Empty for a
    /// transfer with no identity, which is counted in the bytes and in nobody's
    /// name.
    pub principal: String,
}

impl Transfer {
    /// Did anything move at all? A transfer that moved no byte in either
    /// direction and neither pushed nor fetched is not written.
    pub fn is_empty(&self) -> bool {
        self.bytes_in == 0
            && self.bytes_out == 0
            && !self.pushed
            && !self.fetched
            && self.lfs_bytes_in == 0
            && self.lfs_bytes_out == 0
    }
}

/// **The live half: the counters every transfer path bumps.**
///
/// Cheap to clone through an `Arc`. A record is: one read lock, one hash
/// lookup, a handful of relaxed `fetch_add`s, and one `BTreeSet` insert per
/// session. Nothing on that path allocates once the namespace has a slot, and a
/// slot is made once per namespace per process.
#[derive(Debug)]
pub struct FlowLedger {
    store: FlowStore,
    /// One slot per namespace touched since boot. The write lock is taken only
    /// to make a slot.
    live: RwLock<HashMap<String, Arc<Slot>>>,
    /// Days that turned over and have not been written yet. The flush drains
    /// this before it snapshots today.
    completed: Mutex<Vec<(String, i64, FlowDay)>>,
    /// **Who is told a namespace's flow moved** (2026-10-06): the server's
    /// usage ledger, so a batched usage read (`UsageBatch.ReadMany` with a
    /// cursor) answers the namespaces whose `*_today` meters changed — and
    /// re-measures their bytes, since a transfer is also what grows a
    /// repository, its LFS store and its exploded read cache. One call per
    /// recorded transfer, refusal or auth failure; never per object.
    observer: RwLock<Option<FlowObserver>>,
}

/// The callback [`FlowLedger::observe`] installs, with a `Debug` that says
/// only that it is there.
#[derive(Clone)]
pub struct FlowObserver(pub Arc<dyn Fn(&str) + Send + Sync>);

impl std::fmt::Debug for FlowObserver {
    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
        f.write_str("FlowObserver(..)")
    }
}

impl FlowLedger {
    /// Open the ledger over a control store.
    pub fn new(store: FlowStore) -> FlowLedger {
        FlowLedger {
            store,
            live: RwLock::new(HashMap::new()),
            completed: Mutex::new(Vec::new()),
            observer: RwLock::new(None),
        }
    }

    /// **Tell `observer` the name of every namespace whose flow moves** — a
    /// recorded transfer, a refusal, an auth failure. Replaces any earlier
    /// observer. Called on the transfer path, so it must be cheap: the
    /// server's usage ledger inserts the name into a set and returns.
    pub fn observe(&self, observer: FlowObserver) {
        *self.observer.write().expect("flow observer") = Some(observer);
    }

    fn tell_observer(&self, namespace: &str) {
        if let Some(observer) = self.observer.read().expect("flow observer").as_ref() {
            (observer.0)(namespace);
        }
    }

    /// The kept half, for the readers that want a series rather than today.
    pub fn store(&self) -> &FlowStore {
        &self.store
    }

    /// The slot for a namespace, made and seeded from the store on first touch.
    ///
    /// **Seeded, not zeroed.** A server that restarts at noon must go on
    /// counting today from where the last flush left it; starting from zero
    /// would make every restart look like a quiet morning.
    fn slot(&self, namespace: &str, day: i64) -> Arc<Slot> {
        if let Some(slot) = self
            .live
            .read()
            .expect("flow slots")
            .get(namespace)
            .map(Arc::clone)
        {
            self.roll_to(namespace, &slot, day);
            return slot;
        }
        // The store read happens outside the write lock: it is I/O, and a
        // second namespace's first transfer must not wait behind it.
        let seed = self
            .store
            .day(namespace, day)
            .unwrap_or_default()
            .unwrap_or_default();
        let mut live = self.live.write().expect("flow slots");
        let slot = Arc::clone(
            live.entry(namespace.to_owned())
                .or_insert_with(|| Arc::new(Slot::seeded(day, seed))),
        );
        drop(live);
        self.roll_to(namespace, &slot, day);
        slot
    }

    /// Turn the day if it has turned, parking the completed row for the flush.
    fn roll_to(&self, namespace: &str, slot: &Slot, day: i64) {
        if slot.day.load(Ordering::Acquire) == day {
            return;
        }
        let _held = slot.roll.lock().expect("flow roll");
        let was = slot.day.load(Ordering::Acquire);
        if was == day {
            return;
        }
        let finished = slot.read();
        if !finished.is_empty() {
            self.completed.lock().expect("flow completed").push((
                namespace.to_owned(),
                was,
                finished,
            ));
        }
        slot.reset_to(day);
    }

    /// **The counting point for a finished transfer.** One call, one day.
    pub fn record(&self, namespace: &str, at_unix_ms: i64, transfer: &Transfer) {
        if namespace.is_empty() || transfer.is_empty() {
            return;
        }
        let slot = self.slot(namespace, day_of(at_unix_ms));
        add(&slot.bytes_in, transfer.bytes_in);
        add(&slot.bytes_out, transfer.bytes_out);
        add(&slot.lfs_bytes_in, transfer.lfs_bytes_in);
        add(&slot.lfs_bytes_out, transfer.lfs_bytes_out);
        if transfer.pushed {
            slot.pushes.fetch_add(1, Ordering::Relaxed);
            slot.last_push_unix_ms.store(at_unix_ms, Ordering::Relaxed);
        }
        if transfer.fetched {
            slot.fetches.fetch_add(1, Ordering::Relaxed);
            slot.last_fetch_unix_ms.store(at_unix_ms, Ordering::Relaxed);
        }
        slot.saw(&transfer.principal);
        self.tell_observer(namespace);
    }

    /// **The counting point for a refusal.** No bytes moved; the tenant met a
    /// door.
    pub fn refuse(&self, namespace: &str, at_unix_ms: i64, reason: Refusal) {
        if namespace.is_empty() {
            return;
        }
        let slot = self.slot(namespace, day_of(at_unix_ms));
        let counter = match reason {
            Refusal::Admission => &slot.refused_admission,
            Refusal::Cap => &slot.refused_cap,
            Refusal::Policy => &slot.refused_policy,
        };
        counter.fetch_add(1, Ordering::Relaxed);
        self.tell_observer(namespace);
    }

    /// **The counting point for a principal with no grant here.** See the
    /// module doc for the half that is not counted.
    pub fn auth_failure(&self, namespace: &str, at_unix_ms: i64) {
        if namespace.is_empty() {
            return;
        }
        let slot = self.slot(namespace, day_of(at_unix_ms));
        slot.auth_failures.fetch_add(1, Ordering::Relaxed);
        self.tell_observer(namespace);
    }

    /// **Today's live figure for one namespace**, whether or not it has been
    /// flushed. Read straight off the atomics — the store is not consulted,
    /// because the slot holds the whole day and the stored row is a copy of it.
    ///
    /// `None` when this process has not seen the namespace today AND the store
    /// holds no row for it: nothing happened, which is a fact and not a gap.
    pub fn today(&self, namespace: &str, now_unix_ms: i64) -> Option<FlowDay> {
        let day = day_of(now_unix_ms);
        if let Some(slot) = self
            .live
            .read()
            .expect("flow slots")
            .get(namespace)
            .map(Arc::clone)
        {
            if slot.day.load(Ordering::Acquire) == day {
                let row = slot.read();
                return (!row.is_empty()).then_some(row);
            }
        }
        self.store.day(namespace, day).ok().flatten()
    }

    /// **What `Usage.Read` puts beside the stock meters**: today's flow, and
    /// when the namespace was last seen over the WHOLE kept window.
    ///
    /// The two halves answer different questions and must not be folded into
    /// one. *How much moved today* is a rate; *when did anything last move* is
    /// what an operator reads to decide a tenant has gone quiet, and a tenant
    /// who has done nothing for six weeks has a today of zeroes and a last-seen
    /// six weeks ago. Reading the second out of the first is exactly the
    /// mistake this pair exists to prevent.
    pub fn glance(&self, namespace: &str, now_unix_ms: i64) -> Glance {
        let today_number = day_of(now_unix_ms);
        let today = self.today(namespace, now_unix_ms).unwrap_or_default();
        let mut glance = Glance {
            last_push_unix_ms: today.last_push_unix_ms,
            last_fetch_unix_ms: today.last_fetch_unix_ms,
            today,
        };
        // Backwards from yesterday, and stop as soon as both are known: a
        // namespace that pushed yesterday costs one row, not four hundred.
        let kept = self
            .store
            .series(namespace, today_number - RETENTION_DAYS, today_number - 1)
            .unwrap_or_default();
        for (_, row) in kept.iter().rev() {
            if glance.last_push_unix_ms == 0 {
                glance.last_push_unix_ms = row.last_push_unix_ms;
            }
            if glance.last_fetch_unix_ms == 0 {
                glance.last_fetch_unix_ms = row.last_fetch_unix_ms;
            }
            if glance.last_push_unix_ms != 0 && glance.last_fetch_unix_ms != 0 {
                break;
            }
        }
        glance
    }

    /// **The whole box's flow, folded** — what moved today across every
    /// namespace, and how many tenants have gone quiet.
    ///
    /// O(namespaces), and it reads the control document. It is for the
    /// dashboard band and for an operator's fleet view, NEVER for a transfer:
    /// the flush tick computes it once a minute and the screen reads the
    /// result, because the dashboard's own two-second tick is documented as
    /// reading a few atomics and nothing else.
    ///
    /// `silent_after_days` is the caller's policy, not this module's. Fourteen
    /// days is what the console uses; a different deployment may reasonably
    /// call a tenant quiet sooner, and a constant here would make that a fork.
    pub fn fleet(&self, now_unix_ms: i64, silent_after_days: i64) -> Fleet {
        let mut names: BTreeSet<String> = self
            .store
            .namespaces()
            .unwrap_or_default()
            .into_iter()
            .collect();
        names.extend(self.live.read().expect("flow slots").keys().cloned());
        let quiet_before = now_unix_ms.saturating_sub(silent_after_days.saturating_mul(DAY_MS));
        let mut fleet = Fleet {
            namespaces: names.len() as u64,
            ..Fleet::default()
        };
        for name in &names {
            let glance = self.glance(name, now_unix_ms);
            fleet.bytes_in_today = fleet.bytes_in_today.saturating_add(glance.today.bytes_in);
            fleet.bytes_out_today = fleet.bytes_out_today.saturating_add(glance.today.bytes_out);
            fleet.refused_today = fleet.refused_today.saturating_add(glance.today.refused());
            // A namespace that has NEVER been seen inside the kept window is
            // silent too — `last_seen_unix_ms()` is 0 there, which is before
            // any cutoff. That is the right answer: a tenant nobody has ever
            // pushed to is the quietest one there is.
            if glance.last_seen_unix_ms() < quiet_before {
                fleet.silent = fleet.silent.saturating_add(1);
            }
        }
        fleet
    }

    /// **Copy every live slot into the store, and drop what has aged out.**
    ///
    /// The flush. Runs on a tick; touches redb once for the whole batch. It
    /// COPIES rather than adds, so running it twice in a row writes the same
    /// rows twice rather than doubling them — which is what makes it safe to
    /// run it again on shutdown, and safe for a restart to seed from it.
    ///
    /// Returns how many rows it wrote, for the caller's log line.
    pub fn flush(&self, now_unix_ms: i64) -> Result<usize> {
        let today = day_of(now_unix_ms);
        let mut rows: Vec<(String, i64, FlowDay)> =
            std::mem::take(&mut *self.completed.lock().expect("flow completed"));
        let slots: Vec<(String, Arc<Slot>)> = self
            .live
            .read()
            .expect("flow slots")
            .iter()
            .map(|(name, slot)| (name.clone(), Arc::clone(slot)))
            .collect();
        for (namespace, slot) in slots {
            // Roll first: a slot nobody has touched since yesterday still holds
            // yesterday, and the tick is what notices on an idle server.
            self.roll_to(&namespace, &slot, today);
            let _held = slot.roll.lock().expect("flow roll");
            let row = slot.read();
            if !row.is_empty() {
                rows.push((namespace, slot.day.load(Ordering::Acquire), row));
            }
        }
        // A roll during the loop above parks rows in `completed`; take them too
        // rather than leaving them for the next tick.
        rows.extend(std::mem::take(
            &mut *self.completed.lock().expect("flow completed"),
        ));
        let written = rows.len();
        if written > 0 {
            if let Err(err) = self.store.put(&rows, today) {
                // A refused write (the transaction log released while the data set is
                // grown, a full disk) must not LOSE the completed days taken out of
                // `completed` above: a live slot is still in memory, a rolled day is
                // only here. Parked again, the next flush writes them. A live-slot row
                // parked beside them is harmless: `put` copies, it does not add.
                self.completed.lock().expect("flow completed").extend(rows);
                return Err(err);
            }
        }
        Ok(written)
    }

    /// **Bytes moved — in plus out — per UTC day, over the last `days` days,
    /// today included**: `(day, bytes)` for each day that moved anything,
    /// oldest first. The durable half of admission's usage: `edda_admit`'s
    /// table SEEDS a namespace from this once, on first sight, and is the
    /// cache from then on — so this is read once per namespace per process,
    /// never per transfer (it takes the ledger's locks and reads the store).
    ///
    /// Wire bytes only. LFS bytes are already inside them (see
    /// [`FlowDay::lfs_bytes_in`]), so adding the LFS pair would count one
    /// transfer twice.
    ///
    /// * **today** is the live slot, read off its atomics;
    /// * **the closed days** are the store's rows, plus any closed day still
    ///   parked for the flush — a parked row is the FINAL figure for its day
    ///   and replaces the store's last-tick copy of it, never adds to it.
    ///
    /// A slot still holding yesterday is rolled first, so a read just after
    /// midnight does not take yesterday's live bytes for today's.
    pub fn moved(&self, namespace: &str, now_unix_ms: i64, days: u32) -> Vec<(i64, u64)> {
        let today = day_of(now_unix_ms);
        let live_slot = self
            .live
            .read()
            .expect("flow slots")
            .get(namespace)
            .map(Arc::clone);
        if let Some(slot) = live_slot {
            self.roll_to(namespace, &slot, today);
        }
        let from = today - i64::from(days.max(1)) + 1;
        let mut rows: BTreeMap<i64, u64> = if from < today {
            self.store
                .series(namespace, from, today - 1)
                .unwrap_or_default()
                .into_iter()
                .map(|(day, row)| (day, row.bytes_in.saturating_add(row.bytes_out)))
                .collect()
        } else {
            BTreeMap::new()
        };
        rows.extend(
            self.completed
                .lock()
                .expect("flow completed")
                .iter()
                .filter(|(name, day, _)| name == namespace && *day >= from && *day < today)
                .map(|(_, day, row)| (*day, row.bytes_in.saturating_add(row.bytes_out))),
        );
        if let Some(t) = self.today(namespace, now_unix_ms) {
            rows.insert(today, t.bytes_in.saturating_add(t.bytes_out));
        }
        rows.into_iter().filter(|(_, b)| *b > 0).collect()
    }

    /// [`FlowLedger::moved`], summed: bytes in + out over the window.
    pub fn window_bytes(&self, namespace: &str, now_unix_ms: i64, days: u32) -> u64 {
        self.moved(namespace, now_unix_ms, days)
            .iter()
            .fold(0u64, |acc, (_, b)| acc.saturating_add(*b))
    }

    /// Drop a namespace's live slot and its whole kept series. The flow half of
    /// deleting a namespace.
    pub fn forget(&self, namespace: &str) -> Result<bool> {
        self.live.write().expect("flow slots").remove(namespace);
        self.completed
            .lock()
            .expect("flow completed")
            .retain(|(name, _, _)| name != namespace);
        self.store.forget(namespace)
    }
}

/// `fetch_add` that skips the bus traffic for the common zero.
fn add(counter: &AtomicU64, by: u64) {
    if by > 0 {
        counter.fetch_add(by, Ordering::Relaxed);
    }
}

#[cfg(test)]
mod tests {
    use super::*;
    use crate::control::RedbControlStore;

    fn ledger() -> FlowLedger {
        let store = Arc::new(RedbControlStore::in_memory().unwrap()) as Arc<dyn ControlStore>;
        FlowLedger::new(FlowStore::new(store).unwrap())
    }

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

    /// ★ **The admission window is today plus the closed days before it**,
    /// in + out, from the live slot, the parked rows and the store — each day
    /// counted once, whichever of those three it sits in.
    #[test]
    fn the_window_sums_thirty_days_counting_each_day_once() {
        let flow = ledger();
        let day0 = 20_000 * DAY_MS + 3_600_000;
        // 31 days ago: outside a 30-day window.
        flow.record("team", day0, &push(1_000_000, "a"));
        // 29 days ago: the oldest day inside it.
        flow.record("team", day0 + DAY_MS, &push(1000, "a"));
        flow.flush(day0 + DAY_MS).unwrap();
        // Yesterday: rolled but NOT flushed — parked, and the store holds nothing of it.
        flow.record("team", day0 + 29 * DAY_MS, &fetch(200, "a"));
        flow.record("team", day0 + 30 * DAY_MS, &push(30, "a"));
        flow.record("team", day0 + 30 * DAY_MS, &fetch(7, "b"));
        let now = day0 + 30 * DAY_MS;
        assert_eq!(flow.window_bytes("team", now, 30), 1000 + 200 + 30 + 7);
        let days = flow.moved("team", now, 30);
        let today = day_of(now);
        assert_eq!(
            days,
            vec![(today - 29, 1000), (today - 1, 200), (today, 37)],
            "oldest first, per day"
        );
        assert_eq!(
            flow.window_bytes("team", now, 1),
            37,
            "a one-day window is today"
        );
        assert_eq!(flow.window_bytes("other", now, 30), 0);
        // After the flush the parked row is in the store: the same total, not double.
        flow.flush(now).unwrap();
        assert_eq!(flow.window_bytes("team", now, 30), 1237);
        // Today is live: a new transfer today shows at once.
        flow.record("team", now + 1, &push(3, "a"));
        assert_eq!(flow.window_bytes("team", now + 1, 30), 1240);
        // The next day the oldest day falls out.
        assert_eq!(flow.window_bytes("team", now + DAY_MS, 30), 200 + 40);
    }

    /// A slot still holding yesterday is rolled before it is read: the first
    /// read after midnight does not take yesterday's live bytes for today's,
    /// nor lose them.
    #[test]
    fn the_window_rolls_a_stale_slot_before_reading_it() {
        let flow = ledger();
        let day0 = 20_000 * DAY_MS;
        flow.flush(day0).unwrap();
        flow.record("team", day0 + 10, &push(500, "a"));
        flow.flush(day0 + 20).unwrap(); // the store holds a copy of day0 = 500
        flow.record("team", day0 + 30, &push(5, "a")); // live day0 = 505, store 500
        let next = day0 + DAY_MS + 1;
        assert_eq!(flow.window_bytes("team", next, 30), 505);
        assert_eq!(
            flow.window_bytes("team", next, 1),
            0,
            "nothing moved today yet"
        );
    }

    /// ★ **A push charges the namespace it was pushed to, and nobody else.**
    ///
    /// The first thing that has to be true of a per-tenant meter, and the one
    /// that a global counter — which is all gunnar had — cannot be.
    #[test]
    fn a_push_lands_on_its_own_namespace_and_no_other() {
        let flow = ledger();
        let now = 1_757_000_000_000;
        flow.record("team", now, &push(4096, "SHA256:aaa"));

        let mine = flow.today("team", now).expect("the push was counted");
        assert_eq!(mine.bytes_in, 4096);
        assert_eq!(mine.pushes, 1);
        assert_eq!(mine.bytes_out, 0, "a push serves no pack");
        assert_eq!(mine.last_push_unix_ms, now);
        assert_eq!(mine.distinct_principals(), 1);

        assert_eq!(
            flow.today("other", now),
            None,
            "a namespace with no traffic has no row, not a row of zeroes"
        );
    }

    /// A fetch is bytes OUT and a fetch count; a bare advertisement is neither.
    #[test]
    fn a_fetch_is_bytes_out_and_an_advertisement_is_not_a_fetch() {
        let flow = ledger();
        let now = 1_757_000_000_000;
        flow.record("team", now, &fetch(9000, "SHA256:bbb"));
        // The advertisement: bytes out, no session.
        flow.record(
            "team",
            now,
            &Transfer {
                bytes_out: 120,
                ..Transfer::default()
            },
        );

        let row = flow.today("team", now).unwrap();
        assert_eq!(row.bytes_out, 9120, "both wrote bytes");
        assert_eq!(row.fetches, 1, "only one of them asked for objects");
        assert_eq!(row.last_fetch_unix_ms, now);
    }

    /// ★ **A refusal is counted, by reason, and moves no bytes.**
    #[test]
    fn a_refusal_is_counted_by_reason_and_leaves_the_bytes_alone() {
        let flow = ledger();
        let now = 1_757_000_000_000;
        flow.record("team", now, &push(4096, "SHA256:aaa"));
        flow.refuse("team", now, Refusal::Cap);
        flow.refuse("team", now, Refusal::Cap);
        flow.refuse("team", now, Refusal::Admission);

        let row = flow.today("team", now).unwrap();
        assert_eq!(row.bytes_in, 4096, "a refused push moved nothing");
        assert_eq!(row.pushes, 1);
        assert_eq!(row.refused(), 3);
        assert_eq!(row.refused_for(Refusal::Cap), 2);
        assert_eq!(row.refused_for(Refusal::Admission), 1);
        assert_eq!(row.refused_for(Refusal::Policy), 0);
    }

    /// ★ **A day rollover splits the counts**, and yesterday keeps its own.
    #[test]
    fn the_day_boundary_splits_the_counts() {
        let flow = ledger();
        let day_one = day_of(1_757_000_000_000);
        let late = midnight_of(day_one) + DAY_MS - 1;
        let early = midnight_of(day_one + 1) + 1;

        flow.record("team", late, &push(1000, "SHA256:aaa"));
        flow.record("team", early, &push(7, "SHA256:aaa"));

        let today = flow.today("team", early).unwrap();
        assert_eq!(today.bytes_in, 7, "the new day starts from nothing");
        assert_eq!(today.pushes, 1);

        flow.flush(early).unwrap();
        let yesterday = flow
            .store()
            .day("team", day_one)
            .unwrap()
            .expect("yesterday was written");
        assert_eq!(yesterday.bytes_in, 1000);
        assert_eq!(yesterday.pushes, 1);
    }

    /// ★ **A restart keeps yesterday and drops nothing of today.**
    ///
    /// The whole reason the flush copies rather than adds, and the reason a new
    /// slot is seeded from the store: a second ledger over the same control
    /// store must go on counting today from where the first left it.
    #[test]
    fn a_restart_keeps_yesterday_and_resumes_today() {
        let store = Arc::new(RedbControlStore::in_memory().unwrap()) as Arc<dyn ControlStore>;
        let day_one = day_of(1_757_000_000_000);
        let yesterday_at = midnight_of(day_one) + 3_600_000;
        let today_at = midnight_of(day_one + 1) + 3_600_000;
        let later_today = today_at + 60_000;

        {
            let flow = FlowLedger::new(FlowStore::new(Arc::clone(&store)).unwrap());
            flow.record("team", yesterday_at, &push(1000, "SHA256:aaa"));
            flow.flush(yesterday_at).unwrap();
            flow.record("team", today_at, &push(500, "SHA256:aaa"));
            flow.flush(today_at).unwrap();
        }
        // The server restarts.
        let flow = FlowLedger::new(FlowStore::new(Arc::clone(&store)).unwrap());
        assert_eq!(
            flow.today("team", today_at).unwrap().bytes_in,
            500,
            "today survived the restart"
        );
        flow.record("team", later_today, &push(1, "SHA256:aaa"));
        assert_eq!(
            flow.today("team", later_today).unwrap().bytes_in,
            501,
            "counting resumed from the stored figure, not from zero"
        );
        assert_eq!(flow.today("team", later_today).unwrap().pushes, 2);
        assert_eq!(
            flow.store().day("team", day_one).unwrap().unwrap().bytes_in,
            1000,
            "yesterday is untouched"
        );
    }

    /// **Flushing twice writes the same numbers twice, not double them.** The
    /// property that makes the flush safe to run on a tick AND on shutdown.
    #[test]
    fn a_second_flush_is_not_a_second_count() {
        let flow = ledger();
        let now = 1_757_000_000_000;
        flow.record("team", now, &push(4096, "SHA256:aaa"));
        flow.flush(now).unwrap();
        flow.flush(now).unwrap();
        flow.flush(now).unwrap();
        assert_eq!(
            flow.store()
                .day("team", day_of(now))
                .unwrap()
                .unwrap()
                .bytes_in,
            4096
        );
    }

    /// ★ **The series comes back oldest first**, and days with no traffic are
    /// absent rather than zero-filled.
    #[test]
    fn the_series_is_oldest_first_and_silent_days_are_absent() {
        let flow = ledger();
        let first = day_of(1_757_000_000_000);
        for (offset, bytes) in [(0i64, 100u64), (1, 200), (3, 400)] {
            let at = midnight_of(first + offset) + 1000;
            flow.record("team", at, &push(bytes, "SHA256:aaa"));
            flow.flush(at).unwrap();
        }
        let series = flow.store().series("team", first, first + 3).unwrap();
        assert_eq!(
            series.iter().map(|(day, _)| *day).collect::<Vec<_>>(),
            vec![first, first + 1, first + 3],
            "oldest first, and the silent day is not a row"
        );
        assert_eq!(
            series.iter().map(|(_, r)| r.bytes_in).collect::<Vec<_>>(),
            vec![100, 200, 400]
        );
        // A window that starts after the first day starts there.
        let narrow = flow.store().series("team", first + 1, first + 3).unwrap();
        assert_eq!(narrow.len(), 2);
    }

    /// **Retention is 400 days, and the writer prunes.** A row older than the
    /// window is gone the next time its namespace is written.
    #[test]
    fn nothing_older_than_the_window_survives_a_write() {
        let flow = ledger();
        let ancient = day_of(1_757_000_000_000);
        let at = midnight_of(ancient) + 1000;
        flow.record("team", at, &push(100, "SHA256:aaa"));
        flow.flush(at).unwrap();
        assert!(flow.store().day("team", ancient).unwrap().is_some());

        let much_later = midnight_of(ancient + RETENTION_DAYS + 1) + 1000;
        flow.record("team", much_later, &push(1, "SHA256:aaa"));
        flow.flush(much_later).unwrap();
        assert_eq!(flow.store().day("team", ancient).unwrap(), None, "aged out");
        assert!(flow
            .store()
            .day("team", day_of(much_later))
            .unwrap()
            .is_some());
    }

    /// Distinct principals are exact and de-duplicated; the ceiling says so
    /// when it is hit rather than flattening quietly.
    #[test]
    fn principals_are_exact_and_the_ceiling_announces_itself() {
        let flow = ledger();
        let now = 1_757_000_000_000;
        for _ in 0..5 {
            flow.record("team", now, &push(1, "SHA256:aaa"));
            flow.record("team", now, &fetch(1, "SHA256:bbb"));
        }
        let row = flow.today("team", now).unwrap();
        assert_eq!(row.distinct_principals(), 2, "ten transfers, two people");
        assert!(!row.principals_capped);

        let mut wide = FlowDay::default();
        for n in 0..PRINCIPAL_CEILING + 10 {
            wide.principals.insert(format!("SHA256:{n}"));
        }
        let mut folded = FlowDay::default();
        folded.add(&wide);
        assert!(
            folded.principals_capped,
            "a count that stopped growing must say so"
        );
        assert_eq!(folded.distinct_principals(), PRINCIPAL_CEILING as u64);
    }

    /// Forgetting a namespace forgets its whole series and its live slot.
    #[test]
    fn forgetting_a_namespace_forgets_its_flow() {
        let flow = ledger();
        let now = 1_757_000_000_000;
        flow.record("team", now, &push(4096, "SHA256:aaa"));
        flow.record("other", now, &push(1, "SHA256:aaa"));
        flow.flush(now).unwrap();
        assert!(flow.forget("team").unwrap());
        assert_eq!(flow.today("team", now), None);
        assert!(
            flow.today("other", now).is_some(),
            "the other tenant is untouched"
        );
    }

    /// A transfer that moved nothing is not a row, and an empty namespace name
    /// is refused rather than charged to `""`.
    #[test]
    fn nothing_is_never_written_as_something() {
        let flow = ledger();
        let now = 1_757_000_000_000;
        flow.record("team", now, &Transfer::default());
        flow.record("", now, &push(4096, "SHA256:aaa"));
        assert_eq!(flow.today("team", now), None);
        assert_eq!(flow.today("", now), None);
        assert_eq!(flow.flush(now).unwrap(), 0, "nothing to write");
    }

    /// The date renderer, against the two dates in this file's fixtures and the
    /// epoch itself.
    #[test]
    fn the_date_renderer_is_utc_and_floors_before_the_epoch() {
        assert_eq!(iso8601_utc(0), "1970-01-01T00:00:00Z");
        assert_eq!(iso8601_utc(1_757_000_000_000), "2025-09-04T15:33:20Z");
        assert_eq!(iso8601_utc(1_789_041_600_000), "2026-09-10T12:00:00Z");
        assert_eq!(
            iso8601_utc(-1),
            "1969-12-31T23:59:59Z",
            "one millisecond before the epoch"
        );
        // A leap day, which is where a table-free formula goes wrong if it can.
        assert_eq!(iso8601_utc(1_709_164_800_000), "2024-02-29T00:00:00Z");
    }

    /// The day arithmetic, including before the epoch, where truncation toward
    /// zero would put two different days in one bucket.
    #[test]
    fn the_day_arithmetic_floors() {
        assert_eq!(day_of(0), 0);
        assert_eq!(day_of(DAY_MS - 1), 0);
        assert_eq!(day_of(DAY_MS), 1);
        assert_eq!(day_of(-1), -1, "the last millisecond of 1969 is not day 0");
        assert_eq!(midnight_of(day_of(1_757_000_000_000)), 1_756_944_000_000);
    }
}