crates/monetize-server/src/testing/estate.rs
//! **A fake estate main for the server's tests** (A32-13 step 2): what a product's main says
//! and does over the pull channel, with no box behind it.
//!
//! In production every product the server serves is `Pulled`: its reads are answered from the
//! newest `Fleet.Report` of its estate's main (a [`ProductReport`]), and every change leaves as a
//! signed order the main pulls and acks. A test product (`super::products::TestProduct`) is
//! served the same way ([`crate::registry::Registry::pulled_product`]), and its estate is this:
//!
//! * [`FakeEstate::report`]: the main's word, edited by the test and sent at once (a call home);
//! * [`FakeEstate::call_home`]: the main pulls its queued orders, runs them and acks each in its
//! next report: an `entitlement` is HELD ([`FakeEstate::held`]) and its account listed, every
//! other kind is acked with
//! the answer the test scripted for it ([`FakeEstate::answer`]), else `ok`;
//! * [`FakeEstate::orders`]: what is queued for the main and not acked yet.
//!
//! The clock is the wall's, as `Pulled`'s own: a report is fresh for the guardian's bound.
use std::collections::BTreeMap;
use std::sync::{Arc, Mutex};
use monetize_embed::fleet::{OrderAck, Report, SignedOrder};
use monetize_guardian::runner::Guardian;
use monetize_product::guardian::role;
use monetize_product::pull::{order, ProductReport};
use monetize_product::{EntitlementFact, TenantId, TenantOwner, Usage};
use crate::state::AppState;
/// **One estate's main, faked** (see the module doc).
pub struct FakeEstate {
guardian: Arc<Guardian>,
instance: String,
held_report: Mutex<ProductReport>,
pulled: Mutex<Vec<EntitlementFact>>,
answers: Mutex<BTreeMap<String, (bool, String)>>,
ran: Mutex<Vec<SignedOrder>>,
}
/// How long a [`FakeEstate::call_home`] keeps the main's report fresh.
pub const FRESH_FOR_MS: u64 = 3_600_000;
/// The wall clock, in Unix ms (the one `Pulled` judges a report's age by).
fn now_ms() -> u64 {
std::time::SystemTime::now().duration_since(std::time::UNIX_EPOCH).map(|d| d.as_millis() as u64).unwrap_or(0)
}
impl FakeEstate {
/// **The estate `instance` of `product`, declared on `state`'s fleet** (as
/// `[products.<id>] instance = …` declares it at start), its main silent until the first
/// [`Self::report`]. Panics when the fleet store is not open (`AppState::in_memory` opens one).
pub fn declare(state: &AppState, product: &str, instance: &str) -> FakeEstate {
let guardian = Arc::clone(state.fleet.guardian().expect("the fleet store is open (AppState::in_memory opens one)"));
guardian.store.declare(instance, product).expect("the estate is declared");
FakeEstate {
guardian,
instance: instance.to_owned(),
held_report: Mutex::new(ProductReport::default()),
pulled: Mutex::new(Vec::new()),
answers: Mutex::new(BTreeMap::new()),
ran: Mutex::new(Vec::new()),
}
}
/// The estate's name.
pub fn instance(&self) -> &str {
&self.instance
}
/// **The main's word, edited and sent at once**: `edit` changes the report the main holds,
/// and the main calls home with it (running and acking its queued orders, [`Self::call_home`]).
pub fn report(&self, edit: impl FnOnce(&mut ProductReport)) {
edit(&mut self.held_report.lock().unwrap_or_else(|e| e.into_inner()));
self.call_home();
}
/// **`tenant` is listed** (a person's account when `person`), with `meters` measured now.
pub fn tenant(&self, tenant: &str, person: bool, meters: &[(&str, u64)]) {
let at = now_ms();
self.report(|r| {
r.tenants.retain(|t| t.tenant.0 != tenant);
r.tenants.push(TenantOwner { tenant: TenantId(tenant.into()), person, seller: false });
r.usage.insert(tenant.into(), Usage { meters: meters.iter().map(|(k, v)| (k.to_string(), *v)).collect(), measured_at_unix_ms: at });
});
}
/// **The main has no room of its own in `meters`, and bought iron helps** (the answer a sale
/// buys its iron on, as a box an operator grows onto an attached disk says it).
pub fn buys_iron(&self, meters: &[&str]) {
self.report(|r| {
for m in meters {
r.room.insert(m.to_string(), 0);
}
r.with_iron = Some("the test estate grows onto the bought disk".into());
});
}
/// **The main measures `servable_bytes` of `total_bytes` free** (a set nobody said grows).
pub fn measures(&self, servable_bytes: u64, total_bytes: u64) {
self.report(|r| {
r.servable = Some(monetize_product::Servable::Measured { servable_bytes, total_bytes, growth: monetize_product::Growth::Unknown("a test estate".into()), ceilings: Default::default() });
});
}
/// **A store that holds `bytes` more of `meter` and no more**: measured so on a SEALED set
/// (never resized, so nothing past it is sold and queued), room for that much in the report,
/// and no bought iron that helps (a raise past it is refused).
pub fn holds_up_to(&self, meter: &str, bytes: u64) {
self.report(|r| {
r.servable = Some(monetize_product::Servable::Measured { servable_bytes: bytes, total_bytes: bytes, growth: monetize_product::Growth::Sealed("a test estate that is never resized".into()), ceilings: Default::default() });
r.room.insert(meter.to_string(), bytes);
r.with_iron = None;
});
}
/// **How the main answers the next order of `kind`** (its ack: `ok`, `detail`), until changed.
/// Unscripted kinds are acked `ok` with no detail.
pub fn answer(&self, kind: &str, ok: bool, detail: impl Into<String>) {
self.answers.lock().unwrap_or_else(|e| e.into_inner()).insert(kind.into(), (ok, detail.into()));
}
/// **The main calls home**: it pulls every order queued for it, runs each (an entitlement is
/// held), and sends its report with one ack per order. Answers the orders it ran.
pub fn call_home(&self) -> Vec<SignedOrder> {
// Heard an hour AHEAD of the wall: a test slower than the guardian's silence bound (90 s)
// under load never sees its main go silent; [`Self::silent_for`] says when it does.
self.call_home_at(now_ms() + FRESH_FOR_MS)
}
/// **The main fell silent** `ago_ms` ago: its last report (and the acks in it) was heard
/// then, so nothing is judged on it (`Pulled` refuses an old report by name).
pub fn silent_for(&self, ago_ms: u64) {
self.call_home_at(now_ms().saturating_sub(ago_ms));
}
fn call_home_at(&self, now: u64) -> Vec<SignedOrder> {
let orders = self.orders();
let answers = self.answers.lock().unwrap_or_else(|e| e.into_inner()).clone();
let mut acked = Vec::new();
for o in &orders {
let (ok, detail) = if o.kind == order::ENTITLEMENT {
match serde_json::from_str::<EntitlementFact>(&o.body) {
Ok(f) => {
// The account the fact is for is one the main holds and lists.
let mut held = self.held_report.lock().unwrap_or_else(|e| e.into_inner());
if !held.tenants.iter().any(|t| t.tenant == f.tenant) {
held.tenants.push(TenantOwner { tenant: f.tenant.clone(), person: false, seller: false });
held.usage.entry(f.tenant.0.clone()).or_insert_with(|| Usage { meters: BTreeMap::new(), measured_at_unix_ms: now });
}
drop(held);
self.pulled.lock().unwrap_or_else(|e| e.into_inner()).push(f);
(true, String::new())
}
Err(e) => (false, format!("the entitlement does not parse: {e}")),
}
} else {
answers.get(&o.kind).cloned().unwrap_or((true, String::new()))
};
acked.push(OrderAck { seq: o.seq, ok, detail, receipt_json: String::new() });
}
let mut product = self.held_report.lock().unwrap_or_else(|e| e.into_inner()).clone();
product.measured_unix_ms = now;
let report = Report { product: Some(serde_json::to_string(&product).expect("a report serialises")), acked, measured_unix_ms: now, ..Default::default() };
self.guardian.store.record_report(&self.instance, role::MAIN, now, &report).expect("the report is recorded");
self.ran.lock().unwrap_or_else(|e| e.into_inner()).extend(orders.iter().cloned());
orders
}
/// **What is queued for the main** and not acked yet, oldest first.
pub fn orders(&self) -> Vec<SignedOrder> {
self.guardian.store.orders_for(&self.instance, role::MAIN, 0)
}
/// The queued orders of `kind`.
pub fn queued(&self, kind: &str) -> Vec<SignedOrder> {
self.orders().into_iter().filter(|o| o.kind == kind).collect()
}
/// **Every order the main has run**, in the order it ran them.
pub fn ran(&self) -> Vec<SignedOrder> {
self.ran.lock().unwrap_or_else(|e| e.into_inner()).clone()
}
/// **Every entitlement the main has pulled and run**, in the order it ran them.
pub fn pulled(&self) -> Vec<EntitlementFact> {
self.pulled.lock().unwrap_or_else(|e| e.into_inner()).clone()
}
/// **The entitlements the main holds**: the newest it ran per tenant.
pub fn held(&self) -> Vec<EntitlementFact> {
let mut held: Vec<EntitlementFact> = Vec::new();
for f in self.pulled() {
held.retain(|h| h.tenant != f.tenant);
held.push(f);
}
held
}
/// **Every entitlement sent to the main**: those it ran, then those still queued for it.
pub fn entitlements(&self) -> Vec<EntitlementFact> {
let mut all = self.pulled();
all.extend(self.queued(order::ENTITLEMENT).iter().filter_map(|o| serde_json::from_str(&o.body).ok()));
all
}
}