crates/gunnar-core/src/flow.rs
//! **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);
}
}