crates/monetize-server/src/estate_order/tests/guardian.rs
//! **The estate guardian, end to end on the mock** (5.4): a sold estate laid and installed by
//! the shop's one path, its boxes on the pull channel over real gRPC, MockCloud as the provider,
//! the fake kit as the installer, and gunnar's guardian. Monetize never calls into the estate:
//! it queues SIGNED orders the boxes pull, verify under `monetize_pubkey` and run.
use monetize_embed::fleet::kind;
use monetize_embed::testing::pulling::PullingBox;
use monetize_proto as pb;
use tonic::metadata::MetadataValue;
use super::*;
use crate::services::fleet::order_of;
use crate::testing::TestServer;
const TOKEN: &str = "svc-token";
const MAIN_AT: &str = "10.6.0.10:50051";
const TWIN_AT: &str = "10.6.0.12:50051";
fn b64(k: [u8; 32]) -> String {
monetize_embed::signing::base64_encode(&k)
}
/// A gunnar tenant on the ONE ledger: the product's own (`saas-team`) or, with the estate in
/// the path (`@acme/bob`), a sold estate's.
fn saas_fact(state: &AppState, tenant: &str, at: u64) -> monetize_product::EntitlementFact {
let f = monetize_product::EntitlementFact {
tenant: monetize_product::TenantId(tenant.into()),
plan: format!("gunnar/{tenant}/2026-10-10+pack"),
state: monetize_product::State::Paid,
paid_until_unix_ms: Some(at + 30 * 86_400_000),
caps: [("pack_quota_bytes".to_string(), 10u64 << 30)]
.into_iter()
.collect(),
source: "payment:stripe:pi_saas".into(),
signature: Vec::new(),
issued_unix_ms: None,
issued_signature: Vec::new(),
};
let f = state.signer_for("gunnar").sign_fact_at(f, at);
state.ledger.set_entitlement("gunnar", &f).unwrap();
f
}
/// The estate's two gunnar boxes, a delivered order, and the shop's doors served.
struct Field {
r: Rig,
rt: tokio::runtime::Runtime,
srv: Option<TestServer>,
main: PullingBox,
twin: PullingBox,
}
impl Field {
/// `acme`, delivered: the kit's verdict names the twin's fleet key; the main pins its own
/// on its first call inside the install's window.
fn new(name: &str) -> Field {
let r = rig_with(name, cfg(), "PASS");
r.state
.fleet
.link("gunnar", &crate::testing::guardian::GUARDIAN);
let pk = r
.state
.signer_for(r.state.sold_product().unwrap())
.public_key();
let main = PullingBox::new("acme", "main", [1; 32], pk, MAIN_AT);
let twin = PullingBox::new("acme", "twin", [2; 32], pk, TWIN_AT);
r.kit
.appliance_keys
.lock()
.unwrap()
.insert("twin".into(), b64(twin.public_key()));
start_for(
&r.state,
"acme",
"buyer@example.com",
"",
"alice",
&EstateSizes::default(),
)
.unwrap();
r.pay.authorize(&handle_of(&r, "acme"));
drive(&r.state, "acme");
assert_eq!(r.state.estates.get("acme").unwrap().phase, phase::CAPTURED);
let rt = tokio::runtime::Builder::new_multi_thread()
.worker_threads(2)
.enable_all()
.build()
.unwrap();
let srv = rt.block_on(TestServer::spawn(r.state.clone(), TOKEN));
let mut f = Field {
r,
rt,
srv: Some(srv),
main,
twin,
};
f.report_main_pinning();
f.report("twin");
f
}
fn now(&self) -> u64 {
self.r.clock.load(Ordering::SeqCst)
}
fn later(&self, ms: u64) {
self.r.clock.fetch_add(ms, Ordering::SeqCst);
}
fn endpoint(&self) -> String {
self.srv.as_ref().unwrap().endpoint()
}
fn the(&mut self, role: &str) -> &mut PullingBox {
if role == "main" {
&mut self.main
} else {
&mut self.twin
}
}
fn send(
&self,
c: (monetize_embed::fleet::FleetCall, Vec<u8>),
key: Option<[u8; 32]>,
which: &str,
) -> Result<Vec<pb::FleetOrder>, tonic::Status> {
let (c, sig) = c;
let s = pb::FleetSigned {
product: c.product,
instance: c.instance,
role: c.role,
call: c.call,
body: c.body,
at_unix_ms: c.at_unix_ms,
signature: sig,
box_key: key.map(|k| k.to_vec()).unwrap_or_default(),
};
let ep = self.endpoint();
let which = which.to_string();
self.rt.block_on(async move {
let mut f = pb::fleet_client::FleetClient::connect(ep).await.unwrap();
match which.as_str() {
"report" => f.report(s).await.map(|_| Vec::new()),
_ => f.orders(s).await.map(|a| a.into_inner().orders),
}
})
}
fn report_main_pinning(&mut self) {
let now = self.now();
let key = self.main.public_key();
let c = self.main.report(now);
self.send(c, Some(key), "report")
.expect("the main pins its key on its first call");
}
/// The box in `role` reports (its acks with it).
fn report(&mut self, role: &str) {
let now = self.now();
let c = self.the(role).report(now);
self.send(c, None, "report")
.unwrap_or_else(|e| panic!("the {role} was not heard: {e}"));
}
/// The box in `role` pulls its orders (on its heartbeat) and runs them; then reports the acks.
fn pull_and_run(&mut self, role: &str) -> Vec<String> {
let now = self.now();
let c = self.the(role).orders(0, now);
let orders: Vec<_> = self
.send(c, None, "orders")
.unwrap()
.iter()
.map(order_of)
.collect();
let ran = self.the(role).run(&orders, now);
self.report(role);
ran
}
fn tick(&self) -> Vec<String> {
crate::fleet::tick(&self.r.state)
}
fn console(&self) -> pb::guardian_client::GuardianClient<tonic::transport::Channel> {
let ep = self.endpoint();
self.rt.block_on(async move {
pb::guardian_client::GuardianClient::connect(ep)
.await
.unwrap()
})
}
fn press(&self, remedy: &str) -> pb::GuardianCase {
let mut c = self.console();
let mut req = tonic::Request::new(pb::GuardianPressRequest {
instance: "acme".into(),
remedy: remedy.into(),
case_id: String::new(),
principal: "rickard".into(),
strong: true,
});
req.metadata_mut().insert(
"authorization",
MetadataValue::try_from(format!("Bearer {TOKEN}")).unwrap(),
);
self.rt
.block_on(c.press(req))
.expect("the button")
.into_inner()
}
fn estate(&self) -> pb::GuardianEstate {
let mut c = self.console();
let mut req = tonic::Request::new(pb::GuardianCasesRequest {
instance: "acme".into(),
});
req.metadata_mut().insert(
"authorization",
MetadataValue::try_from(format!("Bearer {TOKEN}")).unwrap(),
);
self.rt
.block_on(c.cases(req))
.unwrap()
.into_inner()
.estates
.remove(0)
}
fn case(&self) -> pb::GuardianCase {
self.estate().cases.remove(0)
}
fn disks_of(&self, role: &str) -> Vec<String> {
self.r
.state
.estates
.get("acme")
.unwrap()
.laid
.unwrap()
.machine(role)
.unwrap()
.disks
.iter()
.map(|d| d.id.clone())
.collect()
}
fn server_of(&self, role: &str) -> String {
self.r
.state
.estates
.get("acme")
.unwrap()
.laid
.unwrap()
.machine(role)
.unwrap()
.server
.clone()
}
}
impl Drop for Field {
fn drop(&mut self) {
if let Some(s) = self.srv.take() {
self.rt.block_on(s.stop());
}
}
}
/// ★ **The twin dies → the case → the button → a new box laid by the shop's one path → the
/// installer lays it → it calls home → the primary pulls "adopt twin at X" → done.**
#[test]
fn the_twin_dies_a_human_presses_new_twin_a_new_box_is_laid_and_the_primary_pulls_adopt() {
let mut f = Field::new("guardian-new-twin");
assert!(f.tick().is_empty(), "a healthy estate opens no case");
let old_twin = f.server_of("twin");
let old_disks = f.disks_of("twin");
let lays_before =
f.r.cloud
.calls()
.iter()
.filter(|c| matches!(c, monetize::testing::CloudCall::Lay(_)))
.count();
// The twin stops reporting; the main says so too.
f.later(120_000);
f.main.set("twin", "silent");
f.report("main");
f.tick();
f.tick();
let case = f.case();
assert_eq!(
(case.finding_code.as_str(), case.state.as_str()),
("twin_lost", "investigating"),
"{case:?}"
);
assert!(
case.buttons.contains(&"new_twin".to_string())
&& case.buttons.contains(&"review_twin".to_string())
);
assert!(
case.log.iter().any(|l| l.words.contains("provider")),
"the provider was asked about the twin's box: {:?}",
case.log
);
f.tick();
assert_eq!(
f.case().state,
"investigating",
"auto-fix is OFF: nothing moves without a human"
);
// The button. The new box gets no named key: it pins its own inside the case's window.
f.r.kit.appliance_keys.lock().unwrap().clear();
let pressed = f.press("new_twin");
assert_eq!(
(pressed.approved_by.as_str(), pressed.remedy.as_str()),
("rickard", "new_twin")
);
f.tick();
let c = f.case();
assert_eq!(c.state, "running", "{c:?}");
assert!(
f.r.cloud.destroys().iter().any(|d| d.id == old_twin),
"the old twin was fenced at the provider"
);
let destroyed: Vec<String> = f.r.cloud.destroys().into_iter().map(|d| d.id).collect();
assert!(
!old_disks.is_empty() && old_disks.iter().all(|d| destroyed.contains(d)),
"the fence gave the old twin's disks back too: {old_disks:?} not all in {destroyed:?}"
);
let lays_after =
f.r.cloud
.calls()
.iter()
.filter(|c| matches!(c, monetize::testing::CloudCall::Lay(_)))
.count();
assert_eq!(
lays_after,
lays_before + 1,
"one lay, through the shop's one lay path"
);
assert_eq!(
f.r.kit.roles.lock().unwrap().last().map(String::as_str),
Some("twin"),
"the installer laid the one role"
);
let new_twin = f.server_of("twin");
assert_ne!(new_twin, old_twin, "the run keeps the new box");
assert!(
f.disks_of("twin").iter().all(|d| !old_disks.contains(d)),
"the new twin adopted none of the old twin's disks"
);
assert!(
!f.r.cloud
.destroys()
.iter()
.any(|d| d.id == f.server_of("main")),
"nothing touched the main"
);
// The new box boots, the installer lays it, and it calls home (its own key, pinned on its
// first call), and is heard.
f.later(60_000);
let pk =
f.r.state
.signer_for(f.r.state.sold_product().unwrap())
.public_key();
f.twin = PullingBox::new("acme", "twin", [3; 32], pk, "10.6.0.13:50051");
let now = f.now();
let key = f.twin.public_key();
let c = f.twin.report(now);
f.send(c, Some(key), "report")
.expect("the new twin pins its key inside the window");
f.later(1_000);
f.tick();
// The primary pulls "adopt twin at X", verifies it under monetize_pubkey and runs it.
assert_eq!(f.pull_and_run("main"), [kind::ADOPT_TWIN]);
assert_eq!(f.main.word("twin"), "paired");
f.later(1_000);
f.report("twin");
f.tick();
let c = f.case();
assert_eq!(c.state, "done", "{c:?}");
assert!(c.steps.iter().all(|s| s.state == "met"), "{:?}", c.steps);
assert!(
c.receipt_json.contains("rickard") && c.receipt_json.contains("10.6.0.13:50051"),
"the receipt says who pressed and which twin was adopted"
);
assert!(
f.main.ran.iter().all(|k| !k.starts_with("refused")),
"{:?}",
f.main.ran
);
}
/// ★ **Main data lost → the case → the button → the estate pulls "recharge from twin" → done.**
/// The guardian only queued a signed order; the cloud was not touched.
#[test]
fn main_data_lost_a_human_presses_recharge_and_the_main_pulls_and_runs_it() {
let mut f = Field::new("guardian-recharge");
saas_fact(&f.r.state, "saas-team", f.now());
// G16 (ONE monetize): acme runs no monetize; its tenants are product gunnar's on the ONE
// ledger, the estate in the path.
let bob = saas_fact(
&f.r.state,
&monetize_product::TenantId::on_estate("acme", "bob").0,
f.now(),
);
saas_fact(
&f.r.state,
&monetize_product::TenantId::on_estate("bolt", "eve").0,
f.now(),
);
let calls_before = f.r.cloud.calls().len();
f.later(30_000);
f.main.set("data", "lost").set("serving", "no");
f.report("main");
f.report("twin");
f.tick();
f.tick();
let case = f.case();
assert_eq!(
(case.finding_code.as_str(), case.buttons.as_slice()),
(
"main_data_lost",
["recharge_from_twin".to_string()].as_slice()
),
"{case:?}"
);
f.press("recharge_from_twin");
f.tick();
assert_eq!(
f.pull_and_run("twin"),
Vec::<String>::new(),
"the order is the main's, not the twin's"
);
assert_eq!(
f.pull_and_run("main"),
[kind::RECHARGE_FROM_TWIN],
"the main pulled the signed order and ran it"
);
assert_eq!(
(f.main.word("data"), f.main.word("recharge")),
("ok", "done")
);
f.later(1_000);
f.tick();
assert_eq!(
f.case().state,
"running",
"the money is not back on the main yet"
);
// gap-1: the entitlements are re-stated as a SIGNED order the recharged main pulls (a
// monetize-signed Snapshot it verifies) — never a push into the estate.
assert_eq!(
f.pull_and_run("main"),
[kind::RESTATE_ENTITLEMENTS],
"the main pulled the restatement and ran it"
);
assert_eq!(f.main.restated.len(), 1, "one verified snapshot applied");
// G16 (M-D28 reversed): the restatement carries acme's REAL tenants from the ONE ledger, as
// acme's gunnar names them, verified under the box's monetize key — and nobody else's.
assert_eq!(
f.main.entitled.keys().collect::<Vec<_>>(),
["bob"],
"acme's own tenant only, never the SaaS one or bolt's"
);
assert_eq!(
f.main.entitled["bob"],
f.r.state
.signer_for("gunnar")
.on_estate(&bob, "acme")
.unwrap()
);
monetize::sign::verify_fact(
&f.main.entitled["bob"],
&f.r.state.signer_for("gunnar").verifying_key(),
)
.unwrap();
f.later(1_000);
f.tick();
let c = f.case();
assert_eq!(c.state, "done", "{c:?}");
assert!(
c.receipt_json.contains(TWIN_AT) && c.receipt_json.contains(MAIN_AT),
"the order named the twin and the main as they report"
);
assert!(
c.receipt_json.contains(kind::RESTATE_ENTITLEMENTS),
"the receipt shows the restatement: {}",
c.receipt_json
);
let new_calls: Vec<_> = f.r.cloud.calls().into_iter().skip(calls_before).collect();
assert!(
new_calls.is_empty(),
"a recharge is an order the estate pulls; the cloud is not touched: {new_calls:?}"
);
}
/// ★ **The main is gone and the twin cannot see it → New server**: the old main fenced, a new
/// main laid by the one path, it calls home and FETCHES "recharge from twin" → done.
#[test]
fn the_main_is_lost_a_new_server_is_laid_calls_home_and_pulls_the_recharge() {
let mut f = Field::new("guardian-new-server");
let old_main = f.server_of("main");
let old_disks = f.disks_of("main");
let twin_disks = f.disks_of("twin");
// The main is GONE at the provider (lane G11, the guard's R1: never judged on silence
// alone): nothing of it stands.
{
for id in std::iter::once(old_main.clone()).chain(old_disks.iter().cloned()) {
monetize_cloud::ask::destroy_one(
&f.r.cloud,
&monetize_cloud::ResourceId {
provider: "mockcloud".into(),
id,
},
)
.unwrap();
}
}
f.later(120_000);
f.twin.set("primary", "silent");
f.report("twin");
f.tick();
f.tick();
let case = f.case();
assert_eq!(case.finding_code, "main_lost", "{case:?}");
f.r.kit.appliance_keys.lock().unwrap().clear();
f.press("new_server");
f.tick();
assert!(
f.r.cloud.destroys().iter().any(|d| d.id == old_main),
"the old main fenced"
);
let destroyed: Vec<String> = f.r.cloud.destroys().into_iter().map(|d| d.id).collect();
assert!(
old_disks.iter().all(|d| destroyed.contains(d)),
"the old main's disks given back: {destroyed:?}"
);
assert!(
twin_disks.iter().all(|d| !destroyed.contains(d)),
"the twin, which holds the data, untouched"
);
assert_eq!(
f.r.kit.roles.lock().unwrap().last().map(String::as_str),
Some("main")
);
let pk =
f.r.state
.signer_for(f.r.state.sold_product().unwrap())
.public_key();
f.main = PullingBox::new("acme", "main", [4; 32], pk, "10.6.0.14:50051");
f.main.set("data", "fresh").set("serving", "no");
f.later(60_000);
let now = f.now();
let key = f.main.public_key();
let c = f.main.report(now);
f.send(c, Some(key), "report")
.expect("the new main calls home");
f.later(1_000);
f.tick();
assert_eq!(
f.pull_and_run("main"),
[kind::RECHARGE_FROM_TWIN],
"the new box fetched recharge from twin"
);
f.later(1_000);
f.report("twin");
f.tick();
assert_eq!(f.case().state, "running", "the restatement is owed");
assert_eq!(
f.pull_and_run("main"),
[kind::RESTATE_ENTITLEMENTS],
"the new main pulled its entitlements, signed"
);
assert_eq!(f.main.restated.len(), 1);
f.later(1_000);
f.tick();
let c = f.case();
assert_eq!(c.state, "done", "{c:?}");
assert!(
f.main.ran.iter().all(|k| !k.starts_with("refused")),
"{:?}",
f.main.ran
);
}
/// **Auto-fix ON for this estate**: the guardian approves its own case (and says so), then runs
/// the same steps — still only orders and the cloud.
#[test]
fn with_auto_fix_on_the_guardian_approves_its_own_new_twin_case() {
let mut f = Field::new("guardian-auto");
let g = f.r.state.fleet.guardian().unwrap().clone();
g.set_auto_fix("acme", true, "rickard", true).unwrap();
f.later(120_000);
f.report("main");
f.tick();
f.tick();
f.tick();
let c = f.case();
assert_eq!(
(c.finding_code.as_str(), c.auto, c.approved_by.as_str()),
("twin_lost", true, monetize_guardian::AUTO_PRINCIPAL),
"{c:?}"
);
assert_eq!(
f.r.kit.roles.lock().unwrap().last().map(String::as_str),
Some("twin")
);
}
/// ★ **MONETIZE-V2-REFEED-FLEET: a box the shop lays again on the front's `recharge` is heard.**
/// A lost main (and a lost twin) is laid again by [`sold::recharge_as`] (the front's
/// `Fleet.Ask{recharge}`, the same path for a Gunnar main as for any product's); the new box
/// has a NEW fleet key, named by nobody. The old box's key is forgotten and a pin window opened,
/// as the guardian's own re-lay does: its first report pins the new key, and its order pull and
/// later reports are answered. Before: the old key stayed pinned, and every call of the new box
/// was refused Unauthenticated.
#[test]
fn a_box_laid_again_on_the_fronts_recharge_pins_its_new_key_and_is_heard() {
let mut f = Field::new("guardian-recharge-as");
let pk =
f.r.state
.signer_for(f.r.state.sold_product().unwrap())
.public_key();
for (role, seed, at) in [
("main", [6u8; 32], "10.6.0.16:50051"),
("twin", [7u8; 32], "10.6.0.17:50051"),
] {
let old = f.server_of(role);
f.r.cloud.live.lock().unwrap().remove(&old);
let a = sold::recharge_as(
&f.r.state,
"acme",
role,
"estate = \"acme.gunnar.rs\"\nkind = \"gunnar\"\n",
false,
sold::Caller::Fleet,
)
.unwrap_or_else(|e| panic!("the {role} laid again: {e:?}"));
assert_ne!(a.machine.server, old, "a new {role} server");
let fresh = PullingBox::new("acme", role, seed, pk, at);
let key = fresh.public_key();
if role == "main" {
f.main = fresh
} else {
f.twin = fresh
}
f.later(60_000);
let now = f.now();
let c = f.the(role).report(now);
f.send(c, Some(key), "report")
.unwrap_or_else(|e| panic!("the new {role} pins its new key on its first call: {e}"));
f.later(1_000);
let now = f.now();
let c = f.the(role).orders(0, now);
f.send(c, None, "orders")
.unwrap_or_else(|e| panic!("the new {role}'s order pull is answered: {e}"));
f.report(role);
// The old box's key is gone: a call signed by it is refused.
let mut stale = PullingBox::new(
"acme",
role,
if role == "main" { [1; 32] } else { [2; 32] },
pk,
at,
);
let c = stale.report(f.now());
assert_eq!(
f.send(c, None, "report").unwrap_err().code(),
tonic::Code::Unauthenticated,
"the lost {role}'s old key is forgotten"
);
}
}
/// ★ P4b (lane mon-shared) RED->GREEN: **the guardian's tick is a jera duty** (it was a raw
/// tokio loop whose PANIC was a dropped JoinError): it is in `jobs::running`, its first tick
/// runs at once, and it stops at the stop.
#[test]
fn the_guardian_tick_is_a_jera_duty_that_ticks_at_once_and_stops() {
let r = rig("guardian-tick-duty");
let rt = tokio::runtime::Builder::new_multi_thread()
.worker_threads(2)
.enable_all()
.build()
.unwrap();
let job = rt
.block_on(async { crate::fleet::start(&r.state, std::time::Duration::from_secs(3600)) })
.expect("the tick starts");
assert_eq!(job.kind(), crate::fleet::TICK_JOB);
assert!(
crate::jobs::running().iter().any(|j| Arc::ptr_eq(j, &job)),
"the tick is visible among the running jobs"
);
let until = std::time::Instant::now() + std::time::Duration::from_secs(30);
while job.looks() == 0 && std::time::Instant::now() < until {
std::thread::sleep(std::time::Duration::from_millis(20));
}
assert!(job.looks() >= 1, "the first tick runs at once");
assert_eq!(job.failures(), 0, "{:?}", job.last_failure());
assert!(
job.stop(std::time::Duration::from_secs(10)),
"the tick stops"
);
}