crates/gunnar-repair/src/fleet.rs
//! **The estate side of monetize's pull channel** (`monetize.v1.Fleet`, 5.4 G1-pull), on the
//! SHARED box pump.
//!
//! Estates call monetize; monetize never calls an estate. Every box of an estate REPORTS (its
//! arrival is liveness, its acks close orders), pulls the ORDERS queued for it (a long-poll)
//! and may ASK for a remedy. All of that is monetize-estate-side's [`pump`] — the box key
//! (`<data>/fleet/box.ed25519`), the signed calls, the order check under the monetize key
//! ([`check`]), the puller and what it keeps (`fleet/pulled.json`), and the wire client
//! ([`TonicLink`] over monetize-proto's [`pb`]) — the same pump Holger's boxes run (P4a).
//!
//! What is gunnar's here is only its executor: [`GunnarBox`] is the pump's [`PumpBox`]. It
//! reads a checked order's body ([`crate::decode`]) and runs it through
//! [`crate::exec::execute`] — the same verbs the boot check and the local Twin panel run — or,
//! for a pulled growth, [`crate::exec::grow`] with `Fleet.Ask{grow_set}` on the order's own
//! link.
//!
//! Until 2026-10-26 (A28 M12) this module was a second copy of all of it, over a vendored
//! `proto/monetize/v1/fleet.proto` that had drifted from monetize's. monetize 0.3.0 on
//! crates.io (H20) ended that.
use std::sync::Arc;
pub use monetize_embed::fleet::{
call, kind, Ask, FleetCall, OrderAck as WireAck, OrdersAfter, Report, SignedOrder as WireOrder,
LONG_POLL_MAX_MS,
};
pub use monetize_estate_side::pump::grpc::{order_of, TonicLink};
pub use monetize_estate_side::pump::key::BOX_KEY_REL;
pub use monetize_estate_side::pump::puller::STATE_REL;
pub use monetize_estate_side::pump::{
self, check, grow_set, Addressee, Asked, Asker, BoxKey, Checked, Ctx, Fault, Link, Pull,
Pulled, Puller, PumpBox, Signed,
};
/// The wire: monetize's own generated `monetize.v1` (client and server).
pub use monetize_proto as pb;
use crate::exec::{execute, grow, Estate, GrowSet};
/// This product's name on the wire.
pub const PRODUCT: &str = "gunnar";
/// **Gunnar's half of the shared pump**: the estate an order runs against, and how a host
/// says it is stopping (a recharge between chunks looks at it).
///
/// [`PumpBox::run`] is called on the puller's (blocking) thread. Gunnar's verbs are async
/// over tonic, so `run` drives them to the end on that same thread inside `rt`'s context
/// ([`block_on`]); the pump's own wire calls ([`TonicLink`]) are made from the same thread
/// too, so a pulled growth's `Fleet.Ask` never nests one runtime `block_on` in another.
pub struct GunnarBox<E> {
pub estate: E,
pub rt: tokio::runtime::Handle,
pub spent: Arc<dyn Fn() -> bool + Send + Sync>,
}
impl<E> GunnarBox<E> {
pub fn new(estate: E, rt: tokio::runtime::Handle) -> GunnarBox<E> {
GunnarBox {
estate,
rt,
spent: Arc::new(|| false),
}
}
/// [`GunnarBox::new`] that stops a long verb once `spent` says so.
pub fn stopping(mut self, spent: Arc<dyn Fn() -> bool + Send + Sync>) -> GunnarBox<E> {
self.spent = spent;
self
}
}
impl<E: Estate> PumpBox for GunnarBox<E> {
fn run(&self, _ctx: &Ctx, o: &WireOrder, ask: &mut dyn Asker) -> WireAck {
// The pump checked the signature, the addressee, the sequence and the window
// ([`check`]); what is left is the body, which is gunnar's.
let order = match crate::decode(o.seq, &o.kind, o.body.as_bytes()) {
Ok(order) => order,
Err(refusal) => {
return WireAck {
seq: o.seq,
ok: false,
detail: refusal.to_string(),
receipt_json: String::new(),
}
}
};
let _in_rt = self.rt.enter();
let ack = block_on(async {
match &order {
crate::Order::GrowGoAhead { go_ahead, data_gib } => {
grow(&self.estate, o.seq, go_ahead, *data_gib, &mut AskOn(ask)).await
}
order => execute(&self.estate, o.seq, order, &*self.spent).await,
}
});
WireAck {
seq: ack.seq,
ok: ack.ok,
detail: ack.detail,
receipt_json: ack.receipt_json,
}
}
}
/// A pulled growth's `Fleet.Ask{grow_set}`, through the pump's own asker ([`grow_set`]).
struct AskOn<'a>(&'a mut dyn Asker);
impl GrowSet for AskOn<'_> {
async fn grow_set(&mut self, gib: u64) -> Result<String, crate::Fault> {
// The shared asker words a fault as `code: message` (its `Display`); gunnar's grow
// says the two apart.
grow_set(self.0, gib).map_err(|said| match said.split_once(": ") {
Some((code, message)) => crate::Fault::new(code, message),
None => crate::Fault::new("error", said),
})
}
}
/// **Drive a future to its end on this thread**, parking between polls. For the pump's
/// thread only: it is never a runtime worker, so waiting here blocks no task.
pub fn block_on<F: std::future::Future>(f: F) -> F::Output {
struct Unpark(std::thread::Thread);
impl std::task::Wake for Unpark {
fn wake(self: Arc<Self>) {
self.0.unpark();
}
}
let waker = std::task::Waker::from(Arc::new(Unpark(std::thread::current())));
let mut cx = std::task::Context::from_waker(&waker);
let mut f = std::pin::pin!(f);
loop {
if let std::task::Poll::Ready(v) = f.as_mut().poll(&mut cx) {
return v;
}
std::thread::park();
}
}
#[cfg(test)]
mod tests {
use super::*;
use crate::exec::tests::{lost_main, Growing, Ref};
use ed25519_dalek::Signer;
use std::collections::BTreeMap;
const MONETIZE_SEED: [u8; 32] = [7u8; 32];
fn me() -> Addressee {
Addressee {
product: PRODUCT.into(),
instance: "acme".into(),
role: "main".into(),
}
}
fn monetize_pub() -> [u8; 32] {
ed25519_dalek::SigningKey::from_bytes(&MONETIZE_SEED)
.verifying_key()
.to_bytes()
}
fn order(seq: u64, role: &str, kind: &str, body: &str, signer: [u8; 32]) -> WireOrder {
let mut o = WireOrder {
seq,
product: 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(&signer)
.sign(&monetize_embed::fleet::order_message(&o))
.to_bytes()
.to_vec();
o
}
/// monetize, in memory: a queue, and every call it heard.
#[derive(Default)]
struct Guardian {
queue: Vec<WireOrder>,
heard: Vec<Signed>,
deaf_ask: bool,
}
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<WireOrder>, 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> {
self.heard.push(s);
if self.deaf_ask {
return Err(Fault::new("unavailable", "no answer from the Ask door"));
}
Ok(Asked {
case_id: "c-1".into(),
words: "opened".into(),
url: String::new(),
})
}
}
fn rt() -> tokio::runtime::Runtime {
tokio::runtime::Builder::new_current_thread()
.enable_all()
.build()
.unwrap()
}
/// ★ M12: **gunnar's pull channel IS the shared pump** — a queued recharge is pulled,
/// checked under the monetize key by the pump, run through gunnar's ONE executor over
/// [`GunnarBox`], and acked in the next report; it never runs twice.
#[test]
fn a_queued_recharge_order_is_pulled_checked_by_the_shared_pump_run_by_gunnar_and_acked() {
let rt = rt();
let w = lost_main();
let b = GunnarBox::new(Ref(&w), rt.handle().clone());
let mut g = Guardian::default();
g.queue.push(order(4, "main", "recharge_from_twin", r#"{"main":"main:1","twin":"twin:1","epoch":5,"case":"c-9","fenced":"upcloud: the old main's server was given back and read back gone (case c-9)"}"#, MONETIZE_SEED));
let mut p = Puller::new(me(), BoxKey::from_seed([3u8; 32]), "test");
let pulled = p
.pull(&mut g, &b, Some(&monetize_pub()), 0, 10_000)
.unwrap();
assert_eq!(pulled.acks.len(), 1, "{pulled:?}");
assert!(pulled.acks[0].ok, "{pulled:?}");
assert_eq!(p.pulled.last_seq, 4);
assert_eq!(
w.main.got.lock().unwrap().len(),
5,
"the twin's five records were replayed into the main"
);
p.report(
&mut g,
BTreeMap::from([("data".into(), "ok".into())]),
None,
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!(g.queue.is_empty(), "the ack took the order off the queue");
let again = p
.pull(&mut g, &b, Some(&monetize_pub()), 0, 10_002)
.unwrap();
assert!(again.acks.is_empty(), "nothing runs twice");
}
/// G16-d over the shared pump: a pulled `grow_go_ahead` asks monetize for THIS box's own
/// disks with a signed `Fleet.Ask{grow_set}` on the same link, once `/data` is off; an Ask
/// door that does not answer mounts `/data` back and the ack says why.
#[test]
fn a_pulled_grow_go_ahead_asks_monetize_on_the_pumps_own_link() {
let rt = rt();
let w = lost_main();
let b = GunnarBox::new(Growing::new(&w, false, false), rt.handle().clone());
let mut g = Guardian::default();
g.queue.push(order(
6,
"main",
"grow_go_ahead",
r#"{"go_ahead":"{signed}","data_gib":120}"#,
MONETIZE_SEED,
));
let mut p = Puller::new(me(), BoxKey::from_seed([3u8; 32]), "test");
let pulled = p
.pull(&mut g, &b, Some(&monetize_pub()), 0, 10_000)
.unwrap();
assert!(pulled.acks[0].ok, "{pulled:?}");
assert_eq!(
b.estate.did(),
[
"check",
"twin_holds 120",
"start",
"settle false",
"look_again false",
"settle true"
]
);
let asked: Vec<&Signed> = g
.heard
.iter()
.filter(|s| s.call.call == call::ASK)
.collect();
assert_eq!(asked.len(), 1, "one ask");
let a: Ask = serde_json::from_str(&asked[0].call.body).unwrap();
assert_eq!(
(a.kind.as_str(), a.body.clone()),
("grow_set", serde_json::json!({"members": [], "gib": 120}))
);
let vk = monetize_embed::VerifyingKey::from_bytes(&p.box_public()).unwrap();
monetize_embed::fleet::verify_fleet_call(
&asked[0].call,
&asked[0].signature,
&vk,
asked[0].call.at_unix_ms,
)
.expect("signed by the box key");
let b = GunnarBox::new(Growing::new(&w, false, false), rt.handle().clone());
*b.estate.settle.lock().unwrap() = [
crate::exec::GrowState::WaitingMembers("waiting-members".into()),
crate::exec::GrowState::Failed("gave up: mounted back".into()),
]
.into();
let mut deaf = Guardian {
deaf_ask: true,
..Guardian::default()
};
deaf.queue.push(order(
7,
"main",
"grow_go_ahead",
r#"{"go_ahead":"{signed}","data_gib":120}"#,
MONETIZE_SEED,
));
let pulled = p
.pull(&mut deaf, &b, Some(&monetize_pub()), 0, 10_001)
.unwrap();
let d = &pulled.acks[0].detail;
assert!(
!pulled.acks[0].ok
&& d.contains("(unavailable: no answer from the Ask door)")
&& d.contains("mounted back"),
"{d}"
);
assert_eq!(
b.estate.did()[3..],
["settle false", "look_again true", "settle true"]
);
}
/// The pump's check stands before gunnar's body: a forgery never runs and never moves the
/// sequence; an order for the twin is refused by the main; a body gunnar does not know is
/// acked not-ok by name.
#[test]
fn the_shared_check_stands_before_gunnars_executor() {
let rt = rt();
let w = lost_main();
let b = GunnarBox::new(Ref(&w), rt.handle().clone());
let mut p = Puller::new(me(), BoxKey::from_seed([3u8; 32]), "test");
let mut g = Guardian::default();
g.queue.push(order(
u64::MAX,
"main",
"recharge_from_twin",
r#"{"main":"main:1","twin":"twin:1"}"#,
[9u8; 32],
));
let pulled = p
.pull(&mut g, &b, Some(&monetize_pub()), 0, 10_000)
.unwrap();
assert!(
pulled.acks.is_empty() && pulled.forged.len() == 1 && p.pulled.last_seq == 0,
"{pulled:?}"
);
let mut g = Guardian::default();
g.queue.push(order(
2,
"twin",
"recharge_from_twin",
r#"{"main":"main:1","twin":"twin:1"}"#,
MONETIZE_SEED,
));
g.queue
.push(order(3, "main", "no_such_kind", "{}", MONETIZE_SEED));
let pulled = p
.pull(&mut g, &b, Some(&monetize_pub()), 0, 10_000)
.unwrap();
assert!(
!pulled.acks[0].ok && pulled.acks[0].detail.contains("twin"),
"{pulled:?}"
);
assert!(
!pulled.acks[1].ok && pulled.acks[1].detail.contains("no_such_kind"),
"{pulled:?}"
);
assert!(w.main.got.lock().unwrap().is_empty());
assert_eq!(p.pulled.last_seq, 3);
}
#[test]
fn block_on_waits_for_a_future_that_is_woken_later() {
let (tx, rx) = tokio::sync::oneshot::channel::<u32>();
let t = std::thread::spawn(move || {
std::thread::sleep(std::time::Duration::from_millis(50));
tx.send(7).unwrap();
});
assert_eq!(block_on(rx).unwrap(), 7);
t.join().unwrap();
}
}