crates/monetize-guardian/src/pulled/tests.rs
//! G1 pull: the product monetize never dials.
use std::collections::BTreeMap;
use std::sync::{Arc, OnceLock};
use monetize::sign::Signer;
use monetize_embed::fleet::{verify_order, OrderAck, Report};
use monetize_product::pull::{order, AccountRoom, AddPrincipal, AddUsers, GrowRoom, ProductReport, PurgeAnswer, PurgeTenant};
use monetize_product::*;
use super::*;
use crate::runner::GuardianConfig;
use crate::store::Store;
const NOW: u64 = 1_900_000_000_000;
const GIB: u64 = 1 << 30;
/// The plugin under the wrapper: every verb that would dial the box panics.
struct Dialler;
impl Product for Dialler {
fn facts(&self) -> monetize_product::ProductFacts {
monetize_product::ProductFacts::new("gunnar", &[])
}
fn list_tenants(&self) -> Result<Vec<TenantId>, ProductError> {
panic!("dialled list_tenants")
}
fn read_usage(&self, _: &TenantId) -> Result<Usage, ProductError> {
panic!("dialled read_usage")
}
fn read_usage_page(&self, _: &UsagePageReq) -> Result<Option<UsagePage>, ProductError> {
panic!("dialled read_usage_page")
}
fn tenant_owners(&self) -> Result<Option<Vec<TenantOwner>>, ProductError> {
panic!("dialled tenant_owners")
}
fn read_activity(&self, _: &TenantId, _: u64, _: u64) -> Result<Activity, ProductError> {
panic!("dialled read_activity")
}
fn push_entitlement(&self, _: &EntitlementFact) -> Result<(), ProductError> {
panic!("dialled push_entitlement")
}
fn can_absorb(&self, _: &TenantId, _: &BTreeMap<String, u64>) -> Result<Absorb, String> {
panic!("dialled can_absorb")
}
fn servable(&self) -> Servable {
panic!("dialled servable")
}
fn metered_door(&self) -> MeteredDoor {
panic!("dialled metered_door")
}
fn users_fit(&self, _: &TenantId, _: &UsersAsk) -> Result<UsersAnswer, ProductError> {
panic!("dialled users_fit")
}
fn add_users(&self, _: &TenantId, _: &UsersAsk, _: &str) -> Result<UsersAnswer, ProductError> {
panic!("dialled add_users")
}
fn purge_tenant(&self, _: &TenantId, _: &str) -> Result<Purged, ProductError> {
panic!("dialled purge_tenant")
}
fn twins(&self) -> Result<Vec<TwinFill>, ProductError> {
panic!("dialled twins")
}
}
struct Fixture {
signer: Signer,
guardian: Arc<Guardian>,
now: Arc<std::sync::atomic::AtomicU64>,
pulled: Pulled,
lane: Lane,
}
fn fixture() -> Fixture {
let signer = Signer::from_bytes(&[5u8; 32]);
let store = Arc::new(Store::in_memory().unwrap());
store.declare("gunnar.rs", "gunnar").unwrap();
// G8: gunnar's orders are signed with gunnar's key; the fixture pins it to `signer`.
let keys = monetize::sign::Keyring::new(Signer::from_bytes(&[6u8; 32]));
keys.pin("gunnar", signer.clone()).unwrap();
let guardian = Arc::new(Guardian::new(store, Arc::new(keys), GuardianConfig::default()));
let slot: GuardianSlot = Arc::new(OnceLock::new());
let _ = slot.set(guardian.clone());
let now = Arc::new(std::sync::atomic::AtomicU64::new(NOW));
let n = now.clone();
let clock: Clock = Arc::new(move || n.load(std::sync::atomic::Ordering::SeqCst));
let lane = Lane::new("gunnar", "gunnar.rs", slot.clone(), clock.clone());
let pulled = Pulled::new(Arc::new(Dialler), "gunnar.rs", slot, clock).with_ack_wait(std::time::Duration::ZERO);
Fixture { signer, guardian, now, pulled, lane }
}
fn usage(pack: u64) -> Usage {
Usage { meters: [("pack_bytes".to_string(), pack)].into(), measured_at_unix_ms: NOW - 1_000 }
}
fn product_report() -> ProductReport {
ProductReport {
tenants: vec![
TenantOwner { tenant: TenantId("acme".into()), person: false, seller: false },
TenantOwner { tenant: TenantId("alice".into()), person: true, seller: false },
],
usage: [("acme".to_string(), usage(3 * GIB)), ("alice".to_string(), usage(GIB))].into(),
servable: Some(Servable::Measured { servable_bytes: 40 * GIB, total_bytes: 300 * GIB, growth: Growth::Sealed("raid0 laid once".into()), ceilings: BTreeMap::new() }),
metered: Some(MeteredDoor::Enforcing("Entitlement.Get answered".into())),
room: [("pack_bytes".to_string(), 10 * GIB)].into(),
promised: [("acme".to_string(), [("pack_bytes".to_string(), 5 * GIB)].into())].into(),
with_iron: None,
names: vec!["acme".into(), "alice".into(), "operator".into()],
accounts: BTreeMap::new(),
activity: BTreeMap::new(),
measured_unix_ms: NOW - 1_000,
}
}
fn report(f: &Fixture, p: &ProductReport) {
let r = Report { product: Some(serde_json::to_string(p).unwrap()), measured_unix_ms: p.measured_unix_ms, ..Default::default() };
f.guardian.store.record_report("gunnar.rs", role::MAIN, NOW - 1_000, &r).unwrap();
}
fn fact(tenant: &str, plan: &str) -> EntitlementFact {
EntitlementFact {
tenant: TenantId(tenant.into()),
plan: plan.into(),
state: State::Paid,
paid_until_unix_ms: Some(NOW + 30 * 86_400_000),
caps: [("pack_bytes".to_string(), 10 * GIB)].into(),
source: "test".into(),
signature: vec![1, 2, 3],
issued_unix_ms: Some(NOW),
issued_signature: Vec::new(),
}
}
#[test]
fn tenants_owners_and_usage_are_read_from_the_mains_report() {
let f = fixture();
report(&f, &product_report());
let p = &f.pulled;
assert_eq!(p.list_tenants().unwrap(), vec![TenantId("acme".into()), TenantId("alice".into())]);
assert_eq!(p.read_usage(&TenantId("acme".into())).unwrap(), usage(3 * GIB));
assert!(matches!(p.read_usage(&TenantId("bob".into())), Err(ProductError::Refused(w)) if w.contains("bob")));
let owners = p.tenant_owners().unwrap().expect("the report names its owners");
assert_eq!(owners.len(), 2);
assert!(owners[1].person);
let page = p.read_usage_page(&UsagePageReq::default()).unwrap().expect("one page, the whole report");
assert_eq!(page.rows.len(), 2);
assert!(page.next_page_token.is_empty());
assert_eq!(page.as_of_unix_ms, NOW - 1_000);
}
#[test]
fn before_the_first_report_it_is_unavailable_and_never_dials() {
let f = fixture();
let p = &f.pulled;
match p.list_tenants() {
Err(ProductError::Unavailable(w)) => assert!(w.contains("gunnar.rs") && w.contains("report"), "{w}"),
other => panic!("{other:?}"),
}
assert!(matches!(p.read_usage(&TenantId("acme".into())), Err(ProductError::Unavailable(_))));
assert!(matches!(p.servable(), Servable::Unmeasured(w) if w.contains("gunnar.rs")));
assert!(matches!(p.metered_door(), MeteredDoor::Unproven(_)));
assert!(p.can_absorb(&TenantId("acme".into()), &[("pack_bytes".to_string(), GIB)].into()).is_err());
}
#[test]
fn an_entitlement_is_a_signed_order_the_main_pulls_and_newest_per_tenant_wins() {
let f = fixture();
let p = &f.pulled;
p.push_entitlement(&fact("acme", "gunnar/10gb-1m")).unwrap();
p.push_entitlement(&fact("alice", "gunnar/10gb-1m")).unwrap();
p.push_entitlement(&fact("acme", "gunnar/50gb-1m")).unwrap();
let queued = f.guardian.store.orders_for("gunnar.rs", role::MAIN, 0);
assert_eq!(queued.len(), 2, "the older acme fact is superseded: {queued:?}");
let pk = f.signer.verifying_key();
for o in &queued {
assert_eq!(o.kind, order::ENTITLEMENT);
assert_eq!(o.product, "gunnar");
verify_order(o, &pk, ("gunnar", "gunnar.rs", role::MAIN), NOW, 0).expect("signed by monetize for the main");
assert!(o.not_after_unix_ms >= NOW + ENTITLEMENT_TTL_MS);
}
let acme: EntitlementFact = serde_json::from_str(&queued.iter().find(|o| o.body.contains("acme")).unwrap().body).unwrap();
assert_eq!(acme, fact("acme", "gunnar/50gb-1m"), "the body is the signed fact, whole");
}
/// ★ G16 (ONE monetize): a SOLD estate's tenant (`@acme/bob` on the ONE ledger) is pulled by
/// THAT estate's main — a signed order for `acme`, the fact for `bob` as acme's gunnar knows it,
/// verifying under monetize's key — and nothing of it is queued for the product's own estate.
#[test]
fn a_sold_estates_tenant_is_pulled_by_that_estates_main_as_its_namespace() {
let f = fixture();
let booked = f.signer.sign_fact_at(fact(&TenantId::on_estate("acme", "bob").0, "gunnar/@acme/bob/2026-10-10+pack"), NOW);
match f.pulled.push_entitlement(&booked) {
Err(ProductError::Unavailable(w)) => assert!(w.contains("acme") && w.contains("fleet"), "{w}"),
other => panic!("an estate the fleet does not know is refused by name: {other:?}"),
}
assert!(f.guardian.store.queued("acme").is_empty() && f.guardian.store.queued("gunnar.rs").is_empty());
f.guardian.store.declare("acme", "gunnar").unwrap();
f.pulled.push_entitlement(&booked).unwrap();
assert!(f.guardian.store.orders_for("gunnar.rs", role::MAIN, 0).is_empty(), "nothing of acme's for the product's own main");
let queued = f.guardian.store.orders_for("acme", role::MAIN, 0);
assert_eq!(queued.len(), 1, "{queued:?}");
verify_order(&queued[0], &f.signer.verifying_key(), ("gunnar", "acme", role::MAIN), NOW, 0).expect("signed by monetize for acme's main");
let held: EntitlementFact = serde_json::from_str(&queued[0].body).unwrap();
assert_eq!(held, f.signer.on_estate(&booked, "acme").unwrap(), "bob's fact as acme's gunnar verifies it");
assert_eq!(held.tenant, TenantId("bob".into()));
monetize::sign::verify_fact(&held, &f.signer.verifying_key()).unwrap();
// A newer fact for the same tenant supersedes it there.
let newer = f.signer.sign_fact_at(fact(&TenantId::on_estate("acme", "bob").0, "gunnar/@acme/bob/2026-11-10+pack"), NOW + 1);
f.pulled.push_entitlement(&newer).unwrap();
let queued = f.guardian.store.orders_for("acme", role::MAIN, 0);
assert_eq!(queued.len(), 1, "newest per tenant wins on the estate too: {queued:?}");
assert!(queued[0].body.contains("2026-11-10"));
}
#[test]
fn can_absorb_answers_from_a_fresh_report_and_never_yes_on_a_stale_one() {
let f = fixture();
let mut r = product_report();
report(&f, &r);
let acme = TenantId("acme".into());
let caps = |gib: u64| -> BTreeMap<String, u64> { [("pack_bytes".to_string(), gib * GIB)].into() };
assert_eq!(f.pulled.can_absorb(&acme, &caps(12)), Ok(Absorb::OnHand), "needs 7 of 10 GiB room");
let short = f.pulled.can_absorb(&acme, &caps(20)).unwrap_err();
assert!(short.contains("pack_bytes"), "{short}");
r.with_iron = Some("an operator extends /data".into());
report(&f, &r);
assert_eq!(f.pulled.can_absorb(&acme, &caps(20)), Ok(Absorb::WithIron { note: "an operator extends /data".into() }));
f.now.store(NOW + 10 * 60_000, std::sync::atomic::Ordering::SeqCst);
let stale = f.pulled.can_absorb(&acme, &caps(12)).unwrap_err();
assert!(stale.contains("silent"), "{stale}");
assert!(matches!(f.pulled.servable(), Servable::Unmeasured(w) if w.contains("silent")));
}
#[test]
fn servable_and_the_paid_door_come_from_the_report() {
let f = fixture();
report(&f, &product_report());
assert!(matches!(f.pulled.servable(), Servable::Measured { servable_bytes, .. } if servable_bytes == 40 * GIB));
assert_eq!(f.pulled.metered_door().name(), "enforcing");
}
#[test]
fn growth_verbs_of_the_box_are_none_and_never_dial() {
let f = fixture();
report(&f, &product_report());
let p = &f.pulled;
assert!(p.grow().is_none() && p.twin_grow().is_none());
assert_eq!(p.twins().unwrap(), Vec::new());
assert_eq!(p.facts().id, "gunnar");
}
/// ★ UW-10: **a tenant's activity is read from its main's report** (the days in the asked
/// window, and how far back the box remembers); a report without it says so, never an
/// empty series that would read as a silent tenant.
#[test]
fn activity_is_read_from_the_mains_report_for_the_asked_days() {
let f = fixture();
report(&f, &product_report());
let acme = TenantId("acme".into());
match f.pulled.read_activity(&acme, 0, NOW) {
Err(ProductError::Unavailable(w)) => assert!(w.contains("activity"), "{w}"),
other => panic!("no activity in the report: unavailable, never empty: {other:?}"),
}
const DAY: u64 = 86_400_000;
let day = |n: u64| ActivityDay { day_unix_ms: NOW - n * DAY, meters: [("pushes".to_string(), n)].into() };
let mut p = product_report();
p.activity.insert("acme".into(), Activity { days: vec![day(20), day(5), day(1)], retained_days: 90 });
report(&f, &p);
let a = f.pulled.read_activity(&acme, NOW - 10 * DAY, NOW).unwrap();
assert_eq!((a.days.iter().map(|d| d.meters["pushes"]).collect::<Vec<_>>(), a.retained_days), (vec![5, 1], 90), "only the asked window");
assert!(f.pulled.read_activity(&TenantId("alice".into()), 0, NOW).is_err(), "a tenant the report carries none for");
}
/// ★ UW-11: **the nonpay purge is a signed order the main pulls, asked again until it
/// settles**: the first call queues ONE `purge_tenant` start (still locked); the main's ack
/// names the job, and while it runs each call queues ONE status order for THAT job (still
/// locked); a finished job is the `Purged`; a FAILED one is refused, so nothing is released.
#[test]
fn a_purge_is_a_signed_order_asked_again_until_the_job_settles() {
let f = fixture();
report(&f, &product_report());
let acme = TenantId("acme".into());
let why = "nonpay: acme lapsed 2026-01-01";
for _ in 0..2 {
let p = f.pulled.purge_tenant(&acme, why).expect("queued is an answer: still locked");
assert!(p.still_locked, "{p:?}");
}
let q = f.guardian.store.orders_for("gunnar.rs", role::MAIN, 0);
assert_eq!(q.len(), 1, "asked twice, one start: {q:?}");
assert_eq!(q[0].kind, order::PURGE_TENANT);
verify_order(&q[0], &f.signer.verifying_key(), ("gunnar", "gunnar.rs", role::MAIN), NOW, 0).expect("signed for the main");
let body: PurgeTenant = serde_json::from_str(&q[0].body).unwrap();
assert_eq!((body.tenant.as_str(), body.reason.as_str(), body.job.as_str()), ("acme", why, ""));
let running = PurgeAnswer { job: "j-1".into(), running: true, detail: "1 of 5 stores, 1.0 GiB".into(), still_locked: true, ..Default::default() };
ack(&f, q[0].seq, true, &serde_json::to_string(&running).unwrap());
let p = f.pulled.purge_tenant(&acme, why).unwrap();
assert!(p.still_locked && p.detail.contains("1 of 5"), "{p:?}");
let q2 = f.guardian.store.orders_for("gunnar.rs", role::MAIN, 0);
assert_eq!(q2.len(), 1, "{q2:?}");
let body: PurgeTenant = serde_json::from_str(&q2[0].body).unwrap();
assert_eq!(body.job, "j-1", "the status of THAT job");
let done = PurgeAnswer { job: "j-1".into(), detail: "5 of 5 stores, 3.0 GiB".into(), bytes_reclaimed: Some(3 * GIB), ..Default::default() };
ack(&f, q2[0].seq, true, &serde_json::to_string(&done).unwrap());
let p = f.pulled.purge_tenant(&acme, why).unwrap();
assert_eq!((p.still_locked, p.bytes_reclaimed, p.tenant.0.as_str()), (false, Some(3 * GIB), "acme"), "{p:?}");
assert!(f.guardian.store.orders_for("gunnar.rs", role::MAIN, 0).is_empty(), "settled: nothing asked again");
let bob = TenantId("alice".into());
f.pulled.purge_tenant(&bob, "nonpay: alice").unwrap();
let q3 = f.guardian.store.orders_for("gunnar.rs", role::MAIN, 0);
let failed = PurgeAnswer { job: "j-2".into(), failed: true, error: "disk read-only".into(), failures: vec!["alice/repo".into()], still_locked: true, ..Default::default() };
ack(&f, q3[0].seq, true, &serde_json::to_string(&failed).unwrap());
match f.pulled.purge_tenant(&bob, "nonpay: alice") {
Err(ProductError::Refused(w)) => assert!(w.contains("disk read-only") && w.contains("alice/repo"), "{w}"),
other => panic!("a failed purge releases nothing: {other:?}"),
}
}
fn signup(username: &str, reference: &str) -> AddPrincipal {
AddPrincipal { username: username.into(), kind: "person".into(), signup: reference.into() }
}
fn ack(f: &Fixture, seq: u64, ok: bool, detail: &str) {
let r = Report { acked: vec![OrderAck { seq, ok, detail: detail.into(), receipt_json: String::new() }], product: Some(serde_json::to_string(&product_report()).unwrap()), ..Default::default() };
f.guardian.store.record_report("gunnar.rs", role::MAIN, NOW - 500, &r).unwrap();
}
#[test]
fn a_signups_principal_is_one_signed_order_until_the_main_acks_it() {
let f = fixture();
let bob = signup("bob", "stripe:cs_1");
assert_eq!(f.lane.add_principal(&bob), Ok(Principal::Queued));
assert_eq!(f.lane.add_principal(&bob), Ok(Principal::Queued), "asked again: the same order, not a second");
let queued = f.guardian.store.orders_for("gunnar.rs", role::MAIN, 0);
assert_eq!(queued.len(), 1, "{queued:?}");
let o = &queued[0];
assert_eq!(o.kind, order::ADD_PRINCIPAL);
verify_order(o, &f.signer.verifying_key(), ("gunnar", "gunnar.rs", role::MAIN), NOW, 0).expect("signed by monetize for the main");
assert_eq!(serde_json::from_str::<AddPrincipal>(&o.body).unwrap(), bob);
ack(&f, o.seq, true, "bob");
assert_eq!(f.lane.add_principal(&bob), Ok(Principal::Created("bob".into())), "the namespace the main made is its ack's detail");
assert!(f.guardian.store.orders_for("gunnar.rs", role::MAIN, 0).is_empty(), "nothing queued again once acked");
}
#[test]
fn a_taken_ack_is_the_name_and_any_other_refusal_is_final() {
let f = fixture();
let (bob, eve) = (signup("bob", "stripe:cs_1"), signup("eve", "stripe:cs_2"));
f.lane.add_principal(&bob).unwrap();
f.lane.add_principal(&eve).unwrap();
let q = f.guardian.store.orders_for("gunnar.rs", role::MAIN, 0);
ack(&f, q[0].seq, false, "taken: bob is somebody else's");
ack(&f, q[1].seq, false, "the principal store is read-only");
assert_eq!(f.lane.add_principal(&bob), Ok(Principal::Taken("taken: bob is somebody else's".into())));
assert_eq!(f.lane.add_principal(&eve), Ok(Principal::Refused("the principal store is read-only".into())));
}
#[test]
fn an_order_neither_queued_nor_acked_any_more_is_queued_again() {
let f = fixture();
let bob = signup("bob", "stripe:cs_1");
f.lane.add_principal(&bob).unwrap();
let first = f.guardian.store.orders_for("gunnar.rs", role::MAIN, 0)[0].seq;
ack(&f, first, true, "bob");
// The ack scrolls out of the kept window: the box answered, monetize no longer knows what.
for _ in 0..crate::store::ACKS_KEPT + 1 {
let o = f.guardian.queue_order(Guardian::order("gunnar", "gunnar.rs", role::MAIN, "review_twin", "{}".into(), NOW, 60_000), |_| false).unwrap();
ack(&f, o.seq, true, "");
}
assert_eq!(f.lane.add_principal(&bob), Ok(Principal::Queued), "asked again: the box is idempotent per signup");
let q = f.guardian.store.orders_for("gunnar.rs", role::MAIN, 0);
assert_eq!(q.len(), 1);
assert!(q[0].seq > first);
}
#[test]
fn a_name_is_taken_when_the_main_reports_it_or_a_signup_already_asked_for_it() {
let f = fixture();
assert!(f.lane.name_taken("zed").is_err(), "before a report: unknown, never free");
report(&f, &product_report());
assert_eq!(f.lane.name_taken("zed"), Ok(false));
assert_eq!(f.lane.name_taken("operator"), Ok(true), "a principal with no tenant still holds its name");
f.lane.add_principal(&signup("zed", "stripe:cs_9")).unwrap();
assert_eq!(f.lane.name_taken("zed"), Ok(true), "a paid signup is creating it");
f.now.store(NOW + 10 * 60_000, std::sync::atomic::Ordering::SeqCst);
assert!(f.lane.name_taken("yan").unwrap_err().contains("silent"), "never free on an old report");
}
#[test]
fn the_lane_queues_an_entitlement_as_pulled_does() {
let f = fixture();
f.lane.queue_entitlement(&fact("acme", "gunnar/10gb-1m")).unwrap();
f.lane.queue_entitlement(&fact("acme", "gunnar/50gb-1m")).unwrap();
let q = f.guardian.store.orders_for("gunnar.rs", role::MAIN, 0);
assert_eq!(q.len(), 1);
assert_eq!(q[0].kind, order::ENTITLEMENT);
}
// ── many estates of one product (G1 multi-estate, ONE monetize) ─────────────────────────────
/// `estate`'s main reports `p` (its own namespaces), heard at `NOW - 1_000`.
fn report_of(f: &Fixture, estate: &str, p: &ProductReport) {
let r = Report { product: Some(serde_json::to_string(p).unwrap()), measured_unix_ms: p.measured_unix_ms, ..Default::default() };
f.guardian.store.record_report(estate, role::MAIN, NOW - 1_000, &r).unwrap();
}
/// A sold estate's main: one person, bob, using 2 GiB, promised 3, with 2 GiB room.
fn acme_report() -> ProductReport {
ProductReport {
tenants: vec![TenantOwner { tenant: TenantId("bob".into()), person: true, seller: false }],
usage: [("bob".to_string(), usage(2 * GIB))].into(),
room: [("pack_bytes".to_string(), 2 * GIB)].into(),
promised: [("bob".to_string(), [("pack_bytes".to_string(), 3 * GIB)].into())].into(),
names: vec!["bob".into()],
measured_unix_ms: NOW - 2_000,
..Default::default()
}
}
/// ★ **Every estate of the product is one product to monetize**: the product's own estate's
/// tenants as they are, each SOLD estate's as `@<estate>/<ns>` (M-D32), read from that estate's
/// own main's report. An estate of another product is not this product's.
#[test]
fn a_sold_estates_tenants_owners_and_usage_join_the_products_as_estate_tenants() {
let f = fixture();
report(&f, &product_report());
f.guardian.store.declare("acme", "gunnar").unwrap();
report_of(&f, "acme", &acme_report());
f.guardian.store.declare("holger.rs", "holger").unwrap();
report_of(&f, "holger.rs", &acme_report());
let p = &f.pulled;
let bob = TenantId::on_estate("acme", "bob");
assert_eq!(p.list_tenants().unwrap(), vec![TenantId("acme".into()), TenantId("alice".into()), bob.clone()], "holger.rs is holger's, not gunnar's");
assert_eq!(p.read_usage(&bob).unwrap(), usage(2 * GIB), "bob's usage is acme's main's word");
assert!(matches!(p.read_usage(&TenantId::on_estate("acme", "carol")), Err(ProductError::Refused(w)) if w.contains("acme") && w.contains("carol")));
assert!(matches!(p.read_usage(&TenantId::on_estate("zed", "x")), Err(ProductError::Refused(w)) if w.contains("zed")), "an estate the fleet does not know for gunnar");
assert!(matches!(p.read_usage(&TenantId::on_estate("holger.rs", "bob")), Err(ProductError::Refused(w)) if w.contains("holger.rs")));
let owners = p.tenant_owners().unwrap().unwrap();
assert_eq!(owners.len(), 3);
assert_eq!(owners[2], TenantOwner { tenant: bob.clone(), person: true, seller: false });
let page = p.read_usage_page(&UsagePageReq::default()).unwrap().unwrap();
assert_eq!(page.rows.iter().map(|r| r.tenant.0.as_str()).collect::<Vec<_>>(), ["acme", "alice", "@acme/bob"]);
assert_eq!(page.as_of_unix_ms, NOW - 2_000, "the page is as old as its oldest report");
}
/// ★ **A sold estate that has not called home is left out, never dialled**: the product's
/// own tenants are still listed; the silent estate's tenant is unavailable by name.
#[test]
fn a_silent_sold_estate_is_left_out_and_never_dialled() {
let f = fixture();
report(&f, &product_report());
f.guardian.store.declare("beta", "gunnar").unwrap();
assert_eq!(f.pulled.list_tenants().unwrap().len(), 2);
assert_eq!(f.pulled.tenant_owners().unwrap().unwrap().len(), 2);
assert!(matches!(f.pulled.read_usage(&TenantId::on_estate("beta", "x")), Err(ProductError::Unavailable(w)) if w.contains("beta")));
}
/// ★ **can_absorb for a sold estate's tenant is that estate's room**, judged on a fresh report.
#[test]
fn can_absorb_for_a_sold_estates_tenant_answers_from_that_estates_room() {
let f = fixture();
report(&f, &product_report());
f.guardian.store.declare("acme", "gunnar").unwrap();
report_of(&f, "acme", &acme_report());
let bob = TenantId::on_estate("acme", "bob");
let caps = |gib: u64| -> BTreeMap<String, u64> { [("pack_bytes".to_string(), gib * GIB)].into() };
assert_eq!(f.pulled.can_absorb(&bob, &caps(5)), Ok(Absorb::OnHand), "needs 2 beyond the 3 promised: acme has 2");
let short = f.pulled.can_absorb(&bob, &caps(6)).unwrap_err();
assert!(short.contains("acme") && short.contains("pack_bytes"), "the 10 GiB of the product's own estate do not count: {short}");
f.now.store(NOW + 10 * 60_000, std::sync::atomic::Ordering::SeqCst);
assert!(f.pulled.can_absorb(&bob, &caps(5)).unwrap_err().contains("silent"));
}
// ── ★ UW-5: "Add users" on an attached estate ────────────────────────────────
fn with_rooms() -> ProductReport {
let mut p = product_report();
p.accounts = [("acme".to_string(), AccountRoom { unused_bytes: Some(15 * GIB), room_bytes: 10 * GIB, seats: 2, seats_open: 1 })].into();
p
}
/// ★ **Would the users fit** is answered from the main's report — the account's room
/// neither stored nor promised, its seats, the product's room per user — before any money
/// moves; a name the box holds is refused; an account the report says nothing of is not
/// guessed at.
#[test]
fn users_fit_is_answered_from_the_mains_account_room() {
let f = fixture();
report(&f, &with_rooms());
let acme = TenantId("acme".into());
let a = f.pulled.users_fit(&acme, &UsersAsk { count: 2, ..Default::default() }).unwrap();
assert_eq!((a.room_bytes, a.need_bytes, a.unused_bytes, a.shortfall_bytes, a.seats, a.seats_open), (10 * GIB, 20 * GIB, Some(15 * GIB), 5 * GIB, 2, 1));
assert!(a.unchanged, "a dry run writes nothing");
let fits = f.pulled.users_fit(&acme, &UsersAsk { count: 1, room_bytes: 12 * GIB, name: "eve".into(), ..Default::default() }).unwrap();
assert_eq!((fits.need_bytes, fits.shortfall_bytes), (12 * GIB, 0));
match f.pulled.users_fit(&acme, &UsersAsk { count: 1, name: "alice".into(), ..Default::default() }) {
Err(ProductError::Refused(w)) => assert!(w.contains("alice"), "{w}"),
other => panic!("a held name: {other:?}"),
}
match f.pulled.users_fit(&TenantId("alice".into()), &UsersAsk { count: 1, name: "zed".into(), ..Default::default() }) {
Err(ProductError::Unavailable(w)) => assert!(w.contains("alice"), "{w}"),
other => panic!("no room said for the account: {other:?}"),
}
}
/// ★ **The grant is ONE signed order the main pulls** (`add_users`), per order reference,
/// with no address in it (monetize mails the invite); the main's ack is the answer — ok
/// with the product's `UsersAnswer`, refused final (monetize refunds).
#[test]
fn add_users_is_one_signed_order_until_the_main_acks_it() {
let f = fixture();
report(&f, &with_rooms());
let acme = TenantId("acme".into());
let ask = UsersAsk { count: 1, room_bytes: 10 * GIB, name: "eve".into(), email: "eve@example.invalid".into(), invite_origin: "https://test2.gunnar.rs".into(), principal_id: "p-1".into() };
for _ in 0..2 {
match f.pulled.add_users(&acme, &ask, "gunnar/acme/users-1") {
Err(ProductError::Unavailable(w)) => assert!(w.contains("queued"), "{w}"),
other => panic!("queued, not answered yet: {other:?}"),
}
}
let q = f.guardian.store.orders_for("gunnar.rs", role::MAIN, 0);
assert_eq!(q.len(), 1, "asked twice, one order: {q:?}");
assert_eq!(q[0].kind, order::ADD_USERS);
verify_order(&q[0], &f.signer.verifying_key(), ("gunnar", "gunnar.rs", role::MAIN), NOW, 0).expect("signed for the main");
let body: AddUsers = serde_json::from_str(&q[0].body).unwrap();
assert_eq!((body.tenant.as_str(), body.order.as_str(), body.ask.name.as_str(), body.ask.count), ("acme", "gunnar/acme/users-1", "eve", 1));
assert!(body.ask.email.is_empty() && !q[0].body.contains("eve@"), "no address leaves monetize: {}", q[0].body);
let answer = UsersAnswer { principal: "eve".into(), room_bytes: 10 * GIB, need_bytes: 10 * GIB, unused_bytes: Some(15 * GIB), seats: 3, seats_open: 1, invite_secret: "s3cret".into(), ..Default::default() };
ack(&f, q[0].seq, true, &serde_json::to_string(&answer).unwrap());
assert_eq!(f.pulled.add_users(&acme, &ask, "gunnar/acme/users-1").ok(), Some(answer));
f.pulled.add_users(&acme, &ask, "gunnar/acme/users-2").unwrap_err();
let q2 = f.guardian.store.orders_for("gunnar.rs", role::MAIN, 0);
ack(&f, q2[0].seq, false, "eve is somebody else's");
match f.pulled.add_users(&acme, &ask, "gunnar/acme/users-2") {
Err(ProductError::Refused(w)) => assert_eq!(w, "eve is somebody else's"),
other => panic!("refused, final: {other:?}"),
}
}
// ── ★ UW-4: past the room on an attached estate: sold, queued, resized later ──────────
/// ★ **An estate whose set grows at runtime queues room** (owner: the customer never sees a
/// cap — past room is queued and resized later, the operator told): `queues_room` says so
/// from the main's report, and `queue_room` is ONE signed `grow_room` order the main pulls,
/// the newest want replacing an unacked older one. A sealed set queues nothing.
#[test]
fn past_the_room_an_attached_estate_is_told_to_grow_by_a_signed_order() {
let f = fixture();
report(&f, &product_report());
assert!(!f.pulled.queues_room(), "a sealed set cannot be resized later");
let mut grows = product_report();
grows.servable = Some(Servable::Measured { servable_bytes: 40 * GIB, total_bytes: 300 * GIB, growth: Growth::AtRuntime("grows in place".into()), ceilings: BTreeMap::new() });
report(&f, &grows);
assert!(f.pulled.queues_room());
f.pulled.queue_room(&TenantId("acme".into()), 20 * GIB, "acme needs 20 GiB past the 10 the main has").unwrap();
f.pulled.queue_room(&TenantId("acme".into()), 30 * GIB, "acme needs 30 GiB past the 10 the main has").unwrap();
let q = f.guardian.store.orders_for("gunnar.rs", role::MAIN, 0);
assert_eq!(q.len(), 1, "the newest want replaces the unacked older one: {q:?}");
assert_eq!(q[0].kind, order::GROW_ROOM);
let body: GrowRoom = serde_json::from_str(&q[0].body).unwrap();
assert_eq!(body.need_bytes, 30 * GIB);
assert!(body.reason.contains("acme"), "{body:?}");
}
/// ★ ST-11: **another estate of the product answers what ITS main can serve** — from that
/// estate's own fresh report, never the home's; the home answers as `servable`.
#[test]
fn another_estates_servable_is_its_own_mains() {
let f = fixture();
report(&f, &product_report());
f.guardian.store.declare("test2-gunnar-rs", "gunnar").unwrap();
let mut t2 = product_report();
t2.servable = Some(Servable::Measured { servable_bytes: 7 * GIB, total_bytes: 9 * GIB, growth: Growth::Sealed("x".into()), ceilings: BTreeMap::new() });
let r = Report { product: Some(serde_json::to_string(&t2).unwrap()), measured_unix_ms: t2.measured_unix_ms, ..Default::default() };
f.guardian.store.record_report("test2-gunnar-rs", role::MAIN, NOW - 1_000, &r).unwrap();
let bytes = |s: Servable| match s {
Servable::Measured { servable_bytes, .. } => servable_bytes,
other => panic!("{other:?}"),
};
assert_eq!(bytes(f.pulled.servable_on("test2-gunnar-rs")), 7 * GIB, "test2's own main");
assert_eq!(bytes(f.pulled.servable_on("gunnar.rs")), 40 * GIB, "the home: its servable");
assert!(matches!(f.pulled.servable_on("nowhere"), Servable::Unmeasured(_)), "an estate never heard: unmeasured, never another's number");
}
// ── ALICA-4: a paid fact reaches the site's main even when its first order is refused ───────
/// test2's main: alica holds nothing yet (her account is made, her plan is not), bob holds 3 GiB.
fn test2_report(alica_gib: Option<u64>) -> ProductReport {
let mut promised: BTreeMap<String, BTreeMap<String, u64>> = [("bob".to_string(), [("pack_bytes".to_string(), 3 * GIB)].into())].into();
if let Some(g) = alica_gib {
promised.insert("alica".into(), [("pack_bytes".to_string(), g * GIB)].into());
}
ProductReport { tenants: vec![TenantOwner { tenant: TenantId("alica".into()), person: true, seller: false }], promised, names: vec!["alica".into(), "bob".into()], measured_unix_ms: NOW - 1_000, ..Default::default() }
}
fn entitlement_orders(f: &Fixture, estate: &str) -> Vec<SignedOrder> {
f.guardian.store.queued(estate).into_iter().filter(|o| o.kind == order::ENTITLEMENT).collect()
}
/// ★ ALICA-4 (live 2026-10-26): **a paid account fact on the ledger reaches the site's main
/// even when its first order is refused.** Before: the refused ack removed the order and the
/// fact was never sent again; test2's gunnar said Free for a paid 10 GB plan for good.
#[test]
fn a_paid_fact_the_sites_main_refused_is_queued_again_until_it_holds_it() {
let f = fixture();
report(&f, &product_report());
f.guardian.store.declare("test2-gunnar-rs", "gunnar").unwrap();
let alica = fact("@test2-gunnar-rs/alica", "gunnar/10gb-1m-fi");
f.lane.queue_entitlement(&alica).unwrap();
let first = entitlement_orders(&f, "test2-gunnar-rs");
assert_eq!(first.len(), 1, "the signup's first order");
// test2's main refuses it (its ack is not ok) and still holds nothing for alica.
let r = Report { acked: vec![OrderAck { seq: first[0].seq, ok: false, detail: "refused".into(), receipt_json: String::new() }], product: Some(serde_json::to_string(&test2_report(None)).unwrap()), ..Default::default() };
f.guardian.store.record_report("test2-gunnar-rs", role::MAIN, NOW - 1_000, &r).unwrap();
assert!(entitlement_orders(&f, "test2-gunnar-rs").is_empty(), "the refused order left the queue");
let said = f.lane.resend_unheld(std::slice::from_ref(&alica));
assert_eq!(said.len(), 1, "{said:?}");
let again = entitlement_orders(&f, "test2-gunnar-rs");
assert_eq!(again.len(), 1, "queued again for test2's main");
let body: EntitlementFact = serde_json::from_str(&again[0].body).unwrap();
assert_eq!(body.tenant.0, "alica", "as test2's gunnar names her");
assert!(f.lane.resend_unheld(std::slice::from_ref(&alica)).is_empty(), "one order waits: none more is queued");
// Once the main acks it and reports the 10 GiB, nothing more is sent.
let r = Report { acked: vec![OrderAck { seq: again[0].seq, ok: true, detail: String::new(), receipt_json: String::new() }], product: Some(serde_json::to_string(&test2_report(Some(10))).unwrap()), ..Default::default() };
f.guardian.store.record_report("test2-gunnar-rs", role::MAIN, NOW - 1_000, &r).unwrap();
assert!(f.lane.resend_unheld(&[alica]).is_empty(), "held: nothing to send");
assert!(entitlement_orders(&f, "test2-gunnar-rs").is_empty());
}
/// ★ ALICA-4: nothing is judged on a silent estate or on a main that reports no promises, and
/// only money that buys room now (Paid, Grace) is sent again.
#[test]
fn a_silent_or_unsaying_main_and_a_lapsed_fact_queue_nothing() {
let f = fixture();
report(&f, &product_report());
f.guardian.store.declare("test2-gunnar-rs", "gunnar").unwrap();
let alica = fact("@test2-gunnar-rs/alica", "gunnar/10gb-1m-fi");
assert!(f.lane.resend_unheld(std::slice::from_ref(&alica)).is_empty(), "never heard: left for later");
report_of(&f, "test2-gunnar-rs", &ProductReport { promised: BTreeMap::new(), measured_unix_ms: NOW - 1_000, ..Default::default() });
assert!(f.lane.resend_unheld(std::slice::from_ref(&alica)).is_empty(), "a main that says no promises: nothing judged");
report_of(&f, "test2-gunnar-rs", &test2_report(None));
let lapsed = EntitlementFact { state: State::Suspended, ..alica.clone() };
assert!(f.lane.resend_unheld(&[lapsed]).is_empty(), "a suspended fact is the ladder's");
f.now.store(NOW + 60 * 60_000, std::sync::atomic::Ordering::SeqCst);
assert!(f.lane.resend_unheld(&[alica]).is_empty(), "an old report: nothing judged");
assert!(entitlement_orders(&f, "test2-gunnar-rs").is_empty());
}
/// ★ RA5-16 (OX-MONHUNT 2026-10-28): **a paid fact the main keeps refusing is sent again ONCE,
/// not on every tick.** Before: a refused resend left the queue, so the next poll tick queued
/// it again with one more ALERT, for ever (a tenant deleted on the site; a main whose promise
/// differs from the cap). Now one resend per fact version: three ticks after the main refused
/// the resend queue nothing more, and a NEW version of the fact (a renewal) is sent once again.
#[test]
fn a_paid_fact_the_main_keeps_refusing_is_resent_once_per_version_not_every_tick() {
let f = fixture();
report(&f, &product_report());
f.guardian.store.declare("test2-gunnar-rs", "gunnar").unwrap();
let alica = fact("@test2-gunnar-rs/alica", "gunnar/10gb-1m-fi");
let refuse_all = |f: &Fixture| {
let acked = entitlement_orders(f, "test2-gunnar-rs").iter().map(|o| OrderAck { seq: o.seq, ok: false, detail: "no such tenant".into(), receipt_json: String::new() }).collect();
let r = Report { acked, product: Some(serde_json::to_string(&test2_report(None)).unwrap()), ..Default::default() };
f.guardian.store.record_report("test2-gunnar-rs", role::MAIN, NOW - 1_000, &r).unwrap();
};
refuse_all(&f);
assert_eq!(f.lane.resend_unheld(std::slice::from_ref(&alica)).len(), 1, "the first resend");
let mut sent = entitlement_orders(&f, "test2-gunnar-rs").len();
for tick in 0..3 {
refuse_all(&f);
let said = f.lane.resend_unheld(std::slice::from_ref(&alica));
assert!(said.is_empty(), "tick {tick}: the refused fact was queued again: {said:?}");
sent += entitlement_orders(&f, "test2-gunnar-rs").len();
}
assert_eq!(sent, 1, "one order in all for one version of the fact");
// A renewal is a new version of the fact: sent once more.
let renewed = EntitlementFact { issued_unix_ms: Some(NOW + 1), paid_until_unix_ms: Some(NOW + 60 * 86_400_000), ..alica.clone() };
assert_eq!(f.lane.resend_unheld(std::slice::from_ref(&renewed)).len(), 1, "a new version is sent once");
refuse_all(&f);
assert!(f.lane.resend_unheld(&[renewed]).is_empty(), "and only once");
}