1use std::collections::BTreeMap;
9use std::path::{Path, PathBuf};
10
11use monetize_embed::fleet::{ask, call, kind, order_message, Ask, FleetCall, OrderAck, OrdersAfter, Report, SignedOrder, LONG_POLL_MAX_MS};
12use monetize_embed::VerifyingKey;
13use serde::{Deserialize, Serialize};
14
15use super::key::BoxKey;
16
17pub const STATE_REL: &str = "fleet/pulled.json";
19
20#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
22pub struct Addressee {
23 pub product: String,
24 pub instance: String,
25 pub role: String,
27}
28
29#[derive(Debug, Clone, PartialEq, Eq)]
31pub struct Fault {
32 pub code: String,
33 pub message: String,
34}
35
36impl Fault {
37 pub fn new(code: impl Into<String>, message: impl Into<String>) -> Fault {
38 Fault { code: code.into(), message: message.into() }
39 }
40}
41
42impl std::fmt::Display for Fault {
43 fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
44 write!(f, "{}: {}", self.code, self.message)
45 }
46}
47
48#[derive(Debug, Clone)]
50pub struct Signed {
51 pub call: FleetCall,
52 pub signature: Vec<u8>,
53 pub box_key: Vec<u8>,
55}
56
57#[derive(Debug, Clone, Default, PartialEq, Eq)]
60pub struct Asked {
61 pub case_id: String,
62 pub words: String,
63 pub url: String,
64}
65
66pub trait Link {
68 fn report(&mut self, s: Signed) -> Result<u64, Fault>;
70 fn orders(&mut self, s: Signed) -> Result<Vec<SignedOrder>, Fault>;
71 fn ask(&mut self, s: Signed) -> Result<Asked, Fault>;
72}
73
74pub trait Asker {
77 fn ask(&mut self, a: &Ask) -> Result<Asked, Fault>;
78}
79
80pub fn grow_set(asker: &mut dyn Asker, gib: u64) -> Result<String, String> {
83 let a = Ask { kind: ask::GROW_SET.into(), body: serde_json::json!({ "members": [], "gib": gib }) };
84 let Asked { case_id: case, words, .. } = asker.ask(&a).map_err(|f| f.to_string())?;
85 Ok(if words.is_empty() { case } else { words })
86}
87
88pub struct Ctx<'a> {
90 pub me: &'a Addressee,
91 pub key: &'a VerifyingKey,
93 pub now_unix_ms: u64,
94}
95
96pub trait PumpBox {
98 fn health(&self) -> BTreeMap<String, String> {
100 BTreeMap::new()
101 }
102 fn product_report(&self) -> Option<String> {
104 None
105 }
106 fn own_review(&self) -> Option<String> {
108 None
109 }
110 fn run(&self, ctx: &Ctx, order: &SignedOrder, ask: &mut dyn Asker) -> OrderAck;
115}
116
117pub use monetize_embed::fleet::FUTURE_SKEW_MS;
120
121#[derive(Debug, Clone, PartialEq, Eq)]
123pub enum Checked {
124 Forged(String),
126 Refused(String),
128 Fresh,
130}
131
132pub fn check(o: &SignedOrder, me: &Addressee, key: &VerifyingKey, last_seq: u64, now_unix_ms: u64) -> Checked {
139 let signed = ed25519_dalek::Signature::from_slice(&o.signature).map(|s| key.verify_strict(&order_message(o), &s).is_ok()).unwrap_or(false);
141 if !signed {
142 return Checked::Forged(format!("order {} ({}) is not signed by this product's monetize key", o.seq, o.kind));
143 }
144 if o.product != me.product || o.instance != me.instance || o.role != me.role {
145 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));
146 }
147 if o.seq <= last_seq {
148 return Checked::Refused(format!("order {} is not newer than {last_seq}, the last one this box took: a replay", o.seq));
149 }
150 if o.not_after_unix_ms != 0 && now_unix_ms > o.not_after_unix_ms {
151 return Checked::Refused(format!("order {} expired {} s ago", o.seq, (now_unix_ms - o.not_after_unix_ms) / 1000));
152 }
153 if o.issued_unix_ms > now_unix_ms.saturating_add(FUTURE_SKEW_MS) {
154 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));
155 }
156 Checked::Fresh
157}
158
159#[derive(Debug, Clone, Default, PartialEq, Eq, Serialize, Deserialize)]
161#[serde(default)]
162pub struct Pulled {
163 pub last_seq: u64,
165 pub pending: Vec<OrderAck>,
167 pub twin_review: Option<String>,
169}
170
171#[derive(Debug, Clone, Default, PartialEq, Eq)]
173pub struct Pull {
174 pub acks: Vec<OrderAck>,
175 pub forged: Vec<String>,
176}
177
178pub struct Puller {
179 pub me: Addressee,
180 pub build: String,
181 key: BoxKey,
182 state: Option<PathBuf>,
183 pub pulled: Pulled,
184 heard: bool,
185}
186
187impl Puller {
188 pub fn new(me: Addressee, key: BoxKey, build: impl Into<String>) -> Puller {
190 Puller { me, build: build.into(), key, state: None, pulled: Pulled::default(), heard: false }
191 }
192
193 pub fn at(root: &Path, me: Addressee, build: impl Into<String>) -> std::io::Result<Puller> {
195 let key = BoxKey::load_or_mint(root)?;
196 let state = root.join(STATE_REL);
197 let pulled = match std::fs::read(&state) {
198 Ok(bytes) => serde_json::from_slice(&bytes).map_err(|e| std::io::Error::new(std::io::ErrorKind::InvalidData, format!("{}: {e}", state.display())))?,
199 Err(e) if e.kind() == std::io::ErrorKind::NotFound => Pulled::default(),
200 Err(e) => return Err(e),
201 };
202 Ok(Puller { me, build: build.into(), key, state: Some(state), pulled, heard: false })
203 }
204
205 pub fn box_public(&self) -> [u8; 32] {
206 self.key.public()
207 }
208
209 fn save(&self) -> Result<(), Fault> {
210 let Some(path) = &self.state else { return Ok(()) };
211 let bytes = serde_json::to_vec(&self.pulled).map_err(|e| Fault::new("internal", e.to_string()))?;
212 crate::write_atomic(path, &bytes, false).map_err(|e| Fault::new("io", format!("{}: {e}", path.display())))
213 }
214
215 fn signed(&self, which: &str, body: String, now_unix_ms: u64) -> Signed {
216 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 };
217 let signature = self.key.sign(&c);
218 Signed { call: c, signature, box_key: if self.heard { Vec::new() } else { self.key.public().to_vec() } }
219 }
220
221 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> {
225 let sent = self.pulled.pending.len();
226 let report = Report {
227 build: self.build.clone(),
228 health,
229 acked: self.pulled.pending.clone(),
230 twin_review: own_review.or_else(|| self.pulled.twin_review.clone()),
231 product,
232 measured_unix_ms: now_unix_ms,
233 };
234 let body = serde_json::to_string(&report).map_err(|e| Fault::new("internal", e.to_string()))?;
235 let next = link.report(self.signed(call::REPORT, body, now_unix_ms))?;
236 self.heard = true;
237 self.pulled.pending.drain(..sent);
238 self.pulled.twin_review = None;
239 self.save()?;
240 Ok(next)
241 }
242
243 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> {
247 let Some(raw) = monetize_pubkey else {
248 return Err(Fault::new("no_key", "this box has no monetize key yet, so no order could be verified: none is pulled"));
249 };
250 let key = VerifyingKey::from_bytes(raw).map_err(|e| Fault::new("no_key", format!("the monetize key is not an Ed25519 point: {e}")))?;
251 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()))?;
252 let mut orders = link.orders(self.signed(call::ORDERS, body, now_unix_ms))?;
253 self.heard = true;
254 orders.sort_by_key(|o| o.seq);
255 let mut out = Pull::default();
256 for o in orders {
257 let checked = check(&o, &self.me, &key, self.pulled.last_seq, now_unix_ms);
258 if let Checked::Forged(why) = checked {
259 out.forged.push(why);
260 continue;
261 }
262 if o.seq <= self.pulled.last_seq {
263 continue;
264 }
265 self.pulled.last_seq = o.seq;
266 self.save()?;
267 let ack = match checked {
268 Checked::Fresh => {
269 let me = self.me.clone();
270 let ctx = Ctx { me: &me, key: &key, now_unix_ms };
271 let ack = {
272 let mut via = AskVia { puller: &mut *self, link: &mut *link, now_unix_ms };
273 b.run(&ctx, &o, &mut via)
274 };
275 if o.kind == kind::REVIEW_TWIN && ack.ok {
276 self.pulled.twin_review = Some(ack.receipt_json.clone());
277 }
278 ack
279 }
280 Checked::Refused(detail) => OrderAck { seq: o.seq, ok: false, detail, receipt_json: String::new() },
281 Checked::Forged(_) => unreachable!("forged orders were skipped"),
282 };
283 self.pulled.pending.push(ack.clone());
284 self.save()?;
285 out.acks.push(ack);
286 }
287 Ok(out)
288 }
289
290 pub fn ask(&mut self, link: &mut dyn Link, a: &Ask, now_unix_ms: u64) -> Result<Asked, Fault> {
292 let body = serde_json::to_string(a).map_err(|e| Fault::new("internal", e.to_string()))?;
293 let answer = link.ask(self.signed(call::ASK, body, now_unix_ms))?;
294 self.heard = true;
295 Ok(answer)
296 }
297}
298
299struct AskVia<'a> {
301 puller: &'a mut Puller,
302 link: &'a mut dyn Link,
303 now_unix_ms: u64,
304}
305
306impl Asker for AskVia<'_> {
307 fn ask(&mut self, a: &Ask) -> Result<Asked, Fault> {
308 self.puller.ask(self.link, a, self.now_unix_ms)
309 }
310}
311
312#[cfg(test)]
313pub(crate) mod tests {
314 use super::*;
315 use ed25519_dalek::Signer as _;
316 use std::cell::RefCell;
317
318 pub const MONETIZE_SEED: [u8; 32] = [7u8; 32];
319
320 pub fn me(role: &str) -> Addressee {
321 Addressee { product: "a-product".into(), instance: "acme".into(), role: role.into() }
322 }
323
324 pub fn monetize_key() -> VerifyingKey {
325 ed25519_dalek::SigningKey::from_bytes(&MONETIZE_SEED).verifying_key()
326 }
327
328 fn mkey() -> [u8; 32] {
329 monetize_key().to_bytes()
330 }
331
332 pub fn signed(seq: u64, role: &str, kind: &str, body: &str, seed: [u8; 32]) -> SignedOrder {
334 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() };
335 o.signature = ed25519_dalek::SigningKey::from_bytes(&seed).sign(&order_message(&o)).to_bytes().to_vec();
336 o
337 }
338
339 #[derive(Default)]
341 pub struct Guardian {
342 pub queue: Vec<SignedOrder>,
343 pub heard: Vec<Signed>,
344 }
345
346 impl Link for Guardian {
347 fn report(&mut self, s: Signed) -> Result<u64, Fault> {
348 let r: Report = serde_json::from_str(&s.call.body).unwrap();
349 self.queue.retain(|o| !r.acked.iter().any(|a| a.seq == o.seq));
350 self.heard.push(s);
351 Ok(30_000)
352 }
353 fn orders(&mut self, s: Signed) -> Result<Vec<SignedOrder>, Fault> {
354 let after: OrdersAfter = serde_json::from_str(&s.call.body).unwrap();
355 self.heard.push(s);
356 Ok(self.queue.iter().filter(|o| o.seq > after.after).cloned().collect())
357 }
358 fn ask(&mut self, s: Signed) -> Result<Asked, Fault> {
359 let _: Ask = serde_json::from_str(&s.call.body).unwrap();
360 self.heard.push(s);
361 Ok(Asked { case_id: "c-1".into(), words: "resized".into(), url: String::new() })
362 }
363 }
364
365 #[derive(Default)]
368 pub struct Fake {
369 pub ran: RefCell<Vec<(u64, String)>>,
370 }
371
372 impl PumpBox for Fake {
373 fn run(&self, ctx: &Ctx, o: &SignedOrder, ask: &mut dyn Asker) -> OrderAck {
374 self.ran.borrow_mut().push((o.seq, o.kind.clone()));
375 let said = match o.kind.as_str() {
376 "note" => Ok(format!("noted for {}", ctx.me.role)),
377 "grow" => grow_set(ask, 64),
378 kind::REVIEW_TWIN => return OrderAck { seq: o.seq, ok: true, detail: "reviewed".into(), receipt_json: r#"{"complete":"yes"}"#.into() },
379 other => Err(format!("order {}: this product does not know the order kind {other:?}", o.seq)),
380 };
381 match said {
382 Ok(detail) => OrderAck { seq: o.seq, ok: true, detail, receipt_json: String::new() },
383 Err(detail) => OrderAck { seq: o.seq, ok: false, detail, receipt_json: String::new() },
384 }
385 }
386 }
387
388 #[test]
389 fn an_order_is_pulled_run_and_acked_in_the_next_report_and_never_twice() {
390 let b = Fake::default();
391 let mut g = Guardian::default();
392 g.queue.push(signed(4, "main", "note", "{}", MONETIZE_SEED));
393 let mut p = Puller::new(me("main"), BoxKey::from_seed([3u8; 32]), "test");
394 let pulled = p.pull(&mut g, &b, Some(&mkey()), 0, 10_000).unwrap();
395 assert_eq!(pulled.acks.len(), 1);
396 assert!(pulled.acks[0].ok, "{pulled:?}");
397 assert_eq!(pulled.acks[0].detail, "noted for main");
398 p.report(&mut g, BTreeMap::from([("serving".into(), "yes".into())]), Some("{}".into()), None, 10_001).unwrap();
399 let r: Report = serde_json::from_str(&g.heard.last().unwrap().call.body).unwrap();
400 assert_eq!(r.acked.iter().map(|a| (a.seq, a.ok)).collect::<Vec<_>>(), vec![(4, true)]);
401 assert_eq!((r.product.as_deref(), r.build.as_str()), (Some("{}"), "test"));
402 assert!(g.queue.is_empty() && p.pulled.pending.is_empty());
403 assert!(p.pull(&mut g, &b, Some(&mkey()), 0, 10_002).unwrap().acks.is_empty(), "nothing runs twice");
404 assert_eq!(b.ran.borrow().len(), 1);
405 }
406
407 #[test]
412 fn a_weak_key_signs_no_order_and_a_real_one_still_does() {
413 let mut identity = [0u8; 32];
414 identity[0] = 1;
415 let weak = VerifyingKey::from_bytes(&identity).expect("the identity point decompresses");
416 let mut o = signed(3, "main", "recharge_from_twin", "{}", MONETIZE_SEED);
417 o.signature = { let mut s = vec![0u8; 64]; s[0] = 1; s };
418 assert!(matches!(check(&o, &me("main"), &weak, 0, 9_500), Checked::Forged(_)), "a weak key verifies nothing");
419 let real = signed(3, "main", "recharge_from_twin", "{}", MONETIZE_SEED);
420 assert_eq!(check(&real, &me("main"), &monetize_key(), 0, 9_500), Checked::Fresh);
421 }
422
423 #[test]
426 fn an_order_dated_ahead_of_the_box_clock_is_refused_by_name() {
427 let mut ahead = signed(3, "main", "note", "{}", MONETIZE_SEED);
428 ahead.issued_unix_ms = 9_000 + FUTURE_SKEW_MS + 1;
429 ahead.signature = ed25519_dalek::SigningKey::from_bytes(&MONETIZE_SEED).sign(&order_message(&ahead)).to_bytes().to_vec();
430 match check(&ahead, &me("main"), &monetize_key(), 0, 9_000) {
431 Checked::Refused(why) => assert!(why.contains("in the future"), "{why}"),
432 other => panic!("{other:?}"),
433 }
434 let mut skew = ahead.clone();
435 skew.issued_unix_ms = 9_000 + FUTURE_SKEW_MS;
436 skew.signature = ed25519_dalek::SigningKey::from_bytes(&MONETIZE_SEED).sign(&order_message(&skew)).to_bytes().to_vec();
437 assert_eq!(check(&skew, &me("main"), &monetize_key(), 0, 9_000), Checked::Fresh);
438 }
439
440 #[test]
441 fn a_forged_order_never_runs_never_acks_and_never_moves_the_sequence() {
442 let b = Fake::default();
443 let mut g = Guardian::default();
444 g.queue.push(signed(u64::MAX, "main", "note", "{}", [9u8; 32]));
445 let mut p = Puller::new(me("main"), BoxKey::from_seed([3u8; 32]), "test");
446 let pulled = p.pull(&mut g, &b, Some(&mkey()), 0, 10_000).unwrap();
447 assert!(pulled.acks.is_empty() && pulled.forged.len() == 1);
448 assert_eq!(p.pulled.last_seq, 0);
449 assert!(b.ran.borrow().is_empty());
450 }
451
452 #[test]
456 fn another_boxs_an_expired_or_an_unknown_order_is_refused_by_name() {
457 let k = monetize_key();
458 let twin = signed(5, "twin", "note", "{}", MONETIZE_SEED);
459 assert!(matches!(check(&twin, &me("main"), &k, 0, 10_000), Checked::Refused(w) if w.contains("for a-product/acme/twin")));
460 let mut late = signed(5, "main", "note", "{}", MONETIZE_SEED);
461 late.not_after_unix_ms = 1_000;
462 let late = { let mut l = late.clone(); l.signature = signed_like(&l); l };
463 assert!(matches!(check(&late, &me("main"), &k, 0, 61_000), Checked::Refused(w) if w.contains("expired 60 s ago")));
464 assert!(matches!(check(&signed(5, "main", "note", "{}", MONETIZE_SEED), &me("main"), &k, 5, 1), Checked::Refused(w) if w.contains("a replay")));
465 assert_eq!(check(&signed(6, "main", "note", "{}", MONETIZE_SEED), &me("main"), &k, 5, 1), Checked::Fresh);
466 let b = Fake::default();
467 let mut g = Guardian::default();
468 g.queue.push(twin);
469 g.queue.push(signed(7, "main", "reboot_the_moon", "{}", MONETIZE_SEED));
470 let mut p = Puller::new(me("main"), BoxKey::from_seed([3u8; 32]), "test");
471 let pulled = p.pull(&mut g, &b, Some(&mkey()), 0, 10_000).unwrap();
472 assert_eq!(pulled.acks.iter().map(|a| (a.seq, a.ok)).collect::<Vec<_>>(), vec![(5, false), (7, false)]);
473 assert!(pulled.acks[1].detail.contains("does not know the order kind \"reboot_the_moon\""), "{pulled:?}");
474 assert_eq!(*b.ran.borrow(), vec![(7, "reboot_the_moon".to_string())], "only the box's own order reaches the product");
475 }
476
477 fn signed_like(o: &SignedOrder) -> Vec<u8> {
478 ed25519_dalek::SigningKey::from_bytes(&MONETIZE_SEED).sign(&order_message(o)).to_bytes().to_vec()
479 }
480
481 #[test]
482 fn without_a_monetize_key_nothing_is_pulled_and_monetize_is_not_even_asked() {
483 let mut g = Guardian::default();
484 let mut p = Puller::new(me("main"), BoxKey::from_seed([3u8; 32]), "test");
485 assert_eq!(p.pull(&mut g, &Fake::default(), None, 0, 10_000).unwrap_err().code, "no_key");
486 assert!(g.heard.is_empty());
487 }
488
489 #[test]
490 fn the_first_call_carries_the_box_key_and_later_ones_do_not() {
491 let mut g = Guardian::default();
492 let mut p = Puller::new(me("main"), BoxKey::from_seed([3u8; 32]), "test");
493 p.report(&mut g, BTreeMap::new(), None, None, 1).unwrap();
494 p.report(&mut g, BTreeMap::new(), None, None, 2).unwrap();
495 assert_eq!(g.heard[0].box_key, p.box_public().to_vec());
496 assert!(g.heard[1].box_key.is_empty());
497 }
498
499 #[test]
500 fn a_twins_own_review_rides_every_report_and_an_ordered_one_the_next_only() {
501 let mut g = Guardian::default();
502 let mut p = Puller::new(me("twin"), BoxKey::from_seed([4u8; 32]), "test");
503 g.queue.push(signed(1, "twin", "review_twin", "{}", MONETIZE_SEED));
504 p.pull(&mut g, &Fake::default(), Some(&mkey()), 0, 10).unwrap();
505 p.report(&mut g, BTreeMap::new(), None, None, 11).unwrap();
506 p.report(&mut g, BTreeMap::new(), None, None, 12).unwrap();
507 p.report(&mut g, BTreeMap::new(), None, Some(r#"{"own":1}"#.into()), 13).unwrap();
508 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();
509 assert!(said[0].as_deref().is_some_and(|r| r.contains("complete")), "{said:?}");
510 assert_eq!(said[1], None);
511 assert_eq!(said[2].as_deref(), Some(r#"{"own":1}"#));
512 }
513
514 #[test]
516 fn an_order_asks_monetize_on_its_own_link_signed_by_the_box() {
517 let mut g = Guardian::default();
518 g.queue.push(signed(2, "main", "grow", "{}", MONETIZE_SEED));
519 let mut p = Puller::new(me("main"), BoxKey::from_seed([3u8; 32]), "test");
520 let pulled = p.pull(&mut g, &Fake::default(), Some(&mkey()), 0, 10_000).unwrap();
521 assert_eq!(pulled.acks[0].detail, "resized");
522 let asked = g.heard.iter().find(|s| s.call.call == "ask").expect("asked");
523 let a: Ask = serde_json::from_str(&asked.call.body).unwrap();
524 assert_eq!((a.kind.as_str(), a.body["gib"].as_u64()), ("grow_set", Some(64)));
525 let vk = VerifyingKey::from_bytes(&p.box_public()).unwrap();
526 monetize_embed::fleet::verify_fleet_call(&asked.call, &asked.signature, &vk, 10_000).expect("the ask is the box's");
527 }
528
529 #[test]
530 fn the_puller_remembers_what_it_took_across_a_restart() {
531 let dir = tempfile::tempdir().unwrap();
532 let mut g = Guardian::default();
533 g.queue.push(signed(6, "main", "note", "{}", MONETIZE_SEED));
534 let mut p = Puller::at(dir.path(), me("main"), "test").unwrap();
535 let key = p.box_public();
536 p.pull(&mut g, &Fake::default(), Some(&mkey()), 0, 10_000).unwrap();
537 drop(p);
538 let p = Puller::at(dir.path(), me("main"), "test").unwrap();
539 assert_eq!((p.box_public(), p.pulled.last_seq, p.pulled.pending.len()), (key, 6, 1));
540 }
541}