crates/monetize-server/src/usage_poll.rs
//! **The usage half of a poll tick, batched** (2026-10-06).
//!
//! The poller used to ask [`Product::read_usage`] once per tenant per tick, in
//! sequence. The load test on the UpCloud mockup measured what that costs the
//! appliance: ~25–30 s of 1.4 cores on gunnar's main every five minutes at
//! 1,000 tenants, linear (worse) in tenants, so at ~10,000 one tick took the
//! whole period and usage, billing and the room checks fell behind.
//!
//! Now a tick asks [`Product::read_usage_page`] — pages of many tenants per
//! call — and after the first tick only for the tenants whose usage CHANGED
//! since the cursor the last tick was given. A tenant that did not change is
//! written again with its last reading at the page's `as_of`, so the stored
//! series (and every trend computed from it) reads exactly as it did when
//! every tenant was asked every tick.
//!
//! **Version-aware, both ways.** A product, guest or appliance that predates the
//! batched read answers `None` (an older guest answers the frozen unknown-op
//! words, which the guest host turns into `None`; an older gunnar answers
//! UNIMPLEMENTED, which gunnar's guest turns into `None`), and the tick falls
//! back to one `read_usage` per tenant — with a bounded number in flight at
//! once, so even the old path no longer serialises a whole estate behind one
//! slow call, and never more than [`FALLBACK_CONCURRENCY`] calls press on the
//! box at a time.
use std::collections::{BTreeMap, HashMap};
use std::sync::atomic::{AtomicUsize, Ordering};
use std::sync::Mutex;
use monetize_product::{ProductError, TenantId, Usage, UsagePageReq};
use crate::state::AppState;
/// Rows asked for per page.
pub const PAGE_SIZE: u32 = 1_000;
/// Per-tenant reads in flight at once on the fallback path. TWO, not more:
/// the fallback is the path an OLDER gunnar takes, where every `Usage.Read`
/// costs main O(tenants) work, and the smallest main sold is four cores — two
/// in flight halves the tick's wall time and leaves main half its cores for
/// the git traffic it exists to serve.
pub const FALLBACK_CONCURRENCY: usize = 2;
/// A product that hands out pages forever is refused after this many.
pub const MAX_PAGES: usize = 100_000;
/// **Where each product's last read left off** — the cursor its last
/// successful batched read answered, by product id. In memory: a restart reads
/// every tenant once, which is the honest first tick.
#[derive(Default, Debug)]
pub struct Cursors(Mutex<HashMap<String, String>>);
impl Cursors {
fn get(&self, product: &str) -> String {
self.0.lock().unwrap_or_else(|p| p.into_inner()).get(product).cloned().unwrap_or_default()
}
fn set(&self, product: &str, cursor: String) {
let mut map = self.0.lock().unwrap_or_else(|p| p.into_inner());
if cursor.is_empty() {
map.remove(product);
} else {
map.insert(product.to_owned(), cursor);
}
}
}
/// How this tick's usage was read.
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub enum Mode {
/// Every tenant, in pages.
Batched,
/// Only what changed since the last cursor, in pages.
Changed,
/// One `read_usage` per tenant: the product has no batched read.
PerTenant,
}
impl Mode {
pub fn name(self) -> &'static str {
match self {
Mode::Batched => "batched",
Mode::Changed => "changed",
Mode::PerTenant => "per-tenant",
}
}
}
/// **This tick's usage, per listed tenant.**
#[derive(Debug)]
pub struct Collected {
/// A reading for every tenant that has one this tick: measured, or — on a
/// changed-only read — carried forward unchanged at the page's `as_of`.
pub readings: Vec<(String, Usage)>,
/// The tenants whose read failed, and why.
pub failed: Vec<(String, ProductError)>,
/// How many tenants the product actually measured and answered (the rest
/// were carried forward).
pub measured: usize,
/// Calls made into the product to read usage this tick.
pub calls: usize,
pub mode: Mode,
}
impl Collected {
/// The tick's usage line, for the log: what it cost the product.
pub fn line(&self, product: &str) -> String {
format!(
"usage {product} mode={} calls={} measured={} carried={} failures={}",
self.mode.name(),
self.calls,
self.measured,
self.readings.len().saturating_sub(self.measured),
self.failed.len()
)
}
}
/// One batched read, every page folded: the rows, the cursor to resume
/// from, whether it was a changed-only read, and its as-of.
type Pages = (BTreeMap<String, Usage>, String, bool, u64);
/// Every page of one batched read, or `None` when the product has none.
fn pages(product: monetize_product::Served<'_>, since: &str, calls: &mut usize) -> Result<Option<Pages>, ProductError> {
let mut rows = BTreeMap::new();
let mut token = String::new();
let mut cursor = String::new();
let mut changed_only = true;
let mut as_of = 0u64;
for n in 0.. {
if n >= MAX_PAGES {
return Err(ProductError::Unavailable(format!(
"{} answered more than {MAX_PAGES} pages of usage without a last one",
product.id()
)));
}
let req = UsagePageReq { page_token: token.clone(), page_size: PAGE_SIZE, changed_since: since.to_owned() };
*calls += 1;
let Some(page) = product.read_usage_page(&req)? else {
if n == 0 {
return Ok(None);
}
// The product stopped having a batched read mid-way (replaced
// under the tick): nothing half-read is trusted.
return Err(ProductError::Unavailable(format!("{}'s batched usage read went away mid-read", product.id())));
};
if n == 0 {
cursor = page.cursor.clone();
as_of = page.as_of_unix_ms;
}
changed_only &= page.changed_only;
for row in page.rows {
rows.insert(row.tenant.0, row.usage);
}
if page.next_page_token.is_empty() || page.next_page_token == token {
break;
}
token = page.next_page_token;
}
Ok(Some((rows, cursor, changed_only && !since.is_empty(), as_of)))
}
/// `read_usage` for each of `tenants`, at most [`FALLBACK_CONCURRENCY`] in
/// flight. Answers in `tenants`' order.
pub fn per_tenant(product: monetize_product::Served<'_>, tenants: &[TenantId]) -> Vec<(String, Result<Usage, ProductError>)> {
let next = AtomicUsize::new(0);
let out: Mutex<Vec<Option<Result<Usage, ProductError>>>> = Mutex::new((0..tenants.len()).map(|_| None).collect());
let workers = FALLBACK_CONCURRENCY.min(tenants.len()).max(1);
std::thread::scope(|scope| {
for _ in 0..workers {
// A32-14, stays a task: scoped threads, joined when the scope ends.
scope.spawn(|| loop {
let i = next.fetch_add(1, Ordering::Relaxed);
let Some(tenant) = tenants.get(i) else { break };
let answer = product.read_usage(tenant);
out.lock().unwrap_or_else(|p| p.into_inner())[i] = Some(answer);
});
}
});
let out = out.into_inner().unwrap_or_else(|p| p.into_inner());
tenants
.iter()
.zip(out)
.map(|(t, r)| (t.0.clone(), r.unwrap_or_else(|| Err(ProductError::Unavailable("not read".into())))))
.collect()
}
/// **Read this tick's usage for `tenants`.**
pub fn collect(state: &AppState, product: monetize_product::Served<'_>, tenants: &[TenantId]) -> Collected {
let cursors = &state.usage_cursors;
let since = cursors.get(product.id());
let mut calls = 0;
let mut out = Collected { readings: Vec::new(), failed: Vec::new(), measured: 0, calls: 0, mode: Mode::PerTenant };
let batched = match pages(product, &since, &mut calls) {
Ok(found) => found,
Err(e) => {
// The box did not answer the batched read: every tenant is a failed
// read this tick (as one `read_usage` each would have been), and the
// next tick starts over with a full read.
cursors.set(product.id(), String::new());
out.calls = calls;
out.mode = if since.is_empty() { Mode::Batched } else { Mode::Changed };
out.failed = tenants.iter().map(|t| (t.0.clone(), e.clone())).collect();
return out;
}
};
let Some((mut rows, cursor, changed_only, as_of)) = batched else {
// No batched read: the old path, bounded.
cursors.set(product.id(), String::new());
for (tenant, answer) in per_tenant(product, tenants) {
calls += 1;
match answer {
Ok(usage) => out.readings.push((tenant, usage)),
Err(e) => out.failed.push((tenant, e)),
}
}
out.measured = out.readings.len();
out.calls = calls;
out.mode = Mode::PerTenant;
return out;
};
out.mode = if changed_only { Mode::Changed } else { Mode::Batched };
// Unchanged tenants carry their last reading forward — read once for the
// product, not once per tenant.
let last: HashMap<String, Usage> = if changed_only {
state.usage.all(product.id()).map(|v| v.into_iter().map(|(t, u)| (t.0, u)).collect()).unwrap_or_default()
} else {
HashMap::new()
};
let mut missing = Vec::new();
for tenant in tenants {
if let Some(usage) = rows.remove(&tenant.0) {
out.measured += 1;
out.readings.push((tenant.0.clone(), usage));
} else if let Some(usage) = changed_only.then(|| last.get(&tenant.0)).flatten() {
out.readings.push((tenant.0.clone(), Usage { meters: usage.meters.clone(), measured_at_unix_ms: as_of.max(usage.measured_at_unix_ms) }));
} else {
// Listed, but not in the read and nothing held for it: asked on its
// own, so a tenant is never silently unread.
missing.push(tenant.clone());
}
}
for (tenant, answer) in per_tenant(product, &missing) {
calls += 1;
match answer {
Ok(usage) => {
out.measured += 1;
out.readings.push((tenant, usage));
}
Err(e) => out.failed.push((tenant, e)),
}
}
cursors.set(product.id(), cursor);
out.calls = calls;
out
}