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 {
36                seq: o.seq,
37                ok: false,
38                detail,
39                receipt_json: String::new(),
40            },
41        }
42    }
43}
44
45/// A pulled growth's `Fleet.Ask{grow_set}` on the order's own link (the shared pump's).
46struct GrowVia<'a>(&'a mut dyn Asker);
47
48impl GrowSet for GrowVia<'_> {
49    fn grow_set(&mut self, gib: u64) -> Result<String, String> {
50        grow_set(self.0, gib)
51    }
52    fn fetch(&mut self, f: &monetize_embed::fleet::Fetch) -> Result<Vec<u8>, String> {
53        monetize_estate_side::pump::fetch(self.0, f)
54    }
55}
56
57#[cfg(test)]
58mod tests {
59    use super::*;
60    use crate::exec::tests::{fact, Fake};
61    use crate::order::tests::{me, monetize_key, signed, MONETIZE_SEED};
62    use monetize_embed::fleet::{Ask, OrdersAfter, Report, SignedOrder};
63    use monetize_estate_side::pump::{Asked, BoxKey, Fault, Link, Puller, Signed};
64    use std::collections::BTreeMap;
65
66    /// Monetize in memory: a queue, and every call it heard.
67    #[derive(Default)]
68    struct Guardian {
69        queue: Vec<SignedOrder>,
70        heard: Vec<Signed>,
71    }
72
73    impl Link for Guardian {
74        fn report(&mut self, s: Signed) -> Result<u64, Fault> {
75            let r: Report = serde_json::from_str(&s.call.body).unwrap();
76            self.queue
77                .retain(|o| !r.acked.iter().any(|a| a.seq == o.seq));
78            self.heard.push(s);
79            Ok(30_000)
80        }
81        fn orders(&mut self, s: Signed) -> Result<Vec<SignedOrder>, Fault> {
82            let after: OrdersAfter = serde_json::from_str(&s.call.body).unwrap();
83            self.heard.push(s);
84            Ok(self
85                .queue
86                .iter()
87                .filter(|o| o.seq > after.after)
88                .cloned()
89                .collect())
90        }
91        fn ask(&mut self, s: Signed) -> Result<Asked, Fault> {
92            let _: Ask = serde_json::from_str(&s.call.body).unwrap();
93            self.heard.push(s);
94            Ok(Asked {
95                case_id: "c-1".into(),
96                words: "resized".into(),
97                url: String::new(),
98            })
99        }
100    }
101
102    fn mkey() -> [u8; 32] {
103        monetize_key().to_bytes()
104    }
105
106    /// ★ P4a: Holger's executor over the SHARED pump (monetize-estate-side): a fact is pulled,
107    /// run through Holger's verbs and acked, once.
108    #[test]
109    fn an_entitlement_is_pulled_run_and_acked_in_the_next_report_and_never_twice() {
110        let b = Fake::default();
111        let mut g = Guardian::default();
112        g.queue.push(signed(
113            4,
114            "main",
115            "entitlement",
116            &serde_json::to_string(&fact("libs", 5, MONETIZE_SEED)).unwrap(),
117            MONETIZE_SEED,
118        ));
119        let mut p = Puller::new(me("main"), BoxKey::from_seed([3u8; 32]), "test");
120        let pulled = p
121            .pull(&mut g, &Pumped(&b), Some(&mkey()), 0, 10_000)
122            .unwrap();
123        assert_eq!(pulled.acks.len(), 1);
124        assert!(pulled.acks[0].ok, "{pulled:?}");
125        assert!(b.facts.borrow().contains_key("libs"));
126        p.report(
127            &mut g,
128            BTreeMap::from([("serving".into(), "yes".into())]),
129            Some("{}".into()),
130            None,
131            10_001,
132        )
133        .unwrap();
134        let r: Report = serde_json::from_str(&g.heard.last().unwrap().call.body).unwrap();
135        assert_eq!(
136            r.acked.iter().map(|a| (a.seq, a.ok)).collect::<Vec<_>>(),
137            vec![(4, true)]
138        );
139        assert_eq!(r.product.as_deref(), Some("{}"));
140        assert!(g.queue.is_empty() && p.pulled.pending.is_empty());
141        assert!(
142            p.pull(&mut g, &Pumped(&b), Some(&mkey()), 0, 10_002)
143                .unwrap()
144                .acks
145                .is_empty(),
146            "nothing runs twice"
147        );
148    }
149
150    #[test]
151    fn a_forged_order_never_runs_never_acks_and_never_moves_the_sequence() {
152        let b = Fake::default();
153        let mut g = Guardian::default();
154        g.queue.push(signed(
155            u64::MAX,
156            "main",
157            "purge_tenant",
158            r#"{"tenant":"libs"}"#,
159            [9u8; 32],
160        ));
161        let mut p = Puller::new(me("main"), BoxKey::from_seed([3u8; 32]), "test");
162        let pulled = p
163            .pull(&mut g, &Pumped(&b), Some(&mkey()), 0, 10_000)
164            .unwrap();
165        assert!(pulled.acks.is_empty() && pulled.forged.len() == 1);
166        assert_eq!(p.pulled.last_seq, 0);
167        assert!(b.did.borrow().is_empty());
168    }
169
170    #[test]
171    fn without_a_monetize_key_nothing_is_pulled_and_monetize_is_not_even_asked() {
172        let mut g = Guardian::default();
173        let mut p = Puller::new(me("main"), BoxKey::from_seed([3u8; 32]), "test");
174        assert_eq!(
175            p.pull(&mut g, &Pumped(&Fake::default()), None, 0, 10_000)
176                .unwrap_err()
177                .code,
178            "no_key"
179        );
180        assert!(g.heard.is_empty());
181    }
182
183    #[test]
184    fn the_first_call_carries_the_box_key_and_later_ones_do_not() {
185        let mut g = Guardian::default();
186        let mut p = Puller::new(me("main"), BoxKey::from_seed([3u8; 32]), "test");
187        p.report(&mut g, BTreeMap::new(), None, None, 1).unwrap();
188        p.report(&mut g, BTreeMap::new(), None, None, 2).unwrap();
189        assert_eq!(g.heard[0].box_key, p.box_public().to_vec());
190        assert!(g.heard[1].box_key.is_empty());
191    }
192
193    #[test]
194    fn a_twins_own_review_rides_every_report_and_an_ordered_one_the_next_only() {
195        let mut g = Guardian::default();
196        let mut p = Puller::new(me("twin"), BoxKey::from_seed([4u8; 32]), "test");
197        let b = Fake {
198            twin: true,
199            ..Default::default()
200        };
201        g.queue
202            .push(signed(1, "twin", "review_twin", "{}", MONETIZE_SEED));
203        p.pull(&mut g, &Pumped(&b), Some(&mkey()), 0, 10).unwrap();
204        p.report(&mut g, BTreeMap::new(), None, None, 11).unwrap();
205        p.report(&mut g, BTreeMap::new(), None, None, 12).unwrap();
206        p.report(
207            &mut g,
208            BTreeMap::new(),
209            None,
210            Some(r#"{"own":1}"#.into()),
211            13,
212        )
213        .unwrap();
214        let said: Vec<Option<String>> = g
215            .heard
216            .iter()
217            .filter(|s| s.call.call == "report")
218            .map(|s| {
219                serde_json::from_str::<Report>(&s.call.body)
220                    .unwrap()
221                    .twin_review
222            })
223            .collect();
224        assert!(
225            said[0].as_deref().is_some_and(|r| r.contains("complete")),
226            "{said:?}"
227        );
228        assert_eq!(said[1], None);
229        assert_eq!(said[2].as_deref(), Some(r#"{"own":1}"#));
230    }
231
232    #[test]
233    fn the_puller_remembers_what_it_took_across_a_restart() {
234        let dir = tempfile::tempdir().unwrap();
235        let mut g = Guardian::default();
236        g.queue.push(signed(
237            6,
238            "main",
239            "grow_room",
240            r#"{"need_bytes":5,"reason":"r"}"#,
241            MONETIZE_SEED,
242        ));
243        let mut p = Puller::at(dir.path(), me("main"), "test").unwrap();
244        let key = p.box_public();
245        p.pull(&mut g, &Pumped(&Fake::default()), Some(&mkey()), 0, 10_000)
246            .unwrap();
247        drop(p);
248        let p = Puller::at(dir.path(), me("main"), "test").unwrap();
249        assert_eq!(
250            (p.box_public(), p.pulled.last_seq, p.pulled.pending.len()),
251            (key, 6, 1)
252        );
253    }
254}
holger_fleet/pumped.rs

holger_fleet/
pumped.rs
