crates/monetize-estate-side/src/pump/puller.rs
//! **The puller**: report, pull, verify, run, ack — the box's whole side of
//! `monetize.v1.Fleet`, over any [`Link`] (the gRPC client in [`super::grpc`], an in-process
//! guardian in the tests), running each order through the product's [`PumpBox`].
//!
//! Moved from Holger's `holger-fleet/src/puller.rs` (P4a). What was Holger's there and is the
//! product's here: the order kinds and their bodies ([`PumpBox::run`]), the health words.
use std::collections::BTreeMap;
use std::path::{Path, PathBuf};
use monetize_embed::fleet::{
ask, call, kind, order_message, Ask, FleetCall, OrderAck, OrdersAfter, Report, SignedOrder,
LONG_POLL_MAX_MS,
};
use monetize_embed::VerifyingKey;
use serde::{Deserialize, Serialize};
use super::key::BoxKey;
/// What the puller remembers across restarts.
pub const STATE_REL: &str = "fleet/pulled.json";
/// Who this box is on the wire.
#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
pub struct Addressee {
pub product: String,
pub instance: String,
/// `main` | `twin`.
pub role: String,
}
/// Why a call failed: monetize's word (`unauthenticated`, `unavailable`, …) and its sentence.
#[derive(Debug, Clone, PartialEq, Eq)]
pub struct Fault {
pub code: String,
pub message: String,
}
impl Fault {
pub fn new(code: impl Into<String>, message: impl Into<String>) -> Fault {
Fault {
code: code.into(),
message: message.into(),
}
}
}
impl std::fmt::Display for Fault {
fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
write!(f, "{}: {}", self.code, self.message)
}
}
/// One signed call as it goes on the wire.
#[derive(Debug, Clone)]
pub struct Signed {
pub call: FleetCall,
pub signature: Vec<u8>,
/// The box's public key, sent until monetize has heard the box once (its pin).
pub box_key: Vec<u8>,
}
/// **What an ask is answered**: the case it opened (a remedy), monetize's words, and a page
/// to open (empty for every ask a box makes today).
#[derive(Debug, Clone, Default, PartialEq, Eq)]
pub struct Asked {
pub case_id: String,
pub words: String,
pub url: String,
}
/// **The three calls**, over whatever carries them.
pub trait Link {
/// Answers monetize's `next_report_ms`.
fn report(&mut self, s: Signed) -> Result<u64, Fault>;
fn orders(&mut self, s: Signed) -> Result<Vec<SignedOrder>, Fault>;
fn ask(&mut self, s: Signed) -> Result<Asked, Fault>;
/// [`Link::ask`] with the answer's BYTES (`FleetAskAnswer.answer`): the file a `fetch` ask
/// pulls (MONETIZE-FIX). A link that carries no bytes answers none.
fn ask_bytes(&mut self, s: Signed) -> Result<(Asked, Vec<u8>), Fault> {
self.ask(s).map(|a| (a, Vec::new()))
}
}
/// **A `Fleet.Ask` from inside a running order** (a pulled growth's `grow_set`), signed by this
/// box, on the link the order came on.
pub trait Asker {
fn ask(&mut self, a: &Ask) -> Result<Asked, Fault>;
/// [`Asker::ask`] with the answer's bytes ([`Link::ask_bytes`]).
fn ask_bytes(&mut self, a: &Ask) -> Result<(Asked, Vec<u8>), Fault> {
self.ask(a).map(|x| (x, Vec::new()))
}
}
/// **A pulled growth's disk resize** (`Fleet.Ask{grow_set}`, every member, `gib` a disk): monetize's
/// words, or the case it opened when it said none.
pub fn grow_set(asker: &mut dyn Asker, gib: u64) -> Result<String, String> {
let a = Ask {
kind: ask::GROW_SET.into(),
body: serde_json::json!({ "members": [], "gib": gib }),
};
let Asked {
case_id: case,
words,
..
} = asker.ask(&a).map_err(|f| f.to_string())?;
Ok(if words.is_empty() { case } else { words })
}
/// **The file a product order carries, pulled** (MONETIZE-FIX 2026-11-01: the ONE box never
/// pushes a file into an estate): `Fleet.Ask{fetch}` on the order's own link. The ONE box answers
/// only a file an order queued for this box names, checked against its sha256 there; its size
/// is checked here before it is handed to the product's executor.
pub fn fetch(asker: &mut dyn Asker, f: &monetize_embed::fleet::Fetch) -> Result<Vec<u8>, String> {
let a = Ask {
kind: ask::FETCH.into(),
body: serde_json::json!({ "id": f.id }),
};
let (_, bytes) = asker.ask_bytes(&a).map_err(|e| e.to_string())?;
if bytes.len() as u64 != f.bytes {
return Err(format!(
"the file pulled for {} is {} bytes, and the order names {}",
f.param,
bytes.len(),
f.bytes
));
}
Ok(bytes)
}
/// What the product's executor knows about the order's context.
pub struct Ctx<'a> {
pub me: &'a Addressee,
/// The monetize key the order verified under (a fact or snapshot inside it is verified again).
pub key: &'a VerifyingKey,
pub now_unix_ms: u64,
}
/// **The product's half of the pump**: what its box reports, and what it does with an order.
pub trait PumpBox {
/// The box's health words for `Fleet.Report` (`serving`, `data`, `serves_at`, `twin`, `epoch`).
fn health(&self) -> BTreeMap<String, String> {
BTreeMap::new()
}
/// `monetize_product::pull::ProductReport` as JSON — the main that serves sends it.
fn product_report(&self) -> Option<String> {
None
}
/// A twin reviews itself on every report (G12): its review as JSON.
fn own_review(&self) -> Option<String> {
None
}
/// **Run one order that passed [`check`]** (signed by monetize, for this box, newer, in its
/// window): decode its body, run it through the product's own verbs, and answer its ack. A
/// kind or body the product does not take is an ack not-ok, by name. `ask` is
/// `Fleet.Ask` on the order's own link ([`grow_set`]).
fn run(&self, ctx: &Ctx, order: &SignedOrder, ask: &mut dyn Asker) -> OrderAck;
}
/// How far ahead of this box's clock an order may be dated (Gunnar's order check: five minutes).
/// ONE constant with monetize-embed's `fleet::verify_order` (T16, A28 M11).
pub use monetize_embed::fleet::FUTURE_SKEW_MS;
/// What [`check`] made of one wire order, before its body is read.
#[derive(Debug, Clone, PartialEq, Eq)]
pub enum Checked {
/// Not monetize's: never run, never acked, the sequence does not move.
Forged(String),
/// Monetize's, but not for this box to run now: acked not-ok with these words.
Refused(String),
/// Monetize's, for this box, newer than the last one taken, in its window: run it.
Fresh,
}
/// **The box's check of one order** (the order's body is the product's): signed by monetize
/// (monetize-embed [`order_message`] under `key`), then for this box, newer than the last order
/// taken, inside its window (`not_after` 0 = none), dated no more than [`FUTURE_SKEW_MS`] ahead
/// of this box's clock. An order that does not verify is FORGED —
/// never run, never acked, and it never moves the sequence (a forged high seq must not silence
/// the real ones). Everything else wrong with it is a REFUSAL, acked by name.
pub fn check(
o: &SignedOrder,
me: &Addressee,
key: &VerifyingKey,
last_seq: u64,
now_unix_ms: u64,
) -> Checked {
// STRICT (P4c, A28 M11): no small-order key or `R`, as Gunnar's own check always was.
let signed = ed25519_dalek::Signature::from_slice(&o.signature)
.map(|s| key.verify_strict(&order_message(o), &s).is_ok())
.unwrap_or(false);
if !signed {
return Checked::Forged(format!(
"order {} ({}) is not signed by this product's monetize key",
o.seq, o.kind
));
}
if o.product != me.product || o.instance != me.instance || o.role != me.role {
return Checked::Refused(format!(
"order {} is for {}/{}/{}, and this box is {}/{}/{}",
o.seq, o.product, o.instance, o.role, me.product, me.instance, me.role
));
}
if o.seq <= last_seq {
return Checked::Refused(format!(
"order {} is not newer than {last_seq}, the last one this box took: a replay",
o.seq
));
}
if o.not_after_unix_ms != 0 && now_unix_ms > o.not_after_unix_ms {
return Checked::Refused(format!(
"order {} expired {} s ago",
o.seq,
(now_unix_ms - o.not_after_unix_ms) / 1000
));
}
if o.issued_unix_ms > now_unix_ms.saturating_add(FUTURE_SKEW_MS) {
return Checked::Refused(format!(
"order {} is dated {} s in the future; whose clock is wrong?",
o.seq,
(o.issued_unix_ms - now_unix_ms) / 1000
));
}
Checked::Fresh
}
/// What is kept across restarts.
#[derive(Debug, Clone, Default, PartialEq, Eq, Serialize, Deserialize)]
#[serde(default)]
pub struct Pulled {
/// The last order TAKEN (persisted before it runs: an order runs at most once).
pub last_seq: u64,
/// Acks not yet carried by a Report monetize answered.
pub pending: Vec<OrderAck>,
/// A review a `review_twin` order made: it rides the next Report only.
pub twin_review: Option<String>,
}
/// What one pull did.
#[derive(Debug, Clone, Default, PartialEq, Eq)]
pub struct Pull {
pub acks: Vec<OrderAck>,
pub forged: Vec<String>,
}
pub struct Puller {
pub me: Addressee,
pub build: String,
key: BoxKey,
state: Option<PathBuf>,
pub pulled: Pulled,
heard: bool,
}
impl Puller {
/// A puller that remembers nothing on disk (tests).
pub fn new(me: Addressee, key: BoxKey, build: impl Into<String>) -> Puller {
Puller {
me,
build: build.into(),
key,
state: None,
pulled: Pulled::default(),
heard: false,
}
}
/// The box's puller over its data root: its key (minted on first use) and what it took.
pub fn at(root: &Path, me: Addressee, build: impl Into<String>) -> std::io::Result<Puller> {
let key = BoxKey::load_or_mint(root)?;
let state = root.join(STATE_REL);
let pulled = match std::fs::read(&state) {
Ok(bytes) => serde_json::from_slice(&bytes).map_err(|e| {
std::io::Error::new(
std::io::ErrorKind::InvalidData,
format!("{}: {e}", state.display()),
)
})?,
Err(e) if e.kind() == std::io::ErrorKind::NotFound => Pulled::default(),
Err(e) => return Err(e),
};
Ok(Puller {
me,
build: build.into(),
key,
state: Some(state),
pulled,
heard: false,
})
}
pub fn box_public(&self) -> [u8; 32] {
self.key.public()
}
fn save(&self) -> Result<(), Fault> {
let Some(path) = &self.state else {
return Ok(());
};
let bytes =
serde_json::to_vec(&self.pulled).map_err(|e| Fault::new("internal", e.to_string()))?;
crate::write_atomic(path, &bytes, false)
.map_err(|e| Fault::new("io", format!("{}: {e}", path.display())))
}
fn signed(&self, which: &str, body: String, now_unix_ms: u64) -> Signed {
let c = FleetCall {
product: self.me.product.clone(),
instance: self.me.instance.clone(),
role: self.me.role.clone(),
call: which.to_owned(),
body,
at_unix_ms: now_unix_ms,
};
let signature = self.key.sign(&c);
Signed {
call: c,
signature,
box_key: if self.heard {
Vec::new()
} else {
self.key.public().to_vec()
},
}
}
/// **Report** this box's health, the product's own part (the main that serves), its own
/// twin review (a twin, every report) and every pending ack. The acks leave the queue
/// once monetize answered. Returns monetize's `next_report_ms`.
pub fn report(
&mut self,
link: &mut dyn Link,
health: BTreeMap<String, String>,
product: Option<String>,
own_review: Option<String>,
now_unix_ms: u64,
) -> Result<u64, Fault> {
let sent = self.pulled.pending.len();
let report = Report {
build: self.build.clone(),
health,
acked: self.pulled.pending.clone(),
twin_review: own_review.or_else(|| self.pulled.twin_review.clone()),
product,
measured_unix_ms: now_unix_ms,
};
let body =
serde_json::to_string(&report).map_err(|e| Fault::new("internal", e.to_string()))?;
let next = link.report(self.signed(call::REPORT, body, now_unix_ms))?;
self.heard = true;
self.pulled.pending.drain(..sent);
self.pulled.twin_review = None;
self.save()?;
Ok(next)
}
/// **Pull** the orders after the last one taken, held up to `wait_ms` (≤ 25 s); [`check`]
/// each under `monetize_pubkey`, run it through `b` and queue its ack. Without a key nothing
/// is pulled: nothing could be verified. A `review_twin` that ran rides the next report.
pub fn pull(
&mut self,
link: &mut dyn Link,
b: &dyn PumpBox,
monetize_pubkey: Option<&[u8; 32]>,
wait_ms: u64,
now_unix_ms: u64,
) -> Result<Pull, Fault> {
let Some(raw) = monetize_pubkey else {
return Err(Fault::new(
"no_key",
"this box has no monetize key yet, so no order could be verified: none is pulled",
));
};
let key = VerifyingKey::from_bytes(raw).map_err(|e| {
Fault::new(
"no_key",
format!("the monetize key is not an Ed25519 point: {e}"),
)
})?;
let body = serde_json::to_string(&OrdersAfter {
after: self.pulled.last_seq,
wait_ms: wait_ms.min(LONG_POLL_MAX_MS),
})
.map_err(|e| Fault::new("internal", e.to_string()))?;
let mut orders = link.orders(self.signed(call::ORDERS, body, now_unix_ms))?;
self.heard = true;
orders.sort_by_key(|o| o.seq);
let mut out = Pull::default();
for o in orders {
let checked = check(&o, &self.me, &key, self.pulled.last_seq, now_unix_ms);
if let Checked::Forged(why) = checked {
out.forged.push(why);
continue;
}
if o.seq <= self.pulled.last_seq {
continue;
}
self.pulled.last_seq = o.seq;
self.save()?;
let ack = match checked {
Checked::Fresh => {
let me = self.me.clone();
let ctx = Ctx {
me: &me,
key: &key,
now_unix_ms,
};
let ack = {
let mut via = AskVia {
puller: &mut *self,
link: &mut *link,
now_unix_ms,
};
b.run(&ctx, &o, &mut via)
};
if o.kind == kind::REVIEW_TWIN && ack.ok {
self.pulled.twin_review = Some(ack.receipt_json.clone());
}
ack
}
Checked::Refused(detail) => OrderAck {
seq: o.seq,
ok: false,
detail,
receipt_json: String::new(),
},
Checked::Forged(_) => unreachable!("forged orders were skipped"),
};
self.pulled.pending.push(ack.clone());
self.save()?;
out.acks.push(ack);
}
Ok(out)
}
/// **Ask** monetize (`grow_set`, or a remedy: it opens a case, never approves one).
pub fn ask(&mut self, link: &mut dyn Link, a: &Ask, now_unix_ms: u64) -> Result<Asked, Fault> {
let body = serde_json::to_string(a).map_err(|e| Fault::new("internal", e.to_string()))?;
let answer = link.ask(self.signed(call::ASK, body, now_unix_ms))?;
self.heard = true;
Ok(answer)
}
/// [`Puller::ask`] with the answer's bytes ([`Link::ask_bytes`]).
pub fn ask_bytes(
&mut self,
link: &mut dyn Link,
a: &Ask,
now_unix_ms: u64,
) -> Result<(Asked, Vec<u8>), Fault> {
let body = serde_json::to_string(a).map_err(|e| Fault::new("internal", e.to_string()))?;
let answer = link.ask_bytes(self.signed(call::ASK, body, now_unix_ms))?;
self.heard = true;
Ok(answer)
}
}
/// An order's `Fleet.Ask`, signed by this box, on the link the order came on.
struct AskVia<'a> {
puller: &'a mut Puller,
link: &'a mut dyn Link,
now_unix_ms: u64,
}
impl Asker for AskVia<'_> {
fn ask(&mut self, a: &Ask) -> Result<Asked, Fault> {
self.puller.ask(self.link, a, self.now_unix_ms)
}
fn ask_bytes(&mut self, a: &Ask) -> Result<(Asked, Vec<u8>), Fault> {
self.puller.ask_bytes(self.link, a, self.now_unix_ms)
}
}
#[cfg(test)]
pub(crate) mod tests {
use super::*;
use ed25519_dalek::Signer as _;
use std::cell::RefCell;
pub const MONETIZE_SEED: [u8; 32] = [7u8; 32];
pub fn me(role: &str) -> Addressee {
Addressee {
product: "a-product".into(),
instance: "acme".into(),
role: role.into(),
}
}
pub fn monetize_key() -> VerifyingKey {
ed25519_dalek::SigningKey::from_bytes(&MONETIZE_SEED).verifying_key()
}
fn mkey() -> [u8; 32] {
monetize_key().to_bytes()
}
/// One order for `role`, signed by `seed`.
pub fn signed(seq: u64, role: &str, kind: &str, body: &str, seed: [u8; 32]) -> SignedOrder {
let mut o = SignedOrder {
seq,
product: "a-product".into(),
instance: "acme".into(),
role: role.into(),
kind: kind.into(),
body: body.into(),
issued_unix_ms: 9_000,
not_after_unix_ms: 0,
signature: Vec::new(),
};
o.signature = ed25519_dalek::SigningKey::from_bytes(&seed)
.sign(&order_message(&o))
.to_bytes()
.to_vec();
o
}
/// Monetize in memory: a queue, and every call it heard.
#[derive(Default)]
pub struct Guardian {
pub queue: Vec<SignedOrder>,
pub heard: Vec<Signed>,
}
impl Link for Guardian {
fn report(&mut self, s: Signed) -> Result<u64, Fault> {
let r: Report = serde_json::from_str(&s.call.body).unwrap();
self.queue
.retain(|o| !r.acked.iter().any(|a| a.seq == o.seq));
self.heard.push(s);
Ok(30_000)
}
fn orders(&mut self, s: Signed) -> Result<Vec<SignedOrder>, Fault> {
let after: OrdersAfter = serde_json::from_str(&s.call.body).unwrap();
self.heard.push(s);
Ok(self
.queue
.iter()
.filter(|o| o.seq > after.after)
.cloned()
.collect())
}
fn ask(&mut self, s: Signed) -> Result<Asked, Fault> {
let _: Ask = serde_json::from_str(&s.call.body).unwrap();
self.heard.push(s);
Ok(Asked {
case_id: "c-1".into(),
words: "resized".into(),
url: String::new(),
})
}
}
/// A product box: it runs `note` (its words) and `grow` (asks `grow_set`), reviews as a
/// twin, and refuses any other kind by name.
#[derive(Default)]
pub struct Fake {
pub ran: RefCell<Vec<(u64, String)>>,
}
impl PumpBox for Fake {
fn run(&self, ctx: &Ctx, o: &SignedOrder, ask: &mut dyn Asker) -> OrderAck {
self.ran.borrow_mut().push((o.seq, o.kind.clone()));
let said = match o.kind.as_str() {
"note" => Ok(format!("noted for {}", ctx.me.role)),
"grow" => grow_set(ask, 64),
kind::REVIEW_TWIN => {
return OrderAck {
seq: o.seq,
ok: true,
detail: "reviewed".into(),
receipt_json: r#"{"complete":"yes"}"#.into(),
}
}
other => Err(format!(
"order {}: this product does not know the order kind {other:?}",
o.seq
)),
};
match said {
Ok(detail) => OrderAck {
seq: o.seq,
ok: true,
detail,
receipt_json: String::new(),
},
Err(detail) => OrderAck {
seq: o.seq,
ok: false,
detail,
receipt_json: String::new(),
},
}
}
}
#[test]
fn an_order_is_pulled_run_and_acked_in_the_next_report_and_never_twice() {
let b = Fake::default();
let mut g = Guardian::default();
g.queue.push(signed(4, "main", "note", "{}", MONETIZE_SEED));
let mut p = Puller::new(me("main"), BoxKey::from_seed([3u8; 32]), "test");
let pulled = p.pull(&mut g, &b, Some(&mkey()), 0, 10_000).unwrap();
assert_eq!(pulled.acks.len(), 1);
assert!(pulled.acks[0].ok, "{pulled:?}");
assert_eq!(pulled.acks[0].detail, "noted for main");
p.report(
&mut g,
BTreeMap::from([("serving".into(), "yes".into())]),
Some("{}".into()),
None,
10_001,
)
.unwrap();
let r: Report = serde_json::from_str(&g.heard.last().unwrap().call.body).unwrap();
assert_eq!(
r.acked.iter().map(|a| (a.seq, a.ok)).collect::<Vec<_>>(),
vec![(4, true)]
);
assert_eq!(
(r.product.as_deref(), r.build.as_str()),
(Some("{}"), "test")
);
assert!(g.queue.is_empty() && p.pulled.pending.is_empty());
assert!(
p.pull(&mut g, &b, Some(&mkey()), 0, 10_002)
.unwrap()
.acks
.is_empty(),
"nothing runs twice"
);
assert_eq!(b.ran.borrow().len(), 1);
}
/// ★ **The one order check is STRICT Ed25519** (P4c, A28 M11: Gunnar's gunnar-repair copy
/// always was, `verify_strict`). Under a small-order key (the identity point) the all-zero
/// `R = identity, S = 0` signature satisfies the plain equation for ANY message; a key fed
/// wrong must not turn every order into monetize's. A real key still verifies.
#[test]
fn a_weak_key_signs_no_order_and_a_real_one_still_does() {
let mut identity = [0u8; 32];
identity[0] = 1;
let weak = VerifyingKey::from_bytes(&identity).expect("the identity point decompresses");
let mut o = signed(3, "main", "recharge_from_twin", "{}", MONETIZE_SEED);
o.signature = {
let mut s = vec![0u8; 64];
s[0] = 1;
s
};
assert!(
matches!(check(&o, &me("main"), &weak, 0, 9_500), Checked::Forged(_)),
"a weak key verifies nothing"
);
let real = signed(3, "main", "recharge_from_twin", "{}", MONETIZE_SEED);
assert_eq!(
check(&real, &me("main"), &monetize_key(), 0, 9_500),
Checked::Fresh
);
}
/// ★ **An order dated more than five minutes ahead of this box's clock is refused** (P4c,
/// A28 M11: Gunnar's order check always refused it). Five minutes of skew still runs.
#[test]
fn an_order_dated_ahead_of_the_box_clock_is_refused_by_name() {
let mut ahead = signed(3, "main", "note", "{}", MONETIZE_SEED);
ahead.issued_unix_ms = 9_000 + FUTURE_SKEW_MS + 1;
ahead.signature = ed25519_dalek::SigningKey::from_bytes(&MONETIZE_SEED)
.sign(&order_message(&ahead))
.to_bytes()
.to_vec();
match check(&ahead, &me("main"), &monetize_key(), 0, 9_000) {
Checked::Refused(why) => assert!(why.contains("in the future"), "{why}"),
other => panic!("{other:?}"),
}
let mut skew = ahead.clone();
skew.issued_unix_ms = 9_000 + FUTURE_SKEW_MS;
skew.signature = ed25519_dalek::SigningKey::from_bytes(&MONETIZE_SEED)
.sign(&order_message(&skew))
.to_bytes()
.to_vec();
assert_eq!(
check(&skew, &me("main"), &monetize_key(), 0, 9_000),
Checked::Fresh
);
}
#[test]
fn a_forged_order_never_runs_never_acks_and_never_moves_the_sequence() {
let b = Fake::default();
let mut g = Guardian::default();
g.queue
.push(signed(u64::MAX, "main", "note", "{}", [9u8; 32]));
let mut p = Puller::new(me("main"), BoxKey::from_seed([3u8; 32]), "test");
let pulled = p.pull(&mut g, &b, Some(&mkey()), 0, 10_000).unwrap();
assert!(pulled.acks.is_empty() && pulled.forged.len() == 1);
assert_eq!(p.pulled.last_seq, 0);
assert!(b.ran.borrow().is_empty());
}
/// ★ The check before the body: another box's order, an expired one and a replay are acked
/// not-ok by name and never reach the product; a kind the product does not take is ITS
/// refusal, acked by name too.
#[test]
fn another_boxs_an_expired_or_an_unknown_order_is_refused_by_name() {
let k = monetize_key();
let twin = signed(5, "twin", "note", "{}", MONETIZE_SEED);
assert!(
matches!(check(&twin, &me("main"), &k, 0, 10_000), Checked::Refused(w) if w.contains("for a-product/acme/twin"))
);
let mut late = signed(5, "main", "note", "{}", MONETIZE_SEED);
late.not_after_unix_ms = 1_000;
let late = {
let mut l = late.clone();
l.signature = signed_like(&l);
l
};
assert!(
matches!(check(&late, &me("main"), &k, 0, 61_000), Checked::Refused(w) if w.contains("expired 60 s ago"))
);
assert!(
matches!(check(&signed(5, "main", "note", "{}", MONETIZE_SEED), &me("main"), &k, 5, 1), Checked::Refused(w) if w.contains("a replay"))
);
assert_eq!(
check(
&signed(6, "main", "note", "{}", MONETIZE_SEED),
&me("main"),
&k,
5,
1
),
Checked::Fresh
);
let b = Fake::default();
let mut g = Guardian::default();
g.queue.push(twin);
g.queue
.push(signed(7, "main", "reboot_the_moon", "{}", MONETIZE_SEED));
let mut p = Puller::new(me("main"), BoxKey::from_seed([3u8; 32]), "test");
let pulled = p.pull(&mut g, &b, Some(&mkey()), 0, 10_000).unwrap();
assert_eq!(
pulled
.acks
.iter()
.map(|a| (a.seq, a.ok))
.collect::<Vec<_>>(),
vec![(5, false), (7, false)]
);
assert!(
pulled.acks[1]
.detail
.contains("does not know the order kind \"reboot_the_moon\""),
"{pulled:?}"
);
assert_eq!(
*b.ran.borrow(),
vec![(7, "reboot_the_moon".to_string())],
"only the box's own order reaches the product"
);
}
fn signed_like(o: &SignedOrder) -> Vec<u8> {
ed25519_dalek::SigningKey::from_bytes(&MONETIZE_SEED)
.sign(&order_message(o))
.to_bytes()
.to_vec()
}
#[test]
fn without_a_monetize_key_nothing_is_pulled_and_monetize_is_not_even_asked() {
let mut g = Guardian::default();
let mut p = Puller::new(me("main"), BoxKey::from_seed([3u8; 32]), "test");
assert_eq!(
p.pull(&mut g, &Fake::default(), None, 0, 10_000)
.unwrap_err()
.code,
"no_key"
);
assert!(g.heard.is_empty());
}
#[test]
fn the_first_call_carries_the_box_key_and_later_ones_do_not() {
let mut g = Guardian::default();
let mut p = Puller::new(me("main"), BoxKey::from_seed([3u8; 32]), "test");
p.report(&mut g, BTreeMap::new(), None, None, 1).unwrap();
p.report(&mut g, BTreeMap::new(), None, None, 2).unwrap();
assert_eq!(g.heard[0].box_key, p.box_public().to_vec());
assert!(g.heard[1].box_key.is_empty());
}
#[test]
fn a_twins_own_review_rides_every_report_and_an_ordered_one_the_next_only() {
let mut g = Guardian::default();
let mut p = Puller::new(me("twin"), BoxKey::from_seed([4u8; 32]), "test");
g.queue
.push(signed(1, "twin", "review_twin", "{}", MONETIZE_SEED));
p.pull(&mut g, &Fake::default(), Some(&mkey()), 0, 10)
.unwrap();
p.report(&mut g, BTreeMap::new(), None, None, 11).unwrap();
p.report(&mut g, BTreeMap::new(), None, None, 12).unwrap();
p.report(
&mut g,
BTreeMap::new(),
None,
Some(r#"{"own":1}"#.into()),
13,
)
.unwrap();
let said: Vec<Option<String>> = g
.heard
.iter()
.filter(|s| s.call.call == "report")
.map(|s| {
serde_json::from_str::<Report>(&s.call.body)
.unwrap()
.twin_review
})
.collect();
assert!(
said[0].as_deref().is_some_and(|r| r.contains("complete")),
"{said:?}"
);
assert_eq!(said[1], None);
assert_eq!(said[2].as_deref(), Some(r#"{"own":1}"#));
}
/// ★ An order's `Fleet.Ask` goes on the order's own link, signed by this box.
#[test]
fn an_order_asks_monetize_on_its_own_link_signed_by_the_box() {
let mut g = Guardian::default();
g.queue.push(signed(2, "main", "grow", "{}", MONETIZE_SEED));
let mut p = Puller::new(me("main"), BoxKey::from_seed([3u8; 32]), "test");
let pulled = p
.pull(&mut g, &Fake::default(), Some(&mkey()), 0, 10_000)
.unwrap();
assert_eq!(pulled.acks[0].detail, "resized");
let asked = g
.heard
.iter()
.find(|s| s.call.call == "ask")
.expect("asked");
let a: Ask = serde_json::from_str(&asked.call.body).unwrap();
assert_eq!(
(a.kind.as_str(), a.body["gib"].as_u64()),
("grow_set", Some(64))
);
let vk = VerifyingKey::from_bytes(&p.box_public()).unwrap();
monetize_embed::fleet::verify_fleet_call(&asked.call, &asked.signature, &vk, 10_000)
.expect("the ask is the box's");
}
#[test]
fn the_puller_remembers_what_it_took_across_a_restart() {
let dir = tempfile::tempdir().unwrap();
let mut g = Guardian::default();
g.queue.push(signed(6, "main", "note", "{}", MONETIZE_SEED));
let mut p = Puller::at(dir.path(), me("main"), "test").unwrap();
let key = p.box_public();
p.pull(&mut g, &Fake::default(), Some(&mkey()), 0, 10_000)
.unwrap();
drop(p);
let p = Puller::at(dir.path(), me("main"), "test").unwrap();
assert_eq!(
(p.box_public(), p.pulled.last_seq, p.pulled.pending.len()),
(key, 6, 1)
);
}
/// ★ MONETIZE-FIX: **a product order's file is pulled with a `fetch` ask**, naming the file;
/// bytes of another size than the order names are refused, never handed on.
#[test]
fn a_product_orders_file_is_pulled_by_a_fetch_ask_and_checked() {
struct Shop(Vec<u8>, Vec<Ask>);
impl Asker for Shop {
fn ask(&mut self, _: &Ask) -> Result<Asked, Fault> {
unreachable!("a fetch asks for its bytes")
}
fn ask_bytes(&mut self, a: &Ask) -> Result<(Asked, Vec<u8>), Fault> {
self.1.push(a.clone());
Ok((Asked::default(), self.0.clone()))
}
}
let f = monetize_embed::fleet::Fetch {
id: "ab".repeat(32),
bytes: 3,
param: "bundle".into(),
};
let mut shop = Shop(b"abc".to_vec(), Vec::new());
assert_eq!(fetch(&mut shop, &f).unwrap(), b"abc");
assert_eq!(
(shop.1[0].kind.as_str(), shop.1[0].body["id"].as_str()),
(ask::FETCH, Some(f.id.as_str()))
);
let mut short = Shop(b"ab".to_vec(), Vec::new());
assert!(fetch(&mut short, &f).unwrap_err().contains("names 3"));
}
}