holger_fleet/
pumped.rs

1//! **Holger's box on the shared pump** (P4a): the pump is monetize-estate-side's
2//! ([`monetize_estate_side::pump`]); what is Holger's is its executor ([`crate::exec`]) over its
3//! own order kinds ([`crate::order`]). [`Pumped`] is the one seam between the two.
4
5use std::collections::BTreeMap;
6
7use monetize_embed::fleet::{OrderAck, SignedOrder};
8use monetize_estate_side::pump::{grow_set, Asker, Ctx, PumpBox};
9
10use crate::exec::{execute, GrowSet, HolgerBox};
11use crate::order::decode;
12
13/// **A Holger box as the shared pump runs it**: its health words, its product report, its own
14/// twin review, and every order decoded by Holger's kinds and run through Holger's verbs. Holds
15/// any handle to a [`HolgerBox`] (`&b`, `Arc<dyn HolgerBox + Send + Sync>`).
16pub struct Pumped<B>(pub B);
17
18impl<B, T> PumpBox for Pumped<B>
19where
20    B: std::ops::Deref<Target = T>,
21    T: HolgerBox + ?Sized,
22{
23    fn health(&self) -> BTreeMap<String, String> {
24        self.0.health()
25    }
26    fn product_report(&self) -> Option<String> {
27        self.0.product_report()
28    }
29    fn own_review(&self) -> Option<String> {
30        self.0.own_review()
31    }
32    fn run(&self, ctx: &Ctx, o: &SignedOrder, ask: &mut dyn Asker) -> OrderAck {
33        match decode(o.seq, &o.kind, &o.body) {
34            Ok(order) => execute(&*self.0, ctx, o.seq, &order, &mut GrowVia(ask)),
35            Err(detail) => OrderAck { seq: o.seq, ok: false, detail, receipt_json: String::new() },
36        }
37    }
38}
39
40/// A pulled growth's `Fleet.Ask{grow_set}` on the order's own link (the shared pump's).
41struct GrowVia<'a>(&'a mut dyn Asker);
42
43impl GrowSet for GrowVia<'_> {
44    fn grow_set(&mut self, gib: u64) -> Result<String, String> {
45        grow_set(self.0, gib)
46    }
47}
48
49#[cfg(test)]
50mod tests {
51    use super::*;
52    use crate::exec::tests::{fact, Fake};
53    use crate::order::tests::{me, monetize_key, signed, MONETIZE_SEED};
54    use monetize_embed::fleet::{Ask, OrdersAfter, Report, SignedOrder};
55    use monetize_estate_side::pump::{Asked, BoxKey, Fault, Link, Puller, Signed};
56    use std::collections::BTreeMap;
57
58    /// Monetize in memory: a queue, and every call it heard.
59    #[derive(Default)]
60    struct Guardian {
61        queue: Vec<SignedOrder>,
62        heard: Vec<Signed>,
63    }
64
65    impl Link for Guardian {
66        fn report(&mut self, s: Signed) -> Result<u64, Fault> {
67            let r: Report = serde_json::from_str(&s.call.body).unwrap();
68            self.queue.retain(|o| !r.acked.iter().any(|a| a.seq == o.seq));
69            self.heard.push(s);
70            Ok(30_000)
71        }
72        fn orders(&mut self, s: Signed) -> Result<Vec<SignedOrder>, Fault> {
73            let after: OrdersAfter = serde_json::from_str(&s.call.body).unwrap();
74            self.heard.push(s);
75            Ok(self.queue.iter().filter(|o| o.seq > after.after).cloned().collect())
76        }
77        fn ask(&mut self, s: Signed) -> Result<Asked, Fault> {
78            let _: Ask = serde_json::from_str(&s.call.body).unwrap();
79            self.heard.push(s);
80            Ok(Asked { case_id: "c-1".into(), words: "resized".into(), url: String::new() })
81        }
82    }
83
84    fn mkey() -> [u8; 32] {
85        monetize_key().to_bytes()
86    }
87
88    /// ★ P4a: Holger's executor over the SHARED pump (monetize-estate-side): a fact is pulled,
89    /// run through Holger's verbs and acked, once.
90    #[test]
91    fn an_entitlement_is_pulled_run_and_acked_in_the_next_report_and_never_twice() {
92        let b = Fake::default();
93        let mut g = Guardian::default();
94        g.queue.push(signed(4, "main", "entitlement", &serde_json::to_string(&fact("libs", 5, MONETIZE_SEED)).unwrap(), MONETIZE_SEED));
95        let mut p = Puller::new(me("main"), BoxKey::from_seed([3u8; 32]), "test");
96        let pulled = p.pull(&mut g, &Pumped(&b), Some(&mkey()), 0, 10_000).unwrap();
97        assert_eq!(pulled.acks.len(), 1);
98        assert!(pulled.acks[0].ok, "{pulled:?}");
99        assert!(b.facts.borrow().contains_key("libs"));
100        p.report(&mut g, BTreeMap::from([("serving".into(), "yes".into())]), Some("{}".into()), None, 10_001).unwrap();
101        let r: Report = serde_json::from_str(&g.heard.last().unwrap().call.body).unwrap();
102        assert_eq!(r.acked.iter().map(|a| (a.seq, a.ok)).collect::<Vec<_>>(), vec![(4, true)]);
103        assert_eq!(r.product.as_deref(), Some("{}"));
104        assert!(g.queue.is_empty() && p.pulled.pending.is_empty());
105        assert!(p.pull(&mut g, &Pumped(&b), Some(&mkey()), 0, 10_002).unwrap().acks.is_empty(), "nothing runs twice");
106    }
107
108    #[test]
109    fn a_forged_order_never_runs_never_acks_and_never_moves_the_sequence() {
110        let b = Fake::default();
111        let mut g = Guardian::default();
112        g.queue.push(signed(u64::MAX, "main", "purge_tenant", r#"{"tenant":"libs"}"#, [9u8; 32]));
113        let mut p = Puller::new(me("main"), BoxKey::from_seed([3u8; 32]), "test");
114        let pulled = p.pull(&mut g, &Pumped(&b), Some(&mkey()), 0, 10_000).unwrap();
115        assert!(pulled.acks.is_empty() && pulled.forged.len() == 1);
116        assert_eq!(p.pulled.last_seq, 0);
117        assert!(b.did.borrow().is_empty());
118    }
119
120    #[test]
121    fn without_a_monetize_key_nothing_is_pulled_and_monetize_is_not_even_asked() {
122        let mut g = Guardian::default();
123        let mut p = Puller::new(me("main"), BoxKey::from_seed([3u8; 32]), "test");
124        assert_eq!(p.pull(&mut g, &Pumped(&Fake::default()), None, 0, 10_000).unwrap_err().code, "no_key");
125        assert!(g.heard.is_empty());
126    }
127
128    #[test]
129    fn the_first_call_carries_the_box_key_and_later_ones_do_not() {
130        let mut g = Guardian::default();
131        let mut p = Puller::new(me("main"), BoxKey::from_seed([3u8; 32]), "test");
132        p.report(&mut g, BTreeMap::new(), None, None, 1).unwrap();
133        p.report(&mut g, BTreeMap::new(), None, None, 2).unwrap();
134        assert_eq!(g.heard[0].box_key, p.box_public().to_vec());
135        assert!(g.heard[1].box_key.is_empty());
136    }
137
138    #[test]
139    fn a_twins_own_review_rides_every_report_and_an_ordered_one_the_next_only() {
140        let mut g = Guardian::default();
141        let mut p = Puller::new(me("twin"), BoxKey::from_seed([4u8; 32]), "test");
142        let b = Fake { twin: true, ..Default::default() };
143        g.queue.push(signed(1, "twin", "review_twin", "{}", MONETIZE_SEED));
144        p.pull(&mut g, &Pumped(&b), Some(&mkey()), 0, 10).unwrap();
145        p.report(&mut g, BTreeMap::new(), None, None, 11).unwrap();
146        p.report(&mut g, BTreeMap::new(), None, None, 12).unwrap();
147        p.report(&mut g, BTreeMap::new(), None, Some(r#"{"own":1}"#.into()), 13).unwrap();
148        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();
149        assert!(said[0].as_deref().is_some_and(|r| r.contains("complete")), "{said:?}");
150        assert_eq!(said[1], None);
151        assert_eq!(said[2].as_deref(), Some(r#"{"own":1}"#));
152    }
153
154    #[test]
155    fn the_puller_remembers_what_it_took_across_a_restart() {
156        let dir = tempfile::tempdir().unwrap();
157        let mut g = Guardian::default();
158        g.queue.push(signed(6, "main", "grow_room", r#"{"need_bytes":5,"reason":"r"}"#, MONETIZE_SEED));
159        let mut p = Puller::at(dir.path(), me("main"), "test").unwrap();
160        let key = p.box_public();
161        p.pull(&mut g, &Pumped(&Fake::default()), Some(&mkey()), 0, 10_000).unwrap();
162        drop(p);
163        let p = Puller::at(dir.path(), me("main"), "test").unwrap();
164        assert_eq!((p.box_public(), p.pulled.last_seq, p.pulled.pending.len()), (key, 6, 1));
165    }
166}
holger_fleet/pumped.rs

holger_fleet/
pumped.rs
