crates/monetize-estate-side/tests/the_estate_side_rides_the_real_wire.rs
//! **The estate side over the real wire** (P4a): the shared crate's two halves against an
//! in-process ONE monetize speaking monetize-proto's own gRPC services, every call checked with
//! monetize-embed's own verifiers, as the ONE box checks them:
//!
//! * the box pump ([`monetize_estate_side::pump`]): `Fleet.Report` / `Orders` over
//! `grpc::TonicLink`, each call verified under the box's key, the signed order verified on the
//! box and run through the product's `PumpBox`, its ack carried by the next report;
//! * the front ([`monetize_estate_side::sold_door`]): the shared update step (Skidbladnir
//! nordisk-pairing `frontside::update::look`) over `ShopDoor`'s `Fleet.Ask` wire, the shop's
//! turn order twin → main → front, every ask a `FleetCall` of the front the shop verifies under
//! the estate's client key (H16: the old `SoldEstate` door is gone).
use std::collections::BTreeMap;
use std::sync::{Arc, Mutex};
use std::time::Duration;
use ed25519_dalek::Signer as _;
use monetize_embed::fleet::{order_message, FleetCall, OrderAck, OrdersAfter, Report, SignedOrder};
use monetize_estate_side::pump::{
grpc, Addressee, Asker, BoxKey, Ctx, FleetConfig, Puller, PumpBox,
};
use monetize_estate_side::sold_door::{ShopDoor, FEED_SHOP_DOOR};
use monetize_proto as pb;
use nordisk_pairing::frontside::update::{self, Hands};
use nordisk_pairing::live::{Declared, Running, Stage};
const MONETIZE_SEED: [u8; 32] = [7; 32];
const CLIENT_SEED: [u8; 32] = [0x22; 32];
fn now() -> u64 {
monetize_estate_side::now_unix_ms()
}
/// The ONE monetize's `Fleet`, in memory: one signed order queued for `acme/main`, every call
/// verified under the box key it pinned at the first call.
#[derive(Clone, Default)]
struct OneFleet {
pinned: Arc<Mutex<Option<[u8; 32]>>>,
queue: Arc<Mutex<Vec<SignedOrder>>>,
acked: Arc<Mutex<Vec<OrderAck>>>,
calls: Arc<Mutex<Vec<String>>>,
}
impl OneFleet {
fn verify(&self, s: &pb::FleetSigned) -> Result<FleetCall, tonic::Status> {
let mut pin = self.pinned.lock().unwrap();
if pin.is_none() {
*pin = Some(s.box_key.as_slice().try_into().map_err(|_| {
tonic::Status::unauthenticated("the first call carries no box key")
})?);
}
let key = monetize_embed::VerifyingKey::from_bytes(pin.as_ref().unwrap())
.map_err(|e| tonic::Status::unauthenticated(e.to_string()))?;
let c = FleetCall {
product: s.product.clone(),
instance: s.instance.clone(),
role: s.role.clone(),
call: s.call.clone(),
body: s.body.clone(),
at_unix_ms: s.at_unix_ms,
};
monetize_embed::fleet::verify_fleet_call(&c, &s.signature, &key, now())
.map_err(|e| tonic::Status::unauthenticated(format!("{e:?}")))?;
self.calls.lock().unwrap().push(format!(
"{} {}/{}/{}",
c.call, c.product, c.instance, c.role
));
Ok(c)
}
}
#[tonic::async_trait]
impl pb::fleet_server::Fleet for OneFleet {
async fn report(
&self,
r: tonic::Request<pb::FleetSigned>,
) -> Result<tonic::Response<pb::FleetReportAnswer>, tonic::Status> {
let c = self.verify(r.get_ref())?;
let report: Report = serde_json::from_str(&c.body)
.map_err(|e| tonic::Status::invalid_argument(e.to_string()))?;
self.queue
.lock()
.unwrap()
.retain(|o| !report.acked.iter().any(|a| a.seq == o.seq));
self.acked
.lock()
.unwrap()
.extend(report.acked.iter().cloned());
Ok(tonic::Response::new(pb::FleetReportAnswer {
acked: report.acked.iter().map(|a| a.seq).collect(),
next_report_ms: 5_000,
}))
}
async fn orders(
&self,
r: tonic::Request<pb::FleetSigned>,
) -> Result<tonic::Response<pb::FleetOrdersAnswer>, tonic::Status> {
let c = self.verify(r.get_ref())?;
let after: OrdersAfter = serde_json::from_str(&c.body)
.map_err(|e| tonic::Status::invalid_argument(e.to_string()))?;
let orders = self
.queue
.lock()
.unwrap()
.iter()
.filter(|o| o.seq > after.after)
.map(|o| pb::FleetOrder {
seq: o.seq,
product: o.product.clone(),
instance: o.instance.clone(),
role: o.role.clone(),
kind: o.kind.clone(),
body: o.body.clone(),
issued_unix_ms: o.issued_unix_ms,
not_after_unix_ms: o.not_after_unix_ms,
signature: o.signature.clone(),
})
.collect();
Ok(tonic::Response::new(pb::FleetOrdersAnswer { orders }))
}
async fn ask(
&self,
r: tonic::Request<pb::FleetSigned>,
) -> Result<tonic::Response<pb::FleetAskAnswer>, tonic::Status> {
self.verify(r.get_ref())?;
Ok(tonic::Response::new(pb::FleetAskAnswer {
case_id: "c-1".into(),
words: "resized".into(),
..Default::default()
}))
}
}
/// A product box that runs one kind, `note`.
struct Noting;
impl PumpBox for Noting {
fn health(&self) -> BTreeMap<String, String> {
BTreeMap::from([("serving".into(), "yes".into())])
}
fn run(&self, ctx: &Ctx, o: &SignedOrder, _: &mut dyn Asker) -> OrderAck {
let ok = o.kind == "note";
OrderAck {
seq: o.seq,
ok,
detail: if ok {
format!("noted on {}", ctx.me.role)
} else {
format!("no kind {}", o.kind)
},
receipt_json: String::new(),
}
}
}
/// Serve one gRPC service on a loopback port of `rt`.
macro_rules! serve {
($rt:expr, $svc:expr) => {{
let listener = $rt
.block_on(tokio::net::TcpListener::bind("127.0.0.1:0"))
.unwrap();
let addr = listener.local_addr().unwrap();
$rt.spawn(
tonic::transport::Server::builder()
.add_service($svc)
.serve_with_incoming(tokio_stream::wrappers::TcpListenerStream::new(listener)),
);
addr
}};
}
/// ★ The pump over the real `Fleet` wire: the box reports (pinned by its first call), pulls the
/// one signed order, runs it through its `PumpBox`, and its ack reaches monetize in the next
/// report; a second pull runs nothing.
#[test]
fn a_box_pulls_runs_and_acks_one_signed_order_over_the_fleet_wire() {
let rt = tokio::runtime::Builder::new_multi_thread()
.worker_threads(2)
.enable_all()
.build()
.unwrap();
let one = OneFleet::default();
let mut o = SignedOrder {
seq: 3,
product: "a-product".into(),
instance: "acme".into(),
role: "main".into(),
kind: "note".into(),
body: "{}".into(),
issued_unix_ms: now(),
not_after_unix_ms: 0,
signature: Vec::new(),
};
o.signature = ed25519_dalek::SigningKey::from_bytes(&MONETIZE_SEED)
.sign(&order_message(&o))
.to_bytes()
.to_vec();
one.queue.lock().unwrap().push(o);
let addr = serve!(rt, pb::fleet_server::FleetServer::new(one.clone()));
let cfg = FleetConfig {
url: format!("http://{addr}"),
instance: "acme".into(),
..Default::default()
};
let channel = {
let _g = rt.enter();
grpc::channel(&cfg).expect("a loopback door in plain http")
};
let mut link = grpc::TonicLink::new(channel, rt.handle().clone());
let me = Addressee {
product: "a-product".into(),
instance: "acme".into(),
role: "main".into(),
};
let mut p = Puller::new(me, BoxKey::from_seed([3; 32]), "test");
let mkey = ed25519_dalek::SigningKey::from_bytes(&MONETIZE_SEED)
.verifying_key()
.to_bytes();
assert_eq!(
p.report(&mut link, Noting.health(), None, None, now())
.unwrap(),
5_000
);
let pulled = p.pull(&mut link, &Noting, Some(&mkey), 0, now()).unwrap();
assert_eq!(
pulled
.acks
.iter()
.map(|a| (a.seq, a.ok, a.detail.as_str()))
.collect::<Vec<_>>(),
vec![(3, true, "noted on main")]
);
p.report(&mut link, Noting.health(), None, None, now())
.unwrap();
assert_eq!(
one.acked
.lock()
.unwrap()
.iter()
.map(|a| a.seq)
.collect::<Vec<_>>(),
vec![3],
"the ack reached monetize"
);
assert!(one.queue.lock().unwrap().is_empty());
assert!(
p.pull(&mut link, &Noting, Some(&mkey), 0, now())
.unwrap()
.acks
.is_empty(),
"nothing runs twice"
);
assert_eq!(
*one.calls.lock().unwrap(),
vec![
"report a-product/acme/main",
"orders a-product/acme/main",
"report a-product/acme/main",
"orders a-product/acme/main"
]
);
}
/// The ONE monetize's update door on `Fleet.Ask` (H16: the old `SoldEstate` door is gone): the
/// turn order twin → main → front, every ask a `FleetCall` of `role = front` under the estate's
/// client key, the answer the call's message encoded in `FleetAskAnswer.answer`.
#[derive(Clone)]
struct OneShop {
client: monetize_embed::VerifyingKey,
turn: Arc<Mutex<Vec<&'static str>>>,
reported: Arc<Mutex<Vec<String>>>,
}
fn encoded<M: prost::Message>(m: M) -> Result<tonic::Response<pb::FleetAskAnswer>, tonic::Status> {
Ok(tonic::Response::new(pb::FleetAskAnswer {
answer: m.encode_to_vec(),
..Default::default()
}))
}
#[tonic::async_trait]
impl pb::fleet_server::Fleet for OneShop {
async fn report(
&self,
_: tonic::Request<pb::FleetSigned>,
) -> Result<tonic::Response<pb::FleetReportAnswer>, tonic::Status> {
Err(tonic::Status::unimplemented("not this test's"))
}
async fn orders(
&self,
_: tonic::Request<pb::FleetSigned>,
) -> Result<tonic::Response<pb::FleetOrdersAnswer>, tonic::Status> {
Err(tonic::Status::unimplemented("not this test's"))
}
async fn ask(
&self,
r: tonic::Request<pb::FleetSigned>,
) -> Result<tonic::Response<pb::FleetAskAnswer>, tonic::Status> {
let s = r.into_inner();
let c = FleetCall {
product: s.product.clone(),
instance: s.instance.clone(),
role: s.role.clone(),
call: s.call.clone(),
body: s.body.clone(),
at_unix_ms: s.at_unix_ms,
};
monetize_embed::fleet::verify_fleet_call(&c, &s.signature, &self.client, now())
.map_err(|e| tonic::Status::unauthenticated(format!("{e:?}")))?;
if (c.product.as_str(), c.instance.as_str(), c.role.as_str())
!= ("a-product", "acme", "front")
{
return Err(tonic::Status::unauthenticated("not acme's front"));
}
let ask: monetize_embed::fleet::Ask = serde_json::from_str(&c.body)
.map_err(|e| tonic::Status::invalid_argument(e.to_string()))?;
let body: monetize_embed::fleet::FrontAsk = serde_json::from_value(ask.body)
.map_err(|e| tonic::Status::invalid_argument(e.to_string()))?;
match ask.kind.as_str() {
k if k == monetize_embed::fleet::ask::UPGRADE => encoded(pb::EstateUpgradeResponse {
upgraded: !body.done,
words: format!("the {} keep-booted (done={})", body.role, body.done),
..Default::default()
}),
k if k == monetize_embed::fleet::ask::UPDATE_TAKE => {
encoded(match self.turn.lock().unwrap().first() {
Some(role) => pb::EstateUpdateTakeResponse {
open: true,
release: "03".into(),
live: false,
role: role.to_string(),
words: format!("the {role}'s turn"),
..Default::default()
},
None => pb::EstateUpdateTakeResponse::default(),
})
}
k if k == monetize_embed::fleet::ask::UPDATE_REPORT => {
let mut turn = self.turn.lock().unwrap();
if turn.first().copied() != Some(body.role.as_str()) {
return Err(tonic::Status::failed_precondition(format!(
"[update-out-of-turn] {} reported",
body.role
)));
}
turn.remove(0);
self.reported.lock().unwrap().push(format!(
"{} {}",
body.role,
body.carries.lines().next().unwrap_or("")
));
encoded(pb::EstateUpdateReportResponse {
words: format!("{}: recorded", body.role),
})
}
other => Err(tonic::Status::unimplemented(format!(
"not this test's ({other})"
))),
}
}
}
/// A front's hands: every box on the release's BASE, live off, the keep boot through the
/// shared door's `Upgrade` (and its `done`).
struct FrontHands(ShopDoor);
impl Hands for FrontHands {
fn running(&self, _: &str) -> Result<Running, String> {
Ok(Running {
base: "b".repeat(64),
app: "0".repeat(64),
slot: "a".into(),
live_update: true,
})
}
fn stage(&self, _: &str, _: &Stage, _: &Declared) -> Result<String, String> {
panic!("live is off: nothing is staged")
}
fn fetch(&self, _: &str) -> Result<Vec<u8>, String> {
panic!("live is off: nothing is fetched")
}
fn keep(&self, role: &str) -> Result<String, String> {
let booted = self.0.upgrade(role, false)?;
let done = self.0.upgrade(role, true)?;
Ok(format!("{}; {}", booted.words, done.words))
}
fn wait(&self, _: Duration) {}
fn staged_self(&self, _: &str) -> bool {
false
}
fn mark_staged_self(&self, _: &str) -> Result<(), String> {
Ok(())
}
fn front_keep(&self) -> String {
"A's front from a-front-keep".into()
}
}
/// ★ The shared update step over the shared door, on the real `Fleet` wire: live off, the
/// twin then the main are keep-booted through `Upgrade` and the front asks for its keep boot, in
/// the shop's order, every call verified under the estate's client key; then nothing is open.
#[test]
fn the_shared_update_step_pulls_the_shops_update_over_the_shared_door() {
let rt = tokio::runtime::Builder::new_multi_thread()
.worker_threads(2)
.enable_all()
.build()
.unwrap();
let client = ed25519_dalek::SigningKey::from_bytes(&CLIENT_SEED).verifying_key();
let shop = OneShop {
client,
turn: Arc::new(Mutex::new(vec!["twin", "main", "front"])),
reported: Arc::default(),
};
let addr = serve!(rt, pb::fleet_server::FleetServer::new(shop.clone()));
let feed = BTreeMap::from([
(
FEED_SHOP_DOOR.to_string(),
format!("door = \"http://{addr}\"\ncompany = \"acme\"\nproduct = \"a-product\"\n"),
),
(
"secret.shop_client_key".to_string(),
monetize_estate_side::hex(&CLIENT_SEED),
),
]);
let door = ShopDoor::from_feed(&feed)
.expect("a door is fed")
.expect("and usable");
let hands = FrontHands(door.clone());
for _ in 0..3 {
let (role, outcome, said) = update::look(&door, &hands)
.expect("every look is in turn")
.expect("each look reports");
assert_eq!(said, format!("{role}: recorded"), "{outcome}");
}
assert_eq!(
*shop.reported.lock().unwrap(),
vec!["twin done:keep", "main done:keep", "front needs-keep"]
);
assert_eq!(
update::look(&door, &hands).unwrap(),
None,
"nothing open: nothing reported"
);
let forged = ShopDoor::from_feed(&BTreeMap::from([
(
FEED_SHOP_DOOR.to_string(),
format!("door = \"http://{addr}\"\ncompany = \"acme\"\nproduct = \"a-product\"\n"),
),
("secret.shop_client_key".to_string(), "33".repeat(32)),
]))
.unwrap()
.unwrap();
let e = update::look(&forged, &hands).unwrap_err();
assert!(
e.starts_with("refused: "),
"a call under another key is the shop's refusal by name: {e}"
);
}
/// A ONE box that never answers an ask within the front's bound (a keep boot still running).
#[derive(Clone)]
struct SlowShop;
#[tonic::async_trait]
impl pb::fleet_server::Fleet for SlowShop {
async fn report(
&self,
_: tonic::Request<pb::FleetSigned>,
) -> Result<tonic::Response<pb::FleetReportAnswer>, tonic::Status> {
Err(tonic::Status::unimplemented("not this test's"))
}
async fn orders(
&self,
_: tonic::Request<pb::FleetSigned>,
) -> Result<tonic::Response<pb::FleetOrdersAnswer>, tonic::Status> {
Err(tonic::Status::unimplemented("not this test's"))
}
async fn ask(
&self,
_: tonic::Request<pb::FleetSigned>,
) -> Result<tonic::Response<pb::FleetAskAnswer>, tonic::Status> {
tokio::time::sleep(Duration::from_secs(5)).await;
Err(tonic::Status::unavailable("too late"))
}
}
/// ★ REL-T2-1: an ask whose own bound passed is a CUT (the shop never answered), told apart from
/// the shop's own UNAVAILABLE, so the Upgrade is polled rather than failed; past the poll's
/// bound the cut is said by name.
#[test]
fn an_ask_past_its_own_bound_is_a_cut_not_the_shops_answer() {
use monetize_estate_side::sold_door::{Polling, SHOP_CUT};
let rt = tokio::runtime::Builder::new_multi_thread()
.worker_threads(2)
.enable_all()
.build()
.unwrap();
let addr = serve!(rt, pb::fleet_server::FleetServer::new(SlowShop));
let feed = BTreeMap::from([
(
FEED_SHOP_DOOR.to_string(),
format!("door = \"http://{addr}\"\ncompany = \"acme\"\nproduct = \"a-product\"\n"),
),
(
"secret.shop_client_key".to_string(),
monetize_estate_side::hex(&CLIENT_SEED),
),
]);
let door = ShopDoor::from_feed(&feed)
.unwrap()
.unwrap()
.with_polling(Polling {
ask: Duration::from_millis(200),
every: Duration::from_millis(50),
within: Duration::from_millis(700),
});
let started = std::time::Instant::now();
let e = door.upgrade("twin", false).unwrap_err();
assert!(
e.starts_with("unavailable: ")
&& e.contains(SHOP_CUT)
&& e.contains("no answer from the shop within"),
"{e}"
);
assert!(
started.elapsed() < Duration::from_secs(3),
"every ask kept to its own bound"
);
}