crates/monetize-embed/src/testing/pulling.rs
//! **Test support** (monetize v2.0 P1, A24 §7 step 0): [`PullingBox`], the box on the pull
//! channel modelled, moved here from `monetize-product-gunnar::testing::pulling` so the server's
//! tests keep it when the native product crates go (P2). Behind the `testing` feature; never in
//! a shipped binary.
//!
//! **A gunnar box on the pull channel, modelled** (5.4): what an estate's main or twin does
//! with monetize — report its health, pull the signed orders queued for it, VERIFY each under
//! `monetize_pubkey` (`crate::fleet::verify_order`, the same form gunnar-repair checks)
//! and run the verb, then ack it in its next report. Transport-agnostic: it makes the signed
//! calls and takes the orders; the test carries them (in process, or over gRPC).
//!
//! The verbs are MODELLED here on the box's health words, the way the real ones change them —
//! the real verbs are gunnar-repair's (gunnar `crates/gunnar-repair`), run by the estate.
use std::collections::BTreeMap;
use crate::fleet::{
ask, call, kind, verify_order, Ask, FleetCall, GrowGoAhead, GrowSetAsk, OrderAck, OrdersAfter,
Report, SignedOrder,
};
use crate::{EntitlementFact, Snapshot};
use ed25519_dalek::{Signer as _, SigningKey, VerifyingKey};
/// The product id a gunnar box calls home as.
pub const PRODUCT: &str = "gunnar";
/// One box: its key, what it says of itself, and what it ran.
pub struct PullingBox {
pub instance: String,
pub role: String,
key: SigningKey,
monetize: VerifyingKey,
/// The box's own health words (`serving`, `data`, `serves_at`, `twin`, `primary`, `epoch`, …).
pub health: BTreeMap<String, String>,
/// The highest order seq it executed (replay protection).
pub last_seq: u64,
acks: Vec<OrderAck>,
/// Every order it ran (`kind`), and every one it refused (`refused:<why>`).
pub ran: Vec<String>,
/// A verb the box runs and FAILS (models a broken twin, say).
pub failing: Option<String>,
/// The entitlement facts it holds, by tenant (what `restate_entitlements` applied).
pub entitled: BTreeMap<String, EntitlementFact>,
/// The `issued_unix_ms` of every restatement snapshot it verified and applied.
pub restated: Vec<u64>,
/// The paid data step it verified (`grow_go_ahead`): GB a disk its own data may grow to.
pub grow_to: Option<u64>,
}
impl PullingBox {
/// A box of `instance` in `role` with its fleet key from `seed`, trusting `monetize_pubkey`.
pub fn new(
instance: &str,
role: &str,
seed: [u8; 32],
monetize_pubkey: [u8; 32],
serves_at: &str,
) -> Self {
let mut health = BTreeMap::new();
health.insert(
"serving".into(),
if role == "main" { "yes" } else { "no" }.into(),
);
health.insert("data".into(), "ok".into());
health.insert("serves_at".into(), serves_at.into());
health.insert(
if role == "main" { "twin" } else { "primary" }.into(),
if role == "main" { "paired" } else { "live" }.into(),
);
Self {
instance: instance.into(),
role: role.into(),
key: SigningKey::from_bytes(&seed),
monetize: VerifyingKey::from_bytes(&monetize_pubkey).expect("monetize's public key"),
health,
last_seq: 0,
acks: Vec::new(),
ran: Vec::new(),
failing: None,
entitled: BTreeMap::new(),
restated: Vec::new(),
grow_to: None,
}
}
/// The box's fleet public key (what the installer names, or what it pins on first call).
pub fn public_key(&self) -> [u8; 32] {
self.key.verifying_key().to_bytes()
}
pub fn set(&mut self, word: &str, value: &str) -> &mut Self {
self.health.insert(word.into(), value.into());
self
}
pub fn word(&self, word: &str) -> &str {
self.health.get(word).map(String::as_str).unwrap_or("")
}
/// One signed call: the call and its signature.
pub fn call(&self, which: &str, body: String, now_unix_ms: u64) -> (FleetCall, Vec<u8>) {
let c = FleetCall {
product: PRODUCT.into(),
instance: self.instance.clone(),
role: self.role.clone(),
call: which.into(),
body,
at_unix_ms: now_unix_ms,
};
let sig = self
.key
.sign(&crate::fleet::fleet_call_message(&c))
.to_bytes()
.to_vec();
(c, sig)
}
/// `Fleet.Report`: its health and the acks of the orders it ran since the last one (taken).
pub fn report(&mut self, now_unix_ms: u64) -> (FleetCall, Vec<u8>) {
let r = Report {
build: "gunnar-mock".into(),
health: self.health.clone(),
acked: std::mem::take(&mut self.acks),
twin_review: None,
product: None,
measured_unix_ms: now_unix_ms,
};
self.call(
call::REPORT,
serde_json::to_string(&r).expect("a report"),
now_unix_ms,
)
}
/// The acks a report that was refused must carry again.
pub fn unreport(&mut self, report: &FleetCall) {
if let Ok(r) = serde_json::from_str::<Report>(&report.body) {
let mut back = r.acked;
back.append(&mut self.acks);
self.acks = back;
}
}
/// `Fleet.Orders`: the orders after the last one it ran, held up to `wait_ms`.
pub fn orders(&self, wait_ms: u64, now_unix_ms: u64) -> (FleetCall, Vec<u8>) {
self.call(
call::ORDERS,
serde_json::to_string(&OrdersAfter {
after: self.last_seq,
wait_ms,
})
.expect("orders"),
now_unix_ms,
)
}
/// `Fleet.Ask`.
pub fn ask(&self, remedy: &str, now_unix_ms: u64) -> (FleetCall, Vec<u8>) {
self.call(
call::ASK,
serde_json::to_string(&Ask {
kind: remedy.into(),
body: serde_json::json!({"why": "the box asked"}),
})
.expect("ask"),
now_unix_ms,
)
}
/// `Fleet.Ask{grow_set}`: its own data disks sized at monetize's provider to `gib` a disk
/// (0 = what it is paid to; no `members` = all of its own).
pub fn ask_grow_set(
&self,
members: &[String],
gib: u64,
now_unix_ms: u64,
) -> (FleetCall, Vec<u8>) {
let body = serde_json::to_value(GrowSetAsk {
members: members.to_vec(),
gib,
})
.expect("a grow_set ask");
self.call(
call::ASK,
serde_json::to_string(&Ask {
kind: ask::GROW_SET.into(),
body,
})
.expect("ask"),
now_unix_ms,
)
}
/// **Run the orders it pulled**: each verified under monetize's key (and for this box,
/// newer than the last one), then the verb; the ack waits for the next report. Answers the
/// kinds it ran.
pub fn run(&mut self, orders: &[SignedOrder], now_unix_ms: u64) -> Vec<String> {
let mut ran = Vec::new();
for o in orders {
if let Err(e) = verify_order(
o,
&self.monetize,
(PRODUCT, &self.instance, &self.role),
now_unix_ms,
self.last_seq,
) {
self.ran.push(format!("refused:{e}"));
continue;
}
self.last_seq = o.seq;
let (ok, detail) = self.verb(o, now_unix_ms);
self.ran.push(o.kind.clone());
ran.push(o.kind.clone());
let receipt =
serde_json::json!({"verb": o.kind, "ok": ok, "detail": detail, "at": now_unix_ms})
.to_string();
self.acks.push(OrderAck {
seq: o.seq,
ok,
detail,
receipt_json: receipt,
});
}
ran
}
/// The snapshot verified under monetize's key (envelope and every fact), applied newer-wins
/// per tenant, nothing deleted — and only by a box that serves.
fn restate(&mut self, body: &serde_json::Value) -> (bool, String) {
if self.word("serving") != "yes" {
return (
false,
format!(
"the {} does not serve (serving={}): no entitlements applied",
self.role,
self.word("serving")
),
);
}
let snap: Snapshot = match body.get("snapshot").cloned().map(serde_json::from_value) {
Some(Ok(s)) => s,
Some(Err(e)) => return (false, format!("the snapshot does not parse: {e}")),
None => return (false, "the order carries no snapshot".into()),
};
if let Err(e) = crate::signing::verify_snapshot(&snap, &self.monetize) {
return (
false,
format!("the snapshot does not verify under monetize's key: {e}"),
);
}
let (mut applied, mut held) = (0, 0);
for f in snap.facts {
let newer = self
.entitled
.get(&f.tenant.0)
.is_none_or(|h| f.issued_unix_ms.unwrap_or(0) > h.issued_unix_ms.unwrap_or(0));
if newer {
self.entitled.insert(f.tenant.0.clone(), f);
applied += 1;
} else {
held += 1;
}
}
self.restated.push(snap.issued_unix_ms);
(
true,
format!("restated {applied} entitlement facts ({held} already held as new or newer)"),
)
}
/// The go-ahead verified under monetize's key — for THIS estate, still good — and kept:
/// what its growth (drain, unmount, `grow_set`, remount) may size its own disks to.
fn grow_go_ahead(&mut self, body: &str, now_unix_ms: u64) -> (bool, String) {
let order: GrowGoAhead = match serde_json::from_str(body) {
Ok(o) => o,
Err(e) => return (false, format!("the grow_go_ahead does not parse: {e}")),
};
let g =
match crate::signing::verify_estate_growth(order.go_ahead.as_bytes(), &self.monetize) {
Ok(g) => g,
Err(e) => {
return (
false,
format!("the go-ahead does not verify under monetize's key: {e}"),
)
}
};
if g.estate != self.instance {
return (
false,
format!("the go-ahead is {}'s, not {}'s", g.estate, self.instance),
);
}
if (g.until_unix_ms as u64) < now_unix_ms {
return (false, format!("the go-ahead ({}) ran out", g.reference));
}
self.grow_to = Some(order.data_gib);
(
true,
format!(
"go-ahead {} verified: the {}'s data may grow to {} GB a disk",
g.reference, self.role, order.data_gib
),
)
}
fn verb(&mut self, o: &SignedOrder, now_unix_ms: u64) -> (bool, String) {
if self.failing.as_deref() == Some(o.kind.as_str()) {
if o.kind == kind::RECHARGE_FROM_TWIN {
self.set("recharge", "failed");
}
return (false, format!("{} failed on the {}", o.kind, self.role));
}
let body: serde_json::Value = serde_json::from_str(&o.body).unwrap_or_default();
let s = |k: &str| {
body.get(k)
.map(|v| {
v.as_str()
.map(str::to_string)
.unwrap_or_else(|| v.to_string())
})
.unwrap_or_default()
};
match o.kind.as_str() {
kind::FENCE_PRIMARY => {
self.set("serving", "fenced").set("epoch", &s("epoch"));
(true, format!("fenced at epoch {}", s("epoch")))
}
kind::RECHARGE_FROM_TWIN => {
self.set("data", "ok")
.set("serving", "yes")
.set("recharge", "done")
.set("epoch", &s("epoch"));
(true, format!("recharged from the twin at {}", s("twin")))
}
kind::ADOPT_TWIN => {
self.set("twin", "paired");
(
true,
format!("adopted the twin at {} ({})", s("twin"), s("principal")),
)
}
kind::REVIEW_TWIN => (true, "the twin is current".into()),
kind::RESTATE_ENTITLEMENTS => self.restate(&body),
kind::GROW_GO_AHEAD => self.grow_go_ahead(&o.body, now_unix_ms),
other => (false, format!("{other:?} is no verb this box runs")),
}
}
}