crates/monetize-server/tests/usage_poll_batched.rs
//! **One poll tick at 10,000 tenants costs the product O(pages) calls, not one
//! per tenant — and a product that cannot batch is still polled** (2026-10-06).
//!
//! The load test on the UpCloud mockup measured the poller asking
//! `read_usage` once per tenant every 300 s: ~25–30 s of 1.4 cores on gunnar's
//! main per period at 1,000 tenants, so at ~10,000 one tick took the whole
//! period. A fake product here counts what a tick asks of it:
//!
//! * a product WITH a batched read: the first tick reads every tenant in
//! `ceil(N / PAGE_SIZE)` calls, the next tick (nothing changed) in one call,
//! and a tick after 1 % of tenants moved reads only those — still one call;
//! every reading lands in the store, the unchanged ones carried forward at the
//! page's `as_of`, so the stored series reads as it always did;
//! * a product WITHOUT one (an older plugin, an older guest, an older gunnar —
//! `read_usage_page` answers `None`): one `read_usage` per tenant, as before,
//! never more than `FALLBACK_CONCURRENCY` in flight at once.
//!
//! RED before the batched poller: the first test's tick made 10,000
//! `read_usage` calls.
use std::collections::BTreeMap;
use std::sync::atomic::{AtomicU64, AtomicUsize, Ordering};
use std::sync::{Arc, Mutex};
use std::time::{Duration, Instant};
use monetize_product::{
Absorb, EntitlementFact, Product, ProductError, Servable, TenantId, TenantUsage, Usage, UsagePage, UsagePageReq,
};
use monetize_server::poller;
use monetize_server::registry::Registry;
use monetize_server::state::AppState;
const N: usize = 10_000;
/// A product with N tenants that counts every call a tick makes into it.
struct Fake {
batched: bool,
tenants: Vec<TenantId>,
/// tenant → (pack bytes, the sequence it last changed at)
held: Mutex<BTreeMap<String, (u64, u64)>>,
seq: AtomicU64,
epoch: String,
reads: AtomicUsize,
pages: AtomicUsize,
in_flight: AtomicUsize,
peak_in_flight: AtomicUsize,
/// `read_usage` answers after this long (the fallback's concurrency shows).
read_delay: Duration,
}
impl Fake {
fn new(batched: bool, n: usize) -> Fake {
let tenants: Vec<TenantId> = (0..n).map(|i| TenantId(format!("t{i:05}"))).collect();
let held = tenants.iter().map(|t| (t.0.clone(), (1_000, 0))).collect();
Fake {
batched,
tenants,
held: Mutex::new(held),
seq: AtomicU64::new(0),
epoch: "fake1".into(),
reads: AtomicUsize::new(0),
pages: AtomicUsize::new(0),
in_flight: AtomicUsize::new(0),
peak_in_flight: AtomicUsize::new(0),
read_delay: Duration::ZERO,
}
}
fn touch(&self, tenant: &str) {
let s = self.seq.fetch_add(1, Ordering::SeqCst) + 1;
let mut held = self.held.lock().unwrap();
let e = held.get_mut(tenant).expect("a tenant");
*e = (e.0 + 77, s);
}
fn usage(pack: u64, at: u64) -> Usage {
Usage { meters: [("pack_bytes".to_string(), pack)].into_iter().collect(), measured_at_unix_ms: at }
}
fn calls(&self) -> (usize, usize) {
(self.reads.load(Ordering::SeqCst), self.pages.load(Ordering::SeqCst))
}
}
const NOW: u64 = 1_800_000_000_000;
impl Product for Fake {
fn facts(&self) -> monetize_product::ProductFacts {
monetize_product::ProductFacts::new("gunnar", &[])
}
fn list_tenants(&self) -> Result<Vec<TenantId>, ProductError> {
Ok(self.tenants.clone())
}
fn read_usage(&self, tenant: &TenantId) -> Result<Usage, ProductError> {
self.reads.fetch_add(1, Ordering::SeqCst);
let now = self.in_flight.fetch_add(1, Ordering::SeqCst) + 1;
self.peak_in_flight.fetch_max(now, Ordering::SeqCst);
if !self.read_delay.is_zero() {
std::thread::sleep(self.read_delay);
}
let pack = self.held.lock().unwrap().get(&tenant.0).map(|e| e.0);
self.in_flight.fetch_sub(1, Ordering::SeqCst);
pack.map(|p| Fake::usage(p, NOW)).ok_or_else(|| ProductError::Refused(format!("no tenant {}", tenant.0)))
}
fn read_usage_page(&self, req: &UsagePageReq) -> Result<Option<UsagePage>, ProductError> {
if !self.batched {
return Ok(None);
}
self.pages.fetch_add(1, Ordering::SeqCst);
// `fake1.<seq>`; a cursor of another epoch is "everything".
let since = req.changed_since.strip_prefix(&format!("{}.", self.epoch)).and_then(|s| s.parse::<u64>().ok());
let (cursor_seq, after) = match req.page_token.split_once('|') {
Some((c, a)) => (c.parse::<u64>().unwrap(), a.to_string()),
None => (self.seq.load(Ordering::SeqCst), String::new()),
};
let size = req.page_size.max(1) as usize;
let held = self.held.lock().unwrap();
let mut rows = Vec::new();
let mut more = false;
for (t, (pack, changed)) in held.range::<String, _>((std::ops::Bound::Excluded(after.clone()), std::ops::Bound::Unbounded)) {
if since.is_some_and(|s| *changed <= s) {
continue;
}
if rows.len() == size {
more = true;
break;
}
rows.push(TenantUsage { tenant: TenantId(t.clone()), usage: Fake::usage(*pack, NOW + 5) });
}
let next_page_token = if more { format!("{cursor_seq}|{}", rows.last().unwrap().tenant.0) } else { String::new() };
Ok(Some(UsagePage { rows, next_page_token, cursor: format!("{}.{cursor_seq}", self.epoch), changed_only: since.is_some(), as_of_unix_ms: NOW + 5 }))
}
fn push_entitlement(&self, _fact: &EntitlementFact) -> Result<(), ProductError> {
Ok(())
}
fn can_absorb(&self, _tenant: &TenantId, _caps: &BTreeMap<String, u64>) -> Result<Absorb, String> {
Ok(Absorb::OnHand)
}
fn servable(&self) -> Servable {
Servable::Unmeasured("a fake".into())
}
}
fn shop(fake: Arc<Fake>) -> AppState {
let product: Arc<dyn Product> = fake;
AppState::in_memory(Registry::new().shared_product(product))
}
fn timed_tick(state: &AppState) -> (poller::Tick, Duration) {
let t = Instant::now();
let tick = poller::tick(state);
(tick, t.elapsed())
}
/// ★ **N = 10,000: O(pages) calls per tick, not N** — and only what changed
/// after the first tick.
#[test]
fn a_tick_at_ten_thousand_tenants_reads_usage_in_pages_and_then_only_what_changed() {
let fake = Arc::new(Fake::new(true, N));
let state = shop(Arc::clone(&fake));
let pages = N.div_ceil(monetize_server::usage_poll::PAGE_SIZE as usize);
let (tick, took) = timed_tick(&state);
let (reads, page_calls) = fake.calls();
eprintln!("first tick N={N}: read_usage={reads} pages={page_calls} took={took:?}");
assert_eq!(reads, 0, "no per-tenant read: the product has a batched one (was {N} before the batched poller)");
assert_eq!(page_calls, pages, "every tenant in ceil(N / PAGE_SIZE) calls");
assert_eq!(tick.tenants_read, N);
assert_eq!(tick.read_failures, 0);
assert!(took < Duration::from_secs(60), "one tick at 10,000 tenants took {took:?}");
assert_eq!(state.usage.get("gunnar", "t04711").unwrap().unwrap().meters["pack_bytes"], 1_000);
// Nothing moved: one call answers the whole estate.
let (tick, took) = timed_tick(&state);
let (reads, page_calls2) = fake.calls();
eprintln!("quiet tick N={N}: read_usage={reads} pages={} took={took:?}", page_calls2 - page_calls);
assert_eq!(page_calls2 - page_calls, 1, "a quiet estate is one call");
assert_eq!(reads, 0);
assert_eq!(tick.tenants_read, N, "every tenant still has its reading this tick, carried forward");
let carried = state.usage.get("gunnar", "t04711").unwrap().unwrap();
assert_eq!(carried.meters["pack_bytes"], 1_000);
assert_eq!(carried.measured_at_unix_ms, NOW + 5, "carried forward at the page's as_of");
assert_eq!(state.usage.kept("gunnar", "t04711").unwrap(), 1, "same instant: the series is not doubled");
// One percent of the tenants move: still one call, and their new numbers land.
for i in (0..N).step_by(100) {
fake.touch(&format!("t{i:05}"));
}
let (tick, took) = timed_tick(&state);
let (reads, page_calls3) = fake.calls();
eprintln!("1% moved N={N}: read_usage={reads} pages={} took={took:?}", page_calls3 - page_calls2);
assert_eq!(page_calls3 - page_calls2, 1);
assert_eq!(reads, 0);
assert_eq!(tick.tenants_read, N);
assert_eq!(state.usage.get("gunnar", "t00100").unwrap().unwrap().meters["pack_bytes"], 1_077, "a moved tenant's new number");
assert_eq!(state.usage.get("gunnar", "t00101").unwrap().unwrap().meters["pack_bytes"], 1_000, "an unmoved tenant's old one");
}
/// ★ **Old product, new monetize**: no batched read (`None`) — every tenant is
/// read per tenant, as before, with at most `FALLBACK_CONCURRENCY` in flight.
#[test]
fn a_product_with_no_batched_read_is_polled_per_tenant_with_bounded_concurrency() {
let mut fake = Fake::new(false, 200);
fake.read_delay = Duration::from_millis(2);
let fake = Arc::new(fake);
let state = shop(Arc::clone(&fake));
let (tick, _) = timed_tick(&state);
let (reads, pages) = fake.calls();
assert_eq!(reads, 200, "one read_usage per tenant");
assert_eq!(pages, 0);
assert_eq!(tick.tenants_read, 200);
assert_eq!(tick.read_failures, 0);
let peak = fake.peak_in_flight.load(Ordering::SeqCst);
assert!(peak > 1, "the fallback no longer serialises the estate (peak {peak})");
assert!(peak <= monetize_server::usage_poll::FALLBACK_CONCURRENCY, "never more than the bound press on the box (peak {peak})");
assert_eq!(state.usage.get("gunnar", "t00042").unwrap().unwrap().meters["pack_bytes"], 1_000);
}
/// A product whose batched read FAILS (the box is down) costs one call and
/// counts every tenant as a failed read — as N failed `read_usage` calls did —
/// and the next tick starts over with a full read.
#[test]
fn a_down_box_is_one_failed_call_and_the_next_tick_reads_everything() {
struct Down(Fake, AtomicUsize);
impl Product for Down {
fn facts(&self) -> monetize_product::ProductFacts {
monetize_product::ProductFacts::new("gunnar", &[])
}
fn list_tenants(&self) -> Result<Vec<TenantId>, ProductError> {
self.0.list_tenants()
}
fn read_usage(&self, t: &TenantId) -> Result<Usage, ProductError> {
self.0.read_usage(t)
}
fn read_usage_page(&self, req: &UsagePageReq) -> Result<Option<UsagePage>, ProductError> {
if self.1.fetch_add(1, Ordering::SeqCst) == 1 {
return Err(ProductError::Unavailable("connection refused".into()));
}
self.0.read_usage_page(req)
}
fn push_entitlement(&self, f: &EntitlementFact) -> Result<(), ProductError> {
self.0.push_entitlement(f)
}
fn can_absorb(&self, t: &TenantId, c: &BTreeMap<String, u64>) -> Result<Absorb, String> {
self.0.can_absorb(t, c)
}
fn servable(&self) -> Servable {
self.0.servable()
}
}
let down = Arc::new(Down(Fake::new(true, 50), AtomicUsize::new(0)));
let product: Arc<dyn Product> = down.clone();
let state = AppState::in_memory(Registry::new().shared_product(product));
assert_eq!(poller::tick(&state).tenants_read, 50, "first tick: read");
let t = poller::tick(&state);
assert_eq!((t.tenants_read, t.read_failures), (0, 50), "the box was down: every tenant a failed read");
// The cursor was dropped: this tick asks for everything again (one page).
let before = down.0.pages.load(Ordering::SeqCst);
let t = poller::tick(&state);
assert_eq!((t.tenants_read, t.read_failures), (50, 0));
assert_eq!(down.0.pages.load(Ordering::SeqCst) - before, 1);
}