crates/monetize-product-wasm/src/product.rs
//! **`GuestProduct`** — one `Box<dyn Product>` over a loaded guest. Every
//! verb is one `plugin_call(op, JSON)` in a fresh bounded store with exactly
//! the granted doors defined.
use std::sync::Arc;
use guest_host::module::{Instantiated, LoadedModule};
use guest_host::wasmtime::{self, Linker};
use monetize_product::guardian::{EstateGuardian, EstateView, Finding, Remedy, Step, StepState};
use monetize_product::{Meter, ProductBuild, ProductError};
use monetize_product_guest::wire::{self, LifecycleFacts, WireError};
use monetize_product_guest::{imports, ops, Answer, Manifest};
use crate::doors::{self, CallState, Doors};
use crate::load::{decode_answer, leak_meters, LoadedGuest};
/// A guest, as the server holds it.
pub struct GuestProduct {
module: LoadedModule,
doors: Arc<Doors>,
manifest: Manifest,
abi: u32,
id: &'static str,
meters: &'static [Meter],
/// **The guest's facts (op 90), asked once.** Pure, so one answer per module: what ops 40,
/// 54 and 60 used to answer (`tickets`, `test_actions`, `guardian`) is read from it.
/// `Some(None)`: the guest brings no lifecycle. `None` while the guest could not ANSWER (a
/// trap, a fuel bound, an answer that does not decode): never cached, asked again on the
/// next look (F-G-05, A17).
facts: std::sync::Mutex<Option<Option<LifecycleFacts>>>,
/// The block's `appliance_key`, which the HOST checks actor tickets under (never the guest's).
appliance_key: Option<String>,
}
impl std::fmt::Debug for GuestProduct {
fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
f.debug_struct("GuestProduct")
.field("id", &self.id)
.field("module", &self.module.path())
.field("digest", &self.module.digest_hex())
.field("signer", &self.module.signer())
.field("abi", &self.abi)
.field("grants", &self.manifest.grants)
.finish()
}
}
/// A linker with NO doors, for reading `plugin_abi` and `plugin_manifest`.
/// A module that imports anything cannot instantiate here — which is fine,
/// because those two exports must not need a door.
pub(crate) fn probe_linker(module: &LoadedModule) -> Linker<CallState> {
let mut l = Linker::new(module.engine());
// A module that imports the two grant-free doors still has to instantiate
// for the probe; they are always defined.
let _ = doors::register(&mut l, &[imports::CLOCK, imports::LOG, imports::SLEEP]);
// Any granted door the module imports is defined as a stub that faults —
// the manifest read never reaches a socket, but the module instantiates.
for name in module.imports_from(imports::MODULE) {
if name == imports::CLOCK || name == imports::LOG || name == imports::SLEEP {
continue;
}
let msg = format!("`{name}` is not available while the manifest is read");
// Every door left takes `(ptr, len)` (the one that took none, `host_identity`, went with
// the `sign` grant in T13; an import outside the vocabulary is refused before this).
let _ = l.func_wrap(
imports::MODULE,
name.as_str(),
move |mut c: wasmtime::Caller<'_, CallState>, _p: u32, _l: u32| -> u64 {
guest_host::module::fault(&mut c, msg.clone())
},
);
}
l
}
/// A doorless Doors that cannot open (its runtime's thread refused) is the
/// load's refusal, said by name, never a panic in the loader (R5a).
pub(crate) fn probe_state(module: &LoadedModule) -> Result<CallState, String> {
static NO_DOORS: std::sync::OnceLock<Arc<Doors>> = std::sync::OnceLock::new();
let doors = match NO_DOORS.get() {
Some(d) => d.clone(),
None => {
let opened = Doors::open(
"probe",
"127.0.0.1:1",
None,
None,
None,
&[],
&[],
&[],
std::time::Duration::from_secs(1),
)
.map_err(|e| format!("the probe's doorless doors: {e}"))?;
NO_DOORS.get_or_init(|| Arc::new(opened)).clone()
}
};
Ok(CallState::new(doors, module.limits().limiter()))
}
impl GuestProduct {
/// The whole load order, then the doors, then `id` and `meters`.
pub fn build(b: &ProductBuild<'_>) -> Result<GuestProduct, String> {
GuestProduct::build_after(b, None)
}
/// [`GuestProduct::build`] for a REPLACEMENT of `previous`: the same load
/// order and every gate, and the host's memory for the product
/// ([`Doors::state`]) carried over — shared, not copied, so a call still
/// finishing on `previous` writes where the new guest reads.
pub fn build_after(
b: &ProductBuild<'_>,
previous: Option<&GuestProduct>,
) -> Result<GuestProduct, String> {
let loaded = LoadedGuest::load(b)?;
let call_timeout = loaded
.module
.limits()
.call_timeout
.min(crate::load::PRODUCT_CALL_TIMEOUT);
// A product guest dials nothing (T13, A28 M3): `load::kind_doors` refused `grpc` and
// `sign` above, so no gRPC door opens and no TLS material or other control plane is
// handed to the doors. `endpoint` is what the test kind's typed door answers.
let doors = Doors::open_with(doors::DoorSpec {
product: b.id,
endpoint: b.endpoint,
bearer: b.bearer_token,
twin: None,
tls_ca: None,
tls_client: None,
grants: &loaded.spec.grants,
http_hosts: &loaded.spec.http_hosts,
grpc_hosts: &[],
call_timeout: call_timeout / 2,
http_auth: &loaded.spec.http_auth,
hmac_secrets: &loaded.spec.hmac_secrets,
secrets_dir: loaded.spec.secrets_dir.as_deref(),
http_resolve: &[],
})
.map_err(|e| b.err(e))?;
let mut doors = doors;
if let Some(old) = previous {
doors.state = Arc::clone(&old.doors.state);
}
let mut p = GuestProduct {
module: loaded.module,
doors: Arc::new(doors),
manifest: loaded.manifest,
abi: loaded.abi,
id: "",
meters: &[],
facts: std::sync::Mutex::new(None),
appliance_key: b.appliance_key.map(str::to_string),
};
// 7. The manifest (op 1), once: the id and the meter table.
// G15: a block that names its plugin serves it under the block's id.
let answer: serde_json::Value = p.call(ops::MANIFEST, &()).map_err(|e| b.err(e))?;
let manifest = manifest_answered(answer).map_err(|e| b.err(e))?;
let expected = b.guest.and_then(|g| g.plugin.as_deref()).unwrap_or(b.id);
if manifest.product != expected {
return Err(b.err(format!("the guest's `manifest` op answers `{}`, not `{expected}` — set `plugin = \"{}\"` to serve it as `{}`", manifest.product, manifest.product, b.id)));
}
p.id = crate::leak_id(b.id);
p.meters = leak_meters(manifest.meters);
// The facts, once. A guest that cannot answer now is asked again on the next look.
let _ = p.facts_now();
Ok(p)
}
/// The meter table as it was leaked at load — `'static`, so a caller that
/// read it keeps a valid table after this guest is replaced and dropped.
pub fn meter_table(&self) -> &'static [Meter] {
self.meters
}
pub fn manifest(&self) -> &Manifest {
&self.manifest
}
pub fn abi(&self) -> u32 {
self.abi
}
pub fn digest_hex(&self) -> &str {
self.module.digest_hex()
}
/// **The block's `appliance_key`**: the public key the product's main signs actor tickets
/// with. The host checks tickets under it (monetize-server `site_ticket`, v2.0 P2); it never
/// crosses the ABI.
pub fn appliance_key(&self) -> Option<&str> {
self.appliance_key.as_deref()
}
/// **Spend an actor ticket's `nonce` once** in this product's host memory: `Ok(true)` the
/// first time, `Ok(false)` ever after. The key is the one the Gunnar guest wrote through the
/// `state` door (`ticket.nonce.<nonce>`), so a ticket a guest honoured stays spent for the
/// host, and a live replacement keeps the ledger.
pub fn spend_ticket_nonce(&self, nonce: &str) -> Result<bool, String> {
self.doors.state_admit(&format!("ticket.nonce.{nonce}"))
}
pub fn signer(&self) -> Option<&str> {
self.module.signer()
}
/// Exactly the doors defined on every linker this guest makes.
pub fn linker_defines(&self) -> &[&'static str] {
&self.doors.granted_imports
}
pub fn last_call_fuel(&self) -> u64 {
self.module.last_call_fuel()
}
pub fn peak_call_fuel(&self) -> u64 {
self.module.peak_call_fuel()
}
/// The id to name in a line before `id` is settled at load (the module's path then).
fn id_for_words(&self) -> &str {
if self.id.is_empty() {
self.module.path().to_str().unwrap_or("?")
} else {
self.id
}
}
/// **The estate lifecycle this guest brings** (`lifecycle.facts`, owner 2026-10-17), asked
/// once per module (the op is pure): `Ok(None)` when it brings none (it declares no
/// lifecycle). A failure to answer is `Err` and is asked again next time.
pub fn lifecycle_facts(&self) -> Result<Option<LifecycleFacts>, String> {
let mut held = self.facts.lock().unwrap_or_else(|p| p.into_inner());
if let Some(f) = held.as_ref() {
return Ok(f.clone());
}
let read = match self.call::<LifecycleFacts, ()>(ops::LIFECYCLE_FACTS, &()) {
Ok(f) => Some(f),
Err(e)
if e.contains(ops::UNKNOWN_OP_PHRASE)
|| e.contains("declares no `lifecycle` lane") =>
{
None
}
Err(e) => return Err(e),
};
*held = Some(read.clone());
Ok(read)
}
/// **The guest's own estate guardian** (lane W3), when its facts declare one: a box that
/// links no product (the ONE box) repairs this product's estates with the rules it ships.
pub fn guardian(&self) -> Option<&dyn EstateGuardian> {
self.facts_now()
.is_some_and(|f| f.guardian)
.then_some(self as &dyn EstateGuardian)
}
/// The facts as the merged ops read them: none while the guest could not answer (said once
/// per look, never cached as none).
fn facts_now(&self) -> Option<LifecycleFacts> {
match self.lifecycle_facts() {
Ok(f) => f,
Err(e) => {
eprintln!("[products.{}] ALERT product-guest.facts-unread its guest's facts could not be read ({e}); its tickets, test buttons and guardian are off until the guest answers, asked again on the next look", self.id_for_words());
None
}
}
}
/// **The plan an estate of this product is laid as** (`lifecycle.lay_plan`), as JSON. The
/// guest's refusal keeps its kind (`Refused`: the guest lays no plan of its own); a host-side
/// failure is `Unavailable`.
pub fn lifecycle_lay_plan(
&self,
req: &monetize_product_guest::wire::LayPlanReq,
) -> Result<serde_json::Value, ProductError> {
self.verb(ops::LIFECYCLE_LAY_PLAN, req)
}
/// **One op by its number, JSON in and JSON out** — the conformance table's call (monetize
/// v2.0 P1, A25 §2): which ops a signed guest answers itself, which reach its control plane,
/// and which it leaves to the SDK's default. The server never calls this; it calls the typed
/// verbs.
pub fn call_json(
&self,
op: u32,
req: &serde_json::Value,
) -> Result<serde_json::Value, ProductError> {
self.verb(op, req)
}
fn instantiate(&self) -> Result<Instantiated<CallState>, String> {
let mut linker = Linker::new(self.module.engine());
doors::register(&mut linker, &self.doors.granted_imports)?;
let state = CallState::new(Arc::clone(&self.doors), self.module.limits().limiter());
self.module
.instantiate(state, &linker)
.map_err(|e| e.to_string())
}
/// One op: encode, call under the bounds, decode the envelope. The `Err`
/// string is a HOST-side failure (a bound, a trap, a door fault, a shape),
/// already naming the module and the op; a product-side refusal comes back
/// inside the `Answer`.
fn call<T: serde::de::DeserializeOwned, R: serde::Serialize>(
&self,
op: u32,
req: &R,
) -> Result<T, String> {
let mut inst = self.instantiate()?;
let raw = self.module.call_op(&mut inst, op, &wire::encode(req));
self.sweep(inst.store.data().call);
let raw = raw.map_err(|e| format!("op {} ({}): {e}", op, ops::name(op).unwrap_or("?")))?;
let answer: Answer<T> = decode_answer(op, &raw)?;
answer.map_err(|e| e.to_string())
}
/// `call`, with the host-side failure as `Unavailable` (nothing was
/// written; the next tick may succeed) and the guest's own refusal kept
/// as the kind it was.
fn verb<T: serde::de::DeserializeOwned, R: serde::Serialize>(
&self,
op: u32,
req: &R,
) -> Result<T, ProductError> {
let mut inst = self.instantiate().map_err(|e| self.unavailable(op, e))?;
let raw = self.module.call_op(&mut inst, op, &wire::encode(req));
self.sweep(inst.store.data().call);
let raw = raw.map_err(|e| self.unavailable(op, e.to_string()))?;
let answer: Answer<T> = decode_answer(op, &raw).map_err(|e| self.unavailable(op, e))?;
answer.map_err(ProductError::from)
}
/// **Whatever the guest left open when its call ended.** A guest is a
/// fresh instance per call, so a stream that survived one would be a
/// stream nothing could ever close — and a peer left waiting on a request
/// half that never ends. Swept after EVERY op, including the ones that
/// trapped, because a trap is exactly when a guest does not get to tidy up.
/// ONLY this call's (round 3): the calls of one product guest are not
/// serialized, and a restore's streams must outlive the status read that
/// ends beside it.
fn sweep(&self, call: u64) {
let left = self.doors.sweep_streams_of(call);
if left > 0 {
eprintln!(
"[products.{}] guest {}: {left} stream(s) were still open when the call ended; each was cancelled, \
and nothing pushed through them is reported as delivered",
self.id,
self.module.path().display()
);
}
}
fn unavailable(&self, op: u32, e: String) -> ProductError {
ProductError::Unavailable(format!(
"[products.{}] guest {}: op {} ({}): {e}",
self.id,
self.module.path().display(),
op,
ops::name(op).unwrap_or("?")
))
}
}
/// **What the host reads of a guest product: its facts.** Every estate verb is answered by the
/// served product (`monetize_product::pulled::Pulled`) from the estate's reports.
impl GuestProduct {
/// The manifest's id and meters (op 1, read at load) and the facts (op 90, asked once):
/// `tickets` (the product's main mints actor tickets), `test_actions` and `guardian` (the
/// guest's own estate guardian, lane W3: a box that links no product repairs this product's
/// estates with the rules the product ships). A guest that could not answer is `unread`, and
/// the host asks again on the next look.
pub fn facts(&self) -> monetize_product::ProductFacts {
let mut facts = monetize_product::ProductFacts::new(self.id, self.meters);
match self.lifecycle_facts() {
Ok(Some(f)) => {
facts.tickets = f.tickets;
facts.test_actions = f.test_actions;
facts.guardian = f.guardian;
}
Ok(None) => {}
Err(e) => {
eprintln!("[products.{}] ALERT product-guest.facts-unread its guest's facts could not be read ({e}); its tickets, test buttons and guardian are off until the guest answers, asked again on the next look", self.id_for_words());
facts.unread = Some(e);
}
}
facts
}
}
// ── the estate guardian: pure ops over the estate as its boxes reported ───────────────────
//
// Each verb is ONE `plugin_call` that opens no door. A host-side failure (a bound, a trap)
// must never become a repair: `diagnose` then finds nothing this tick (the next tick asks
// again), `plan` refuses by name, and `observe` keeps waiting (the step's own deadline still
// runs).
impl EstateGuardian for GuestProduct {
fn diagnose(&self, view: &EstateView) -> Vec<Finding> {
match self.verb::<Vec<Finding>, _>(ops::GUARDIAN_DIAGNOSE, view) {
Ok(f) => f,
Err(e) => {
eprintln!(
"[products.{}] guardian.diagnose of {}: {e}; nothing found this tick",
self.id, view.instance
);
Vec::new()
}
}
}
fn plan(&self, remedy: Remedy, view: &EstateView) -> Result<Vec<Step>, String> {
self.verb::<Result<Vec<Step>, String>, _>(
ops::GUARDIAN_PLAN,
&wire::GuardianPlanReq {
remedy,
view: view.clone(),
},
)
.map_err(|e| e.to_string())?
}
fn observe(&self, role: &str, what: &str, view: &EstateView) -> StepState {
match self.verb::<StepState, _>(
ops::GUARDIAN_OBSERVE,
&wire::GuardianObserveReq {
role: role.into(),
what: what.into(),
view: view.clone(),
},
) {
Ok(s) => s,
Err(e) => StepState::Waiting(format!("the {} guest could not say: {e}", self.id)),
}
}
}
/// A guest's refusal crosses back as the product's own error, at compile time.
const _: fn(WireError) -> ProductError = <ProductError as From<WireError>>::from;
/// **Op 1's answer, as the manifest.** A guest built against monetize-product-guest 0.1 answers
/// op 1 with its bare id: it speaks the 28-op contract this host no longer calls, refused by
/// name.
fn manifest_answered(answer: serde_json::Value) -> Result<Manifest, String> {
if let Some(id) = answer.as_str() {
return Err(format!(
"its guest (`{id}`) was built against monetize-product-guest 0.1 (op 1 answers an id, not the manifest): it speaks the \
28-op contract this host no longer calls. Rebuild it against 0.2 (the six ops) and sign it again"
));
}
serde_json::from_value(answer)
.map_err(|e| format!("op 1 (manifest): the guest's manifest does not decode: {e}"))
}
#[cfg(test)]
mod manifest_tests {
use super::*;
/// ★ An older guest (op 1 answers its id) is refused by name; a 0.2 guest's manifest brings
/// its meter table.
#[test]
fn an_older_guest_is_refused_by_name_and_a_manifest_brings_its_meters() {
let e = manifest_answered(serde_json::json!("gunnar")).unwrap_err();
assert!(
e.contains("`gunnar`") && e.contains("0.1") && e.contains("Rebuild"),
"{e}"
);
let m =
Manifest::new("gunnar", &[]).with_meters(vec![monetize_product_guest::WireMeter::new(
"repos",
"repositories",
monetize_product::Enforcement::Measured,
)]);
let back = manifest_answered(serde_json::to_value(&m).unwrap()).unwrap();
assert_eq!(back.meters.len(), 1);
assert!(manifest_answered(serde_json::json!({"nope": 1}))
.unwrap_err()
.contains("does not decode"));
}
}