crates/monetize-product/src/flow.rs
//! **Reading a DAILY COUNTER out of a series of readings** — the arithmetic
//! behind the console's SILENT and FRICTION lists.
//!
//! Pure. No store, no clock, no product — [`crate::trend`]'s shape, and for the
//! same reason: the fold has to be assertable without a wire.
//!
//! # The problem this exists to solve
//!
//! gunnar's flow meters are `*_today` counters. `refused_today` is *how many
//! transfers this namespace has had refused since midnight UTC*: it starts at 0
//! every day, only ever grows within a day, and drops back at the boundary. The
//! poller reads it every fifteen minutes and `UsageStore` keeps every reading
//! for 400 days.
//!
//! So the question *how many refusals last week* has an obvious wrong answer and
//! a correct one:
//!
//! * **Wrong: sum the readings.** A tenant refused once at 09:00 has
//! `refused_today = 1` in every one of the day's remaining sixty readings, and
//! summing them reports sixty refusals. The busier the poller, the bigger the
//! lie — a monitoring interval would be silently multiplying a customer's
//! friction.
//! * **Right: the MAXIMUM reading of each UTC day is that day's total**, because
//! the counter is monotonic within the day; then add the days. That is
//! [`daily_totals`].
//!
//! # Why the maximum and not the last reading of the day
//!
//! The last reading of a day is the natural choice and it is wrong at exactly
//! one moment: the poll that lands a few milliseconds after midnight, whose
//! reading belongs to the NEW day but whose `measured_at` may still be
//! attributed to the old one by a clock that disagrees by a hair. That reading
//! is a small number where the day's real total is large, and taking it would
//! erase the day. The maximum cannot be wrong in that direction — a counter
//! that only grows within a day has its total at its peak — and its failure
//! mode if a reading is ever attributed to the wrong day is to keep the larger
//! of two true numbers rather than to invent one.
//!
//! # What is NOT here
//!
//! This does not fold `last_push_unix_ms`. That meter is not a daily counter —
//! it is an instant, and gunnar already looks back over its whole kept window to
//! produce it — so the answer to *when was this tenant last seen* is the latest
//! reading's value, read straight off, and folding it would be arithmetic on a
//! timestamp. [`quiet_for_ms`] is the whole of what the SILENT list needs.
use std::collections::BTreeMap;
use crate::Usage;
/// 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: u64 = 86_400_000;
/// The UTC day a reading falls in, as whole days since the epoch.
pub fn day_of(unix_ms: u64) -> u64 {
unix_ms / DAY_MS
}
/// **Per UTC day, the total of one `*_today` counter** — the day's maximum
/// reading. See the module doc for why the maximum.
///
/// Readings that do not carry the meter are skipped, not counted as zero: a
/// product that stopped reporting a meter and a product reporting zero are
/// different facts, and only the second is a day with no refusals.
///
/// A day with no reading at all is ABSENT from the map. The caller knows which
/// days it asked about, and a zero for a day nobody measured is the one number
/// that would make an unreachable product look like a quiet week.
pub fn daily_totals(readings: &[Usage], meter: &str) -> BTreeMap<u64, u64> {
let mut days: BTreeMap<u64, u64> = BTreeMap::new();
for reading in readings {
let Some(value) = reading.meters.get(meter).copied() else {
continue;
};
let day = days.entry(day_of(reading.measured_at_unix_ms)).or_insert(0);
*day = (*day).max(value);
}
days
}
/// **The total of one `*_today` counter over the readings whose day falls in
/// `[since_unix_ms, until_unix_ms]`.**
///
/// The window is resolved to whole UTC days at both ends, because the counter
/// is a per-day quantity and half a day of one is not a number that exists.
pub fn window_total(
readings: &[Usage],
meter: &str,
since_unix_ms: u64,
until_unix_ms: u64,
) -> u64 {
let (from, to) = (day_of(since_unix_ms), day_of(until_unix_ms));
daily_totals(readings, meter)
.range(from..=to)
.map(|(_, total)| *total)
.fold(0u64, |acc, n| acc.saturating_add(n))
}
/// **How long a tenant has been quiet**, or `None` when it has not been.
///
/// `last_seen_unix_ms` is the later of the two `last_*_unix_ms` meters, which
/// gunnar computes over its WHOLE kept window rather than over today — so a
/// tenant with a today of zeroes and a push six weeks ago is six weeks quiet
/// and not silent forever.
///
/// **`0` means NEVER, not 1970.** proto3 cannot tell an unset timestamp from a
/// zero one, and gunnar spells "this namespace has never been touched inside
/// the kept window" as 0. Rendering that as an epoch date would put 1970 on a
/// console beside real dates, which reads as a very old customer rather than as
/// one nobody has ever used — so `None` for `last_seen == 0` is refused here
/// and [`Quiet::Never`] says it out loud.
pub fn quiet_for(last_seen_unix_ms: Option<u64>, now_unix_ms: u64) -> Quiet {
// **NOTHING WAS MEASURED.** Not a silence — an absence of the meters that
// would have measured one. See [`last_seen`] and [`Quiet::NoData`].
let Some(last_seen_unix_ms) = last_seen_unix_ms else {
return Quiet::NoData;
};
if last_seen_unix_ms == 0 {
return Quiet::Never;
}
match now_unix_ms.checked_sub(last_seen_unix_ms) {
// A reading from the future is a clock disagreement, not a visit that
// has not happened yet. Treated as "just now", which is the closest
// true statement and never puts a working tenant on the SILENT list.
None => Quiet::For(0),
Some(ms) => Quiet::For(ms),
}
}
/// [`quiet_for`], as bare milliseconds — the ordering key the SILENT list sorts
/// by, descending.
///
/// `Never` is `u64::MAX`, so the tenant nobody has ever used sorts above the one
/// last seen a year ago. [`Quiet::NoData`] is `0` and therefore sorts LAST,
/// which is defensive only: it is not a member of that list at all
/// ([`Quiet::at_least_days`] is false for it), and if a caller ever puts one
/// there anyway it must not displace a real finding from the top of the page.
pub fn quiet_for_ms(last_seen_unix_ms: Option<u64>, now_unix_ms: u64) -> u64 {
match quiet_for(last_seen_unix_ms, now_unix_ms) {
Quiet::NoData => 0,
Quiet::Never => u64::MAX,
Quiet::For(ms) => ms,
}
}
/// **How long since a tenant was last seen** — three arms, because there are
/// three facts and only two of them are about the customer.
#[derive(Clone, Copy, PartialEq, Eq, PartialOrd, Ord, Debug)]
pub enum Quiet {
/// **The product did not report the flow meters at all**, so nothing is
/// known about whether this tenant is being used.
///
/// Ordered first so it can never outrank a measured silence.
///
/// This is NOT [`Quiet::Never`] and the difference is the whole point:
/// `Never` is a MEASUREMENT — the product counts flow, and it counted none
/// — while `NoData` is a GAP, in a plugin or a poller or an older product
/// build. Rendering a gap as `never seen` publishes a finding about a
/// customer that nobody measured, which is exactly the sentence an operator
/// would act on by ringing them up.
NoData,
/// Nothing has ever moved through this tenant inside the product's kept
/// window. **Not the same as "a very long time ago"**, and a console must
/// not render it as a date.
Never,
/// Milliseconds since the last push or fetch.
For(u64),
}
impl Quiet {
/// Whole days, or `None` for the two arms that have no number of days.
///
/// **`None` is ambiguous on purpose and must not be the only thing a
/// renderer gets**: both [`Quiet::NoData`] and [`Quiet::Never`] answer
/// `None` here, and they are different facts. Pass the `Quiet` itself to
/// anything that puts words on a screen; this is for arithmetic.
pub fn days(self) -> Option<u64> {
match self {
Quiet::NoData | Quiet::Never => None,
Quiet::For(ms) => Some(ms / DAY_MS),
}
}
/// Has this tenant been quiet for at least `days`?
///
/// `Never` always has. **`NoData` never has** — the SILENT list is a list of
/// MEASURED silences, and a tenant whose product does not count flow has not
/// been shown to be quiet, only to be unmeasured.
pub fn at_least_days(self, days: u64) -> bool {
match self {
Quiet::NoData => false,
Quiet::Never => true,
Quiet::For(ms) => ms >= days.saturating_mul(DAY_MS),
}
}
/// Is this an absence of measurement rather than a measurement?
pub fn is_no_data(self) -> bool {
matches!(self, Quiet::NoData)
}
}
/// **The later of a tenant's two last-seen timestamps**, from a usage reading —
/// or `None` when the reading carries NEITHER of them.
///
/// One writer, because "last seen" is a word two lists and one panel use and a
/// second `max` somewhere else is how they would come to disagree about whether
/// a fetch counts as being seen. It does: a customer who clones every morning
/// and never pushes is using the product.
///
/// # Why this is an `Option` and not a `u64`
///
/// It returned `u64` until 2026-09-10, with `.unwrap_or(0)` on each meter —
/// which folded a meter that was **absent** into a meter that **read zero**,
/// and those are the two facts this module exists to keep apart. [`daily_totals`]
/// three functions up already refuses exactly that fold, in those words: *"a
/// product that stopped reporting a meter and a product reporting zero are
/// different facts"*. This function contradicted it.
///
/// The consequence was on the screen. A tenant whose product does not count
/// flow at all — an older gunnar, the `Admin.StorageInfo` fallback path, or one
/// the poller has never read — carries neither meter, folded to `0`, and came
/// out of [`quiet_for`] as `Quiet::Never`: **`never seen`, at the top of the
/// fleet page's SILENT list**, sorted above every real finding by
/// [`quiet_for_ms`]'s `u64::MAX`. That is a claim about a customer manufactured
/// from a gap in a plugin, and an operator acts on it by ringing them up.
///
/// The tenant page had already settled this the other way — it draws *"this
/// product does not report flow"* when neither key is present, and its comment
/// says why: *"`never seen` about a product that does not count flow would be a
/// claim about a customer made from a gap in a plugin"*. Two views of one fact
/// disagreed; this makes the type carry the distinction so they cannot.
///
/// `Some(0)` is still meaningful and still means NEVER: the product reported the
/// meters and they read zero. Only `None` is *nobody measured*.
pub fn last_seen(usage: &Usage) -> Option<u64> {
let at = |meter: &str| usage.meters.get(meter).copied();
match (at("last_push_unix_ms"), at("last_fetch_unix_ms")) {
// Neither meter reported: this product does not count flow.
(None, None) => None,
// At least one reported. A missing twin is `0` for the `max` only —
// a product may report a push meter and no fetch meter.
(push, fetch) => Some(push.unwrap_or(0).max(fetch.unwrap_or(0))),
}
}
#[cfg(test)]
mod tests {
use super::*;
fn reading(at: u64, meters: &[(&str, u64)]) -> Usage {
Usage {
meters: meters.iter().map(|(k, v)| ((*k).to_string(), *v)).collect(),
measured_at_unix_ms: at,
}
}
const NOON: u64 = 1_789_041_600_000;
/// ★ **Sixty readings of one refusal are one refusal, not sixty.**
///
/// The defect this module exists to prevent, and the one a naive `sum` has:
/// a `*_today` counter is repeated in every reading of the day it belongs
/// to, so summing readings reports a number proportional to the POLL RATE
/// rather than to what happened.
#[test]
fn a_daily_counter_is_read_once_per_day_however_often_it_is_polled() {
let midnight = day_of(NOON) * DAY_MS;
// One refusal at 09:00, then the counter sits at 1 for the rest of the
// day across sixty more polls.
let mut readings = vec![reading(midnight + 8 * 3_600_000, &[("refused_today", 0)])];
for tick in 0..60u64 {
readings.push(reading(
midnight + 9 * 3_600_000 + tick * 900_000,
&[("refused_today", 1)],
));
}
let totals = daily_totals(&readings, "refused_today");
assert_eq!(totals.len(), 1, "one day: {totals:?}");
assert_eq!(
totals[&day_of(NOON)],
1,
"summing the readings would say 60; the counter is per day, not per poll"
);
}
/// The maximum, not the last reading — see the module doc for the midnight
/// poll that makes the last reading wrong.
#[test]
fn the_days_total_is_its_peak_and_not_its_last_reading() {
let midnight = day_of(NOON) * DAY_MS;
let readings = vec![
reading(midnight + 3_600_000, &[("refused_today", 4)]),
reading(midnight + 7_200_000, &[("refused_today", 9)]),
// A reading attributed to this day whose counter has already
// rolled: taking the last would erase eight of the nine.
reading(midnight + DAY_MS - 1, &[("refused_today", 1)]),
];
assert_eq!(daily_totals(&readings, "refused_today")[&day_of(NOON)], 9);
}
/// ★ **A window adds whole days, and a day with no reading is absent
/// rather than zero.**
#[test]
fn a_window_adds_the_days_it_names_and_invents_none() {
let day = |n: u64| day_of(NOON) * DAY_MS + n * DAY_MS + 3_600_000;
let readings = vec![
reading(day(0), &[("refused_today", 2)]),
reading(day(1), &[("refused_today", 5)]),
// day 2: the product was unreachable and the poller stored nothing.
reading(day(3), &[("refused_today", 1)]),
];
let totals = daily_totals(&readings, "refused_today");
assert_eq!(totals.len(), 3, "three days were measured: {totals:?}");
assert!(
!totals.contains_key(&(day_of(NOON) + 2)),
"a day nobody measured must not appear as a day with no refusals"
);
assert_eq!(window_total(&readings, "refused_today", day(0), day(3)), 8);
assert_eq!(
window_total(&readings, "refused_today", day(1), day(3)),
6,
"the window is honoured"
);
assert_eq!(
window_total(&readings, "refused_today", day(4), day(9)),
0,
"a window with nothing in it"
);
}
/// A meter the readings do not carry is skipped, never counted as zero: a
/// product that stopped reporting and a product reporting nothing are
/// different facts.
#[test]
fn a_meter_that_is_not_reported_is_not_a_zero() {
let readings = vec![reading(NOON, &[("bytes_in_today", 4_096)])];
assert!(
daily_totals(&readings, "refused_today").is_empty(),
"a meter nobody reported must not produce a day of zero refusals"
);
assert_eq!(window_total(&readings, "refused_today", 0, NOON), 0);
}
/// ★ **NEVER is its own answer, and it sorts above every date.**
///
/// `0` on the wire means "nothing has ever moved through this tenant". Read
/// as a timestamp it is 1970, which on a console beside real dates reads as
/// a very old customer rather than as one nobody has ever used.
#[test]
fn never_is_not_nineteen_seventy() {
assert_eq!(quiet_for(Some(0), NOON), Quiet::Never);
assert_eq!(
quiet_for(Some(0), NOON).days(),
None,
"never has no number of days"
);
assert!(quiet_for(Some(0), NOON).at_least_days(14));
assert_eq!(
quiet_for_ms(Some(0), NOON),
u64::MAX,
"the tenant nobody has ever used must sort above one last seen a year ago"
);
let three_days = NOON - 3 * DAY_MS;
assert_eq!(quiet_for(Some(three_days), NOON), Quiet::For(3 * DAY_MS));
assert_eq!(quiet_for(Some(three_days), NOON).days(), Some(3));
assert!(!quiet_for(Some(three_days), NOON).at_least_days(14));
assert!(
quiet_for(Some(NOON - 14 * DAY_MS), NOON).at_least_days(14),
"exactly at the threshold is quiet"
);
}
/// ★ **A tenant nobody MEASURED is NO DATA, and a tenant measured at zero is
/// NEVER SEEN. They are different facts and nothing may fold them.**
///
/// The distinction this whole module turns on, at the one function that used
/// to destroy it. Until 2026-09-10 [`last_seen`] read each meter with
/// `.unwrap_or(0)`, so a product that does not count flow at all — an older
/// gunnar, the `Admin.StorageInfo` fallback, a tenant the poller has never
/// read — was indistinguishable from one whose meters were reported and read
/// zero. Both came out `Quiet::Never`.
///
/// What that put on the screen: `never seen`, on the fleet page's SILENT
/// list, sorted to the TOP by `quiet_for_ms`'s `u64::MAX` — above every real
/// finding. A claim about a customer, manufactured from a gap in a plugin,
/// ranked as the most urgent row on the page. The tenant page had already
/// refused to make that claim (*"this product does not report flow"*); the
/// fleet page made it.
///
/// **Measure, never assert**: every arm below has its twin, so a fold in
/// either direction goes red here.
#[test]
fn a_product_that_reports_no_flow_is_no_data_and_never_a_zero() {
// MEASURED, and the measurement is zero: the meters are present.
let measured_zero = reading(NOON, &[("last_push_unix_ms", 0), ("last_fetch_unix_ms", 0)]);
// NOT MEASURED: this product reports a stock meter and no flow meter.
let unmeasured = reading(NOON, &[("pack_bytes", 4_096)]);
assert_eq!(
last_seen(&measured_zero),
Some(0),
"reported, and it reads zero"
);
assert_eq!(
last_seen(&unmeasured),
None,
"no flow meter was reported at all"
);
assert_ne!(
last_seen(&measured_zero),
last_seen(&unmeasured),
"an absent meter and a meter reading zero must not be the same value"
);
assert_eq!(quiet_for(last_seen(&measured_zero), NOON), Quiet::Never);
assert_eq!(quiet_for(last_seen(&unmeasured), NOON), Quiet::NoData);
assert!(!quiet_for(last_seen(&measured_zero), NOON).is_no_data());
assert!(quiet_for(last_seen(&unmeasured), NOON).is_no_data());
// THE CONSEQUENCE, and the reason the arm exists: the SILENT list is a
// list of MEASURED silences. A gap is not a finding about a customer.
assert!(
quiet_for(last_seen(&measured_zero), NOON).at_least_days(14),
"a tenant measured as never used belongs on the SILENT list"
);
assert!(
!quiet_for(last_seen(&unmeasured), NOON).at_least_days(14),
"a tenant nobody measured must not be published as a silence"
);
// And it can never outrank a real finding if a caller lists it anyway.
assert_eq!(quiet_for_ms(last_seen(&unmeasured), NOON), 0);
assert!(
quiet_for_ms(last_seen(&unmeasured), NOON)
< quiet_for_ms(last_seen(&measured_zero), NOON)
);
// `days()` answers `None` for BOTH, which is why it is not enough on its
// own — a renderer that reads only this cannot tell them apart, and the
// one that matters is `Quiet` itself.
assert_eq!(quiet_for(last_seen(&measured_zero), NOON).days(), None);
assert_eq!(quiet_for(last_seen(&unmeasured), NOON).days(), None);
}
/// A reading from the future is a clock disagreement and reads as "just
/// now" — never as a negative age, and never as a tenant to put on the
/// SILENT list.
#[test]
fn a_clock_that_runs_ahead_does_not_make_a_busy_tenant_silent() {
assert_eq!(quiet_for(Some(NOON + DAY_MS), NOON), Quiet::For(0));
assert!(!quiet_for(Some(NOON + DAY_MS), NOON).at_least_days(1));
}
/// **A fetch counts as being seen.** A customer who clones every morning
/// and never pushes is using the product, and one writer of that decision
/// is what stops two lists disagreeing about it.
#[test]
fn last_seen_is_the_later_of_the_two_and_a_fetch_counts() {
let only_fetch = reading(
NOON,
&[("last_push_unix_ms", 0), ("last_fetch_unix_ms", NOON)],
);
assert_eq!(last_seen(&only_fetch), Some(NOON));
let older_fetch = reading(
NOON,
&[
("last_push_unix_ms", NOON),
("last_fetch_unix_ms", NOON - DAY_MS),
],
);
assert_eq!(last_seen(&older_fetch), Some(NOON), "the later of the two");
// ONE of the two reported is still a measurement — a product may count
// pushes and not fetches. Only NEITHER is an absence.
let push_only = reading(NOON, &[("last_push_unix_ms", NOON - DAY_MS)]);
assert_eq!(
last_seen(&push_only),
Some(NOON - DAY_MS),
"one reported meter is a measurement"
);
let neither = reading(NOON, &[("pack_bytes", 1)]);
assert_eq!(
last_seen(&neither),
None,
"a product that reports neither has not been MEASURED — it has not `never been seen`"
);
}
}