estate/holger-fleet/src/pumped.rs
//! **Holger's box on the shared pump** (P4a): the pump is monetize-estate-side's
//! ([`monetize_estate_side::pump`]); what is Holger's is its executor ([`crate::exec`]) over its
//! own order kinds ([`crate::order`]). [`Pumped`] is the one seam between the two.
use std::collections::BTreeMap;
use monetize_embed::fleet::{OrderAck, SignedOrder};
use monetize_estate_side::pump::{grow_set, Asker, Ctx, PumpBox};
use crate::exec::{execute, GrowSet, HolgerBox};
use crate::order::decode;
/// **A Holger box as the shared pump runs it**: its health words, its product report, its own
/// twin review, and every order decoded by Holger's kinds and run through Holger's verbs. Holds
/// any handle to a [`HolgerBox`] (`&b`, `Arc<dyn HolgerBox + Send + Sync>`).
pub struct Pumped<B>(pub B);
impl<B, T> PumpBox for Pumped<B>
where
B: std::ops::Deref<Target = T>,
T: HolgerBox + ?Sized,
{
fn health(&self) -> BTreeMap<String, String> {
self.0.health()
}
fn product_report(&self) -> Option<String> {
self.0.product_report()
}
fn own_review(&self) -> Option<String> {
self.0.own_review()
}
fn run(&self, ctx: &Ctx, o: &SignedOrder, ask: &mut dyn Asker) -> OrderAck {
match decode(o.seq, &o.kind, &o.body) {
Ok(order) => execute(&*self.0, ctx, o.seq, &order, &mut GrowVia(ask)),
Err(detail) => OrderAck {
seq: o.seq,
ok: false,
detail,
receipt_json: String::new(),
},
}
}
}
/// A pulled growth's `Fleet.Ask{grow_set}` on the order's own link (the shared pump's).
struct GrowVia<'a>(&'a mut dyn Asker);
impl GrowSet for GrowVia<'_> {
fn grow_set(&mut self, gib: u64) -> Result<String, String> {
grow_set(self.0, gib)
}
fn fetch(&mut self, f: &monetize_embed::fleet::Fetch) -> Result<Vec<u8>, String> {
monetize_estate_side::pump::fetch(self.0, f)
}
}
#[cfg(test)]
mod tests {
use super::*;
use crate::exec::tests::{fact, Fake};
use crate::order::tests::{me, monetize_key, signed, MONETIZE_SEED};
use monetize_embed::fleet::{Ask, OrdersAfter, Report, SignedOrder};
use monetize_estate_side::pump::{Asked, BoxKey, Fault, Link, Puller, Signed};
use std::collections::BTreeMap;
/// Monetize in memory: a queue, and every call it heard.
#[derive(Default)]
struct Guardian {
queue: Vec<SignedOrder>,
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(),
})
}
}
fn mkey() -> [u8; 32] {
monetize_key().to_bytes()
}
/// ★ P4a: Holger's executor over the SHARED pump (monetize-estate-side): a fact is pulled,
/// run through Holger's verbs and acked, once.
#[test]
fn an_entitlement_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",
"entitlement",
&serde_json::to_string(&fact("libs", 5, MONETIZE_SEED)).unwrap(),
MONETIZE_SEED,
));
let mut p = Puller::new(me("main"), BoxKey::from_seed([3u8; 32]), "test");
let pulled = p
.pull(&mut g, &Pumped(&b), Some(&mkey()), 0, 10_000)
.unwrap();
assert_eq!(pulled.acks.len(), 1);
assert!(pulled.acks[0].ok, "{pulled:?}");
assert!(b.facts.borrow().contains_key("libs"));
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(), Some("{}"));
assert!(g.queue.is_empty() && p.pulled.pending.is_empty());
assert!(
p.pull(&mut g, &Pumped(&b), Some(&mkey()), 0, 10_002)
.unwrap()
.acks
.is_empty(),
"nothing runs twice"
);
}
#[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",
"purge_tenant",
r#"{"tenant":"libs"}"#,
[9u8; 32],
));
let mut p = Puller::new(me("main"), BoxKey::from_seed([3u8; 32]), "test");
let pulled = p
.pull(&mut g, &Pumped(&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.did.borrow().is_empty());
}
#[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, &Pumped(&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");
let b = Fake {
twin: true,
..Default::default()
};
g.queue
.push(signed(1, "twin", "review_twin", "{}", MONETIZE_SEED));
p.pull(&mut g, &Pumped(&b), 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}"#));
}
#[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",
"grow_room",
r#"{"need_bytes":5,"reason":"r"}"#,
MONETIZE_SEED,
));
let mut p = Puller::at(dir.path(), me("main"), "test").unwrap();
let key = p.box_public();
p.pull(&mut g, &Pumped(&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)
);
}
}