crates/monetize-product-guest/src/wire.rs
//! **The wire**: JSON requests and answers per op, and the frame host doors
//! carry bytes in.
use monetize_product::{Backing, Enforcement, ProductError, Scope};
/// Encode a request or an answer.
pub fn encode<T: serde::Serialize>(v: &T) -> Vec<u8> {
serde_json::to_vec(v).expect("every wire type serializes")
}
/// Decode a request or an answer.
pub fn decode<T: serde::de::DeserializeOwned>(bytes: &[u8]) -> Result<T, String> {
serde_json::from_slice(bytes).map_err(|e| e.to_string())
}
/// What a guest publishes from `plugin_manifest()`.
#[derive(Clone, Debug, PartialEq, Eq, serde::Serialize, serde::Deserialize)]
pub struct Manifest {
pub abi_min: u32,
pub abi_max: u32,
/// The product id — `[products.<id>]` must match it, or the guest is a
/// module for another product and is refused by name.
pub product: String,
/// The grants this guest needs. Gate A: every one must be granted. Gate B:
/// every import the module really has must be unlocked by one of these.
pub grants: Vec<String>,
/// **Frozen: always `false`.** Whether the guest wanted an identity key minted for
/// `host_sign`; the door went in monetize v2.0 P2 T13 (A28 M3), and a host refuses a
/// manifest that says `true`. Kept on the wire so every guest built before reads as itself.
#[serde(default)]
pub identity: bool,
/// **What this guest is**: [`crate::kinds`]. Absent in every guest built
/// before vendors were guests, which is why it defaults to `product`.
#[serde(default = "default_kind")]
pub kind: String,
/// **The meter table** (a product guest's; ops 1 and 2 merged, the 6-op contract): with
/// backings and enforcement. Empty for every other kind, and absent on the wire then.
#[serde(default, skip_serializing_if = "Vec::is_empty")]
pub meters: Vec<WireMeter>,
}
fn default_kind() -> String {
crate::kinds::PRODUCT.to_string()
}
impl Manifest {
/// A manifest for this crate's ABI, asking for `grants`.
pub fn new(product: &str, grants: &[&str]) -> Manifest {
Manifest {
abi_min: crate::abi::ABI_MIN,
abi_max: crate::abi::ABI_MAX,
product: product.to_string(),
grants: grants.iter().map(|s| s.to_string()).collect(),
identity: false,
kind: crate::kinds::PRODUCT.to_string(),
meters: Vec::new(),
}
}
/// This manifest with the product's meter table (what op 1 answers a host with).
pub fn with_meters(self, meters: Vec<WireMeter>) -> Manifest {
Manifest { meters, ..self }
}
/// A manifest of another kind ([`crate::kinds`]); `id` is the block's name.
pub fn of_kind(kind: &str, id: &str, grants: &[&str]) -> Manifest {
Manifest { kind: kind.to_string(), ..Manifest::new(id, grants) }
}
}
/// [`monetize_product::Meter`] with owned strings — the same six facts.
#[derive(Clone, Debug, PartialEq, Eq, serde::Serialize, serde::Deserialize)]
pub struct WireMeter {
pub name: String,
pub meaning: String,
pub enforcement: Enforcement,
pub reports_usage: bool,
pub backing: Option<Backing>,
pub scope: Scope,
}
impl WireMeter {
pub fn new(name: &str, meaning: &str, enforcement: Enforcement) -> WireMeter {
WireMeter { name: name.into(), meaning: meaning.into(), enforcement, reports_usage: true, backing: None, scope: Scope::Tenant }
}
pub fn cap_only(name: &str, meaning: &str, enforcement: Enforcement) -> WireMeter {
WireMeter { reports_usage: false, ..WireMeter::new(name, meaning, enforcement) }
}
pub fn with_backing(self, backing: Backing) -> WireMeter {
WireMeter { backing: Some(backing), ..self }
}
pub fn per_product(self) -> WireMeter {
WireMeter { scope: Scope::Product, ..self }
}
/// From a linked plugin's `&'static` table.
pub fn from_meter(m: &monetize_product::Meter) -> WireMeter {
WireMeter { name: m.name.into(), meaning: m.meaning.into(), enforcement: m.enforcement, reports_usage: m.reports_usage, backing: m.backing, scope: m.scope }
}
}
/// A refusal across the wire. The kind survives: a host turns `Refused` and
/// `Unavailable` back into the same [`ProductError`] arms, and `Plain` into a
/// `String` where the trait answers one.
#[derive(Clone, Debug, PartialEq, Eq, serde::Serialize, serde::Deserialize)]
pub enum WireError {
Refused(String),
Unavailable(String),
Plain(String),
}
impl std::fmt::Display for WireError {
fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
match self {
WireError::Refused(s) => write!(f, "product refused: {s}"),
WireError::Unavailable(s) => write!(f, "product unavailable: {s}"),
WireError::Plain(s) => write!(f, "{s}"),
}
}
}
impl From<ProductError> for WireError {
fn from(e: ProductError) -> Self {
match e {
ProductError::Refused(s) => WireError::Refused(s),
ProductError::Unavailable(s) => WireError::Unavailable(s),
}
}
}
impl From<WireError> for ProductError {
fn from(e: WireError) -> Self {
match e {
WireError::Refused(s) => ProductError::Refused(s),
WireError::Unavailable(s) => ProductError::Unavailable(s),
WireError::Plain(s) => ProductError::Refused(s),
}
}
}
/// Every op's answer envelope.
pub type Answer<T> = Result<T, WireError>;
/// `guardian.plan`: the remedy to plan, over the estate as reported.
#[derive(Clone, Debug, PartialEq, Eq, serde::Serialize, serde::Deserialize)]
pub struct GuardianPlanReq {
pub remedy: monetize_product::guardian::Remedy,
pub view: monetize_product::guardian::EstateView,
}
/// `guardian.observe`: is the `Await` on `role` for `what` met, over the estate as reported?
#[derive(Clone, Debug, PartialEq, Eq, serde::Serialize, serde::Deserialize)]
pub struct GuardianObserveReq {
pub role: String,
pub what: String,
pub view: monetize_product::guardian::EstateView,
}
/// **The answer to `lifecycle.facts` (op 90) is [`crate::facts::Facts`]** (v2.0 P3): one type,
/// what an estate of the product IS — its lifecycle, `tickets`, `test_actions` and `guardian`
/// (ops 40, 54 and 60 merged in, T13), its contract with the shop and its sale proof. The old
/// name stays for one release cycle; every field is defaulted, so an older guest's answer reads
/// as `Facts` with nothing lost.
pub type LifecycleFacts = crate::facts::Facts;
/// The product's words (`gunnar`, `Gunnar`, `gunnar.rs`).
#[derive(Clone, Debug, Default, PartialEq, Eq, serde::Serialize, serde::Deserialize)]
pub struct WireWords {
pub name: String,
pub title: String,
pub account_home: String,
}
/// One machine a data set grows on.
#[derive(Clone, Debug, Default, PartialEq, Eq, serde::Serialize, serde::Deserialize)]
pub struct WireDataRole {
pub role: String,
pub mirrors_primary: bool,
pub set_suffix: String,
}
/// `lifecycle.lay_plan`: what the server's estate config says (as JSON: `shape`, `labels`,
/// `zone`, and the tier of the template each medium kind's system disk is cloned from), the
/// company, the buyer's sizes and the twin's plan.
#[derive(Clone, Debug, Default, PartialEq, serde::Serialize, serde::Deserialize)]
pub struct LayPlanReq {
pub config: serde_json::Value,
pub company: String,
pub sizes: serde_json::Value,
pub twin_plan: String,
/// medium kind → the template tier its system disk is laid in (absent: the shape's tier).
pub template_tiers: std::collections::BTreeMap<String, String>,
}
// ── host doors ────────────────────────────────────────────────────────────────
/// What `host_http` is asked for. The body rides in the frame, not here.
#[derive(Clone, Debug, Default, PartialEq, Eq, serde::Serialize, serde::Deserialize)]
pub struct HttpHead {
pub method: String,
pub url: String,
pub headers: Vec<(String, String)>,
}
/// An HTTP request, as a guest builds one.
#[derive(Clone, Debug, Default, PartialEq, Eq)]
pub struct HttpRequest {
pub method: String,
pub url: String,
pub headers: Vec<(String, String)>,
pub body: Vec<u8>,
}
/// An HTTP answer, as the host returns one.
#[derive(Clone, Debug, Default, PartialEq, Eq)]
pub struct HttpResponse {
pub status: u16,
pub headers: Vec<(String, String)>,
pub body: Vec<u8>,
}
/// What `host_grpc_unary` is asked for. The request bytes ride in the frame.
#[derive(Clone, Debug, Default, PartialEq, Eq, serde::Serialize, serde::Deserialize)]
pub struct GrpcHead {
/// `/package.Service/Method`.
pub path: String,
/// **Metadata the host puts on this one call.** The block's bearer is the
/// host's own and is always added; these are the product's, and the only
/// thing that can be in one is what the guest already holds.
#[serde(default, skip_serializing_if = "Vec::is_empty")]
pub headers: Vec<(String, String)>,
/// **Which box.** `None` is the block's own `endpoint`; anything else must
/// be named in `guest_grpc_hosts`. A restore asks the TWIN for its log
/// head while it fills the primary.
#[serde(default, skip_serializing_if = "Option::is_none")]
pub endpoint: Option<String>,
}
/// What `host_grpc_stream_open` is asked for.
#[derive(Clone, Debug, Default, PartialEq, Eq, serde::Serialize, serde::Deserialize)]
pub struct StreamOpen {
/// `/package.Service/Method`.
pub path: String,
/// Metadata for the call — the credential, for a control plane with no
/// bearer.
#[serde(default, skip_serializing_if = "Vec::is_empty")]
pub headers: Vec<(String, String)>,
/// **What the guest believes the whole stream needs, in seconds.** The
/// host refuses a stream it could not carry to the end. `0` asks for
/// whatever the call's deadline allows, and a stream that then runs out of
/// time is an ERROR at the close, never a quiet success.
pub expect_secs: u32,
/// **Which box.** `None` is the block's own `endpoint`; anything else must
/// be named in `[products.<id>] guest_grpc_hosts`, because a guest that
/// could dial an address nobody wrote down is a guest that could take a
/// tenant's log somewhere else. A restore needs two: the primary it fills
/// and the twin it reads.
#[serde(default, skip_serializing_if = "Option::is_none")]
pub endpoint: Option<String>,
}
/// One open stream, by the handle the host gave out.
#[derive(Clone, Copy, Debug, Default, PartialEq, Eq, serde::Serialize, serde::Deserialize)]
pub struct StreamRef {
pub stream: u64,
}
/// What `host_hmac_verify` is asked: which secret, which algorithm, and the
/// signatures the sender claimed, hex. The message rides in the frame's body.
#[derive(Clone, Debug, Default, PartialEq, Eq, serde::Serialize, serde::Deserialize)]
pub struct HmacHead {
pub secret: String,
/// `sha256` — the only one today.
pub algo: String,
pub candidates_hex: Vec<String>,
}
/// One key in the host's memory for this product.
#[derive(Clone, Debug, Default, PartialEq, Eq, serde::Serialize, serde::Deserialize)]
pub struct StateKey {
pub key: String,
}
/// A gRPC status, as the host reports one: the code's name and its message.
#[derive(Clone, Debug, Default, PartialEq, Eq, serde::Serialize, serde::Deserialize)]
pub struct GrpcError {
pub code: String,
pub message: String,
}
impl GrpcError {
/// **The product said NO**, and asking again would change nothing: a credential or grant
/// it does not accept, a request it calls malformed, a state that does not allow it, a
/// thing it does not have or already has. `Refused` is FINAL — monetize never pushes a
/// refused fact again (the poller's `retry_pushes`) — so it is kept for these codes only.
/// Everything else is asked again next tick: an outage, a timeout, and every transport
/// failure on the way (tonic turns an HTTP/2 stream reset into `Internal`; R5.1-F,
/// 2026-10-07, Gunnar's rule moved here for every product, A17 F-G-03).
pub fn is_refusal(&self) -> bool {
self.is_refusal_but("")
}
/// [`GrpcError::is_refusal`], with one `Unauthenticated` that is not final: the one whose
/// message carries `transient_phrase` (Gunnar's clock-skew words: a skewed clock is about
/// time, not about the key). An empty phrase makes every `Unauthenticated` final.
pub fn is_refusal_but(&self, transient_phrase: &str) -> bool {
match self.code.as_str() {
"PermissionDenied" | "InvalidArgument" | "FailedPrecondition" | "NotFound" | "AlreadyExists" => true,
"Unauthenticated" => transient_phrase.is_empty() || !self.message.contains(transient_phrase),
_ => false,
}
}
/// The box could not take it now and it is asked again: every status that is not
/// [`GrpcError::is_refusal`].
pub fn is_unavailable(&self) -> bool {
!self.is_refusal()
}
/// The [`ProductError`] this status is, naming the endpoint: `Refused` for an explicit NO
/// ([`GrpcError::is_refusal`]), `Unavailable` for the rest.
pub fn to_product_error(&self, endpoint: &str) -> ProductError {
if self.is_unavailable() {
ProductError::Unavailable(format!("{endpoint}: {}", self.message))
} else {
ProductError::Refused(format!("{endpoint} {}: {}", self.code, self.message))
}
}
}
/// The head every host door answers with. `ok = false` carries `error`; the
/// other fields are per door (`status`/`headers` for http, `grpc` for a gRPC
/// status, `value` for env).
#[derive(Clone, Debug, Default, PartialEq, Eq, serde::Serialize, serde::Deserialize)]
pub struct HostHead {
pub ok: bool,
#[serde(default, skip_serializing_if = "Option::is_none")]
pub error: Option<String>,
#[serde(default, skip_serializing_if = "Option::is_none")]
pub status: Option<u16>,
#[serde(default, skip_serializing_if = "Vec::is_empty")]
pub headers: Vec<(String, String)>,
#[serde(default, skip_serializing_if = "Option::is_none")]
pub grpc: Option<GrpcError>,
#[serde(default, skip_serializing_if = "Option::is_none")]
pub value: Option<String>,
/// The handle `host_grpc_stream_open` gave out.
#[serde(default, skip_serializing_if = "Option::is_none")]
pub stream: Option<u64>,
/// The peer FINISHED: there is no further message, and this is the only
/// honest way a server stream stops.
#[serde(default, skip_serializing_if = "std::ops::Not::not")]
pub end: bool,
}
impl HostHead {
pub fn ok() -> HostHead {
HostHead { ok: true, ..Default::default() }
}
pub fn err(e: impl Into<String>) -> HostHead {
HostHead { ok: false, error: Some(e.into()), ..Default::default() }
}
}
/// **The frame**: `[u32 LE head_len][head bytes][body bytes]`. A JSON head and
/// a raw body, so a protobuf or a signature never becomes a JSON array.
pub mod frame {
pub fn encode(head: &[u8], body: &[u8]) -> Vec<u8> {
let mut out = Vec::with_capacity(4 + head.len() + body.len());
out.extend_from_slice(&(head.len() as u32).to_le_bytes());
out.extend_from_slice(head);
out.extend_from_slice(body);
out
}
pub fn decode(bytes: &[u8]) -> Result<(&[u8], &[u8]), String> {
if bytes.len() < 4 {
return Err(format!("frame of {} bytes has no length prefix", bytes.len()));
}
let n = u32::from_le_bytes([bytes[0], bytes[1], bytes[2], bytes[3]]) as usize;
if bytes.len() < 4 + n {
return Err(format!("frame declares a {n}-byte head but holds {}", bytes.len() - 4));
}
Ok((&bytes[4..4 + n], &bytes[4 + n..]))
}
/// A frame whose head is JSON.
pub fn encode_json<T: serde::Serialize>(head: &T, body: &[u8]) -> Vec<u8> {
encode(&super::encode(head), body)
}
/// Decode a frame whose head is JSON.
pub fn decode_json<T: serde::de::DeserializeOwned>(bytes: &[u8]) -> Result<(T, &[u8]), String> {
let (head, body) = decode(bytes)?;
Ok((super::decode(head)?, body))
}
/// Read a host door's answer frame: the body on `ok`, the error otherwise.
pub fn host_answer(bytes: &[u8]) -> Result<(super::HostHead, &[u8]), String> {
let (head, body): (super::HostHead, &[u8]) = decode_json(bytes)?;
if head.ok {
Ok((head, body))
} else {
Err(head.error.unwrap_or_else(|| "the host door refused without saying why".into()))
}
}
}
#[cfg(test)]
mod tests {
use super::*;
/// ★ F-G-03 (A17): only a product's explicit NO is a final refusal. A transport failure on
/// the way (tonic turns an HTTP/2 reset into `Internal`), an abort or a box out of room is
/// asked again: a PAID fact filed `Refused` is never pushed again (R5.1-F, Gunnar's rule).
#[test]
fn only_an_explicit_no_is_refused_and_a_transport_reset_is_asked_again() {
let e = |code: &str| GrpcError { code: code.into(), message: "m".into() };
for code in ["PermissionDenied", "InvalidArgument", "FailedPrecondition", "NotFound", "AlreadyExists", "Unauthenticated"] {
assert!(matches!(e(code).to_product_error("holger"), ProductError::Refused(_)), "{code} is a final no");
}
for code in ["Internal", "Aborted", "ResourceExhausted", "Unavailable", "DeadlineExceeded", "Cancelled", "Unknown", "Unimplemented", "OutOfRange", "DataLoss"] {
assert!(matches!(e(code).to_product_error("holger"), ProductError::Unavailable(_)), "{code} is asked again");
}
let skew = GrpcError { code: "Unauthenticated".into(), message: "90000 ms away from this server's clock".into() };
assert!(skew.is_refusal(), "with no phrase every Unauthenticated is final");
assert!(!skew.is_refusal_but("away from this server's clock"), "the named phrase is asked again");
assert!(e("Unauthenticated").is_refusal_but("away from this server's clock"), "any other Unauthenticated stays final");
}
/// `live_slot` (B4): a guest built before the field answers facts without it; that reads
/// as a live slot (Gunnar's boxes), so an older guest keeps the behaviour it had.
#[test]
fn lifecycle_facts_without_live_slot_read_as_a_live_slot() {
let mut v = serde_json::to_value(LifecycleFacts { live_slot: false, ..LifecycleFacts::default() }).unwrap();
assert_eq!(v["live_slot"], serde_json::json!(false));
v.as_object_mut().unwrap().remove("live_slot");
let f: LifecycleFacts = serde_json::from_value(v).unwrap();
assert!(f.live_slot, "absent means a live slot");
assert!(LifecycleFacts::default().live_slot, "the default is a live slot");
}
#[test]
fn the_frame_round_trips_and_names_a_short_one() {
let f = frame::encode_json(&GrpcHead { path: "/a.B/C".into(), headers: vec![("gunnar-authorization".into(), "gunnar-sshsig-v2 …".into())], endpoint: None }, &[0, 1, 2, 255]);
let (head, body): (GrpcHead, &[u8]) = frame::decode_json(&f).unwrap();
assert_eq!(head.path, "/a.B/C");
assert_eq!(body, &[0, 1, 2, 255]);
assert!(frame::decode(&[1, 0]).unwrap_err().contains("length prefix"));
assert!(frame::decode(&[9, 0, 0, 0, 1]).unwrap_err().contains("declares"));
}
#[test]
fn a_host_answer_carries_the_error_or_the_body() {
let ok = frame::encode_json(&HostHead { value: Some("v".into()), ..HostHead::ok() }, b"sig");
let (head, body) = frame::host_answer(&ok).unwrap();
assert_eq!((head.value.as_deref(), body), (Some("v"), &b"sig"[..]));
let bad = frame::encode_json(&HostHead::err("nope"), &[]);
assert_eq!(frame::host_answer(&bad).unwrap_err(), "nope");
}
#[test]
fn a_wire_error_keeps_its_kind_across_the_product_error_seam() {
let e: WireError = ProductError::Unavailable("x".into()).into();
assert!(matches!(ProductError::from(e.clone()), ProductError::Unavailable(s) if s == "x"));
let json: Answer<()> = decode(&encode(&Answer::<()>::Err(e))).unwrap();
assert!(matches!(json, Err(WireError::Unavailable(_))));
}
}