crates/monetize-server/src/services/fleet.rs
//! `Fleet` (the pull channel, no bearer: every call signed by the box) and `Guardian` (the
//! operator's console, behind the bearer) over [`crate::fleet`].
use std::sync::Arc;
use std::time::Duration;
use monetize_embed::fleet::{FleetCall, SignedOrder};
use monetize_guardian::{Case, FleetRefusal, PressRefusal};
use monetize_product::guardian::Remedy;
use monetize_proto as pb;
use tonic::{Request, Response, Status};
use crate::fleet::ShopWorld;
use crate::services::blocking;
use crate::state::AppState;
/// How often a held `Orders` call looks at the queue again.
pub const ORDERS_RELOOK: Duration = Duration::from_millis(250);
/// When a box reports next.
pub const NEXT_REPORT_MS: u64 = 30_000;
const NO_GUARDIAN: &str = "this monetize runs no estate guardian";
fn refusal(r: FleetRefusal) -> Status {
match r {
FleetRefusal::Unauthenticated => Status::unauthenticated(r.to_string()),
FleetRefusal::Malformed(w) => Status::invalid_argument(w),
FleetRefusal::Unavailable(w) => Status::unavailable(w),
FleetRefusal::Refused(w) => Status::failed_precondition(w),
}
}
pub(crate) fn press_refusal(r: PressRefusal) -> Status {
match r {
PressRefusal::Weak(w) => Status::permission_denied(w),
PressRefusal::Busy(w) => Status::failed_precondition(w),
PressRefusal::Unknown(w) => Status::not_found(w),
PressRefusal::Refused(w) => Status::failed_precondition(w),
}
}
fn call_of(s: &pb::FleetSigned) -> FleetCall {
FleetCall {
product: s.product.clone(),
instance: s.instance.clone(),
role: s.role.clone(),
call: s.call.clone(),
body: s.body.clone(),
at_unix_ms: s.at_unix_ms,
}
}
pub fn order_pb(o: &SignedOrder) -> pb::FleetOrder {
pb::FleetOrder {
seq: o.seq,
product: o.product.clone(),
instance: o.instance.clone(),
role: o.role.clone(),
kind: o.kind.clone(),
body: o.body.clone(),
issued_unix_ms: o.issued_unix_ms,
not_after_unix_ms: o.not_after_unix_ms,
signature: o.signature.clone(),
}
}
pub fn order_of(o: &pb::FleetOrder) -> SignedOrder {
SignedOrder {
seq: o.seq,
product: o.product.clone(),
instance: o.instance.clone(),
role: o.role.clone(),
kind: o.kind.clone(),
body: o.body.clone(),
issued_unix_ms: o.issued_unix_ms,
not_after_unix_ms: o.not_after_unix_ms,
signature: o.signature.clone(),
}
}
/// Authenticate the call; a box presenting its key inside a pin window is pinned first.
fn admit(
state: &AppState,
s: &pb::FleetSigned,
) -> Result<(Arc<monetize_guardian::Guardian>, FleetCall), Status> {
let g = state
.fleet
.guardian()
.cloned()
.ok_or_else(|| Status::unavailable(NO_GUARDIAN))?;
let c = call_of(s);
if !s.box_key.is_empty() {
g.first_call(&c, &s.signature, &s.box_key, state.now())
.map_err(refusal)?;
}
Ok((g, c))
}
pub struct FleetService(pub Arc<AppState>);
#[tonic::async_trait]
impl pb::fleet_server::Fleet for FleetService {
async fn report(
&self,
req: Request<pb::FleetSigned>,
) -> Result<Response<pb::FleetReportAnswer>, Status> {
let s = req.into_inner();
let answer = blocking(&self.0, move |state| {
let (g, c) = admit(state, &s)?;
let acked = g.report(&c, &s.signature, state.now()).map_err(refusal)?;
Ok(Response::new(pb::FleetReportAnswer {
acked,
next_report_ms: NEXT_REPORT_MS,
}))
})
.await?;
// An ack may be a paid account the main just made: its room is queued now, behind the
// answer (a settle may wait on the store), not at the poller's next tick.
if !answer.get_ref().acked.is_empty() {
let (state, acked) = (self.0.clone(), answer.get_ref().acked.clone());
// Not awaited: the answer goes now. A settle that never ran is the poller's next tick's.
drop(settle_behind(state, acked));
}
Ok(answer)
}
async fn orders(
&self,
req: Request<pb::FleetSigned>,
) -> Result<Response<pb::FleetOrdersAnswer>, Status> {
let s = Arc::new(req.into_inner());
let started = std::time::Instant::now();
loop {
let s2 = s.clone();
let (orders, wait_ms) = blocking(&self.0, move |state| {
let (g, c) = admit(state, &s2)?;
g.orders(&c, &s2.signature, state.now()).map_err(refusal)
})
.await?;
if !orders.is_empty() || started.elapsed() >= Duration::from_millis(wait_ms) {
return Ok(Response::new(pb::FleetOrdersAnswer {
orders: orders.iter().map(order_pb).collect(),
}));
}
tokio::time::sleep(ORDERS_RELOOK).await;
}
}
async fn ask(
&self,
req: Request<pb::FleetSigned>,
) -> Result<Response<pb::FleetAskAnswer>, Status> {
let s = req.into_inner();
blocking(&self.0, move |state| {
let (g, c) = admit(state, &s)?;
let a = match serde_json::from_str::<monetize_embed::fleet::Ask>(&c.body) {
Ok(ask) if monetize_embed::fleet::ask::is_front(&ask.kind) => {
front_answer(state, &g, &c, &s.signature, ask)?
}
_ => crate::fleet::ask(state, &g, &c, &s.signature).map_err(refusal)?,
};
Ok(Response::new(pb::FleetAskAnswer {
case_id: a.case_id,
words: a.words,
url: a.url,
answer: a.answer,
}))
})
.await
}
}
/// A sold call's refusal on the Fleet door: the old door's words, the same codes.
fn sold_refusal(e: crate::estate_order::sold::SoldRefusal) -> Status {
use crate::estate_order::sold::SoldRefusal;
match e {
SoldRefusal::NotSigned => refusal(FleetRefusal::Unauthenticated),
SoldRefusal::Refused(w) => Status::failed_precondition(w),
SoldRefusal::Unavailable(w) => Status::unavailable(w),
}
}
/// **A front's ask answered** (plan §8: the five `SoldEstate` calls folded into `Fleet.Ask`):
/// proven by [`crate::fleet::front_ask`], done by the same estate-order functions the old door
/// calls ([`crate::estate_order::sold::Caller::Fleet`]), the old door's answer encoded into
/// `answer` and its sentence in `words`. Refused the way the old door refuses.
fn front_answer(
state: &AppState,
g: &monetize_guardian::Guardian,
c: &FleetCall,
signature: &[u8],
ask: monetize_embed::fleet::Ask,
) -> Result<crate::fleet::Asked, Status> {
use crate::estate_order::{sold, update};
use crate::services::estate_order::{material_pb, recharge_pb, take_pb, upgrade_pb};
use monetize_embed::fleet::ask as kind;
use prost::Message as _;
let f = crate::fleet::front_ask(state, g, c, signature, ask).map_err(refusal)?;
let (co, b, by) = (f.company.as_str(), &f.body, sold::Caller::Fleet);
let said = |words: &str, answer: Vec<u8>| crate::fleet::Asked {
words: words.to_string(),
answer,
..Default::default()
};
Ok(match f.kind.as_str() {
kind::UPGRADE => {
let a = upgrade_pb(
sold::upgrade_as(state, co, b.role.trim(), b.done, by).map_err(sold_refusal)?,
);
said(&a.words, a.encode_to_vec())
}
kind::RECHARGE => {
let a = recharge_pb(
sold::recharge_as(state, co, b.role.trim(), &b.seed, b.done, by)
.map_err(sold_refusal)?,
);
said(&a.words, a.encode_to_vec())
}
kind::MATERIAL => {
let a =
material_pb(sold::take_material_as(state, co, b.done, by).map_err(sold_refusal)?);
said(&a.words, a.encode_to_vec())
}
kind::UPDATE_TAKE => {
let a = take_pb(update::take_as(state, co, by).map_err(sold_refusal)?);
said(&a.words, a.encode_to_vec())
}
kind::UPDATE_REPORT => {
let words = update::report_as(state, co, b.role.trim(), &b.carries, by)
.map_err(sold_refusal)?;
said(
&words,
pb::EstateUpdateReportResponse {
words: words.clone(),
}
.encode_to_vec(),
)
}
other => {
return Err(Status::invalid_argument(format!(
"{other:?} is no front's ask"
)))
}
})
}
pub fn case_pb(c: &Case) -> pb::GuardianCase {
pb::GuardianCase {
id: c.id.clone(),
instance: c.instance.clone(),
product: c.product.clone(),
state: c.state.as_str().into(),
opened_unix_ms: c.opened_unix_ms,
updated_unix_ms: c.updated_unix_ms,
finding_code: c.finding.code.clone(),
finding_words: c.finding.words.clone(),
evidence: c.finding.evidence.clone(),
buttons: c.buttons().iter().map(|r| r.as_str().to_string()).collect(),
remedy: c.remedy.map(|r| r.as_str().to_string()).unwrap_or_default(),
approved_by: c.approved_by.clone(),
approved_unix_ms: c.approved_unix_ms,
auto: c.auto,
epoch: c.epoch,
steps: c
.steps
.iter()
.map(|s| pb::GuardianStep {
words: s.step.words(),
state: s.state.clone(),
detail: s.words.clone(),
seq: s.seq,
})
.collect(),
log: c
.log
.iter()
.map(|l| pb::GuardianLine {
at_unix_ms: l.at_unix_ms,
words: l.words.clone(),
})
.collect(),
outcome: c.outcome.clone(),
receipt_json: if c.state.open() {
String::new()
} else {
c.receipt_json()
},
}
}
pub struct GuardianAdminService(pub Arc<AppState>);
#[tonic::async_trait]
impl pb::guardian_server::Guardian for GuardianAdminService {
async fn cases(
&self,
req: Request<pb::GuardianCasesRequest>,
) -> Result<Response<pb::GuardianCasesResponse>, Status> {
let want = req.into_inner().instance;
blocking(&self.0, move |state| {
let g = state
.fleet
.guardian()
.ok_or_else(|| Status::unavailable(NO_GUARDIAN))?;
let world = ShopWorld { state };
let now = state.now();
let mut estates = Vec::new();
for (instance, product) in g.store.estates() {
if !want.is_empty() && want != instance {
continue;
}
let view = g.view(&world, &instance, now);
let boxes = view
.boxes
.values()
.map(|b| pb::GuardianBox {
role: b.role.clone(),
heard_unix_ms: b.heard_unix_ms,
silent: b.silent(now, view.silent_after_ms),
health: b.health.clone(),
laid_ipv4: b.laid_ipv4.clone(),
})
.collect();
let mut cases: Vec<_> = g.cases(&instance).iter().map(case_pb).collect();
cases.sort_by(|a, b| {
b.opened_unix_ms
.cmp(&a.opened_unix_ms)
.then(b.id.cmp(&a.id))
});
let bearer = state.fleet.bearer_of(&instance).unwrap_or_default();
estates.push(pb::GuardianEstate {
auto_fix: g.store.auto_fix(&instance),
instance,
product,
boxes,
cases,
bearer_until_unix_ms: bearer.until_unix_ms,
bearer_revoked_unix_ms: bearer.revoked_unix_ms,
});
}
if !want.is_empty() && estates.is_empty() {
return Err(Status::not_found(format!(
"the guardian knows no estate {want:?}"
)));
}
Ok(Response::new(pb::GuardianCasesResponse { estates }))
})
.await
}
async fn press(
&self,
req: Request<pb::GuardianPressRequest>,
) -> Result<Response<pb::GuardianCase>, Status> {
let r = req.into_inner();
blocking(&self.0, move |state| {
let g = state
.fleet
.guardian()
.ok_or_else(|| Status::unavailable(NO_GUARDIAN))?;
let remedy = Remedy::parse(&r.remedy).ok_or_else(|| {
Status::invalid_argument(format!(
"{:?} is no button: recharge_from_twin, review_twin, new_twin or new_server",
r.remedy
))
})?;
let case = g
.press(
&ShopWorld { state },
&r.instance,
remedy,
&r.case_id,
&r.principal,
r.strong,
state.now(),
)
.map_err(press_refusal)?;
eprintln!(
"monetize-server: guardian: {} pressed {} on {} (case {})",
r.principal,
remedy.as_str(),
r.instance,
case.id
);
Ok(Response::new(case_pb(&case)))
})
.await
}
async fn set_auto_fix(
&self,
req: Request<pb::GuardianAutoFixRequest>,
) -> Result<Response<pb::GuardianAutoFixResponse>, Status> {
let r = req.into_inner();
blocking(&self.0, move |state| {
let g = state
.fleet
.guardian()
.ok_or_else(|| Status::unavailable(NO_GUARDIAN))?;
if g.store.product_of(&r.instance).is_none() {
return Err(Status::not_found(format!(
"the guardian knows no estate {:?}",
r.instance
)));
}
let on = g
.set_auto_fix(&r.instance, r.on, &r.principal, r.strong)
.map_err(press_refusal)?;
eprintln!(
"monetize-server: guardian: {} turned auto-fix {} for {}",
r.principal,
if on { "ON" } else { "off" },
r.instance
);
Ok(Response::new(pb::GuardianAutoFixResponse { on }))
})
.await
}
/// ★ SEC-S1: an operator holds (stops on purpose) or releases one box.
async fn set_hold(
&self,
req: Request<pb::GuardianHoldRequest>,
) -> Result<Response<pb::GuardianHoldResponse>, Status> {
let r = req.into_inner();
blocking(&self.0, move |state| {
let g = state
.fleet
.guardian()
.ok_or_else(|| Status::unavailable(NO_GUARDIAN))?;
if g.store.product_of(&r.instance).is_none() {
return Err(Status::not_found(format!(
"the guardian knows no estate {:?}",
r.instance
)));
}
if r.on {
let words = g
.hold_box(
&ShopWorld { state },
&r.instance,
&r.role,
&r.why,
&r.principal,
r.strong,
state.now(),
)
.map_err(press_refusal)?;
eprintln!("monetize-server: guardian: {words}");
Ok(Response::new(pb::GuardianHoldResponse {
held: true,
words,
}))
} else {
let was = g
.release_box(&r.instance, &r.role, &r.principal, r.strong)
.map_err(press_refusal)?;
let words = if was {
format!(
"{} released the {} box of {}: the start pass may start it again",
r.principal, r.role, r.instance
)
} else {
format!("the {} box of {} was not held", r.role, r.instance)
};
eprintln!("monetize-server: guardian: {words}");
Ok(Response::new(pb::GuardianHoldResponse {
held: false,
words,
}))
}
})
.await
}
}
/// **What a report acked, settled behind the answer** on the blocking pool (a settle may wait
/// on the store): the ack of a paid account queues its room now, not at the poller's next tick.
/// It answers how many settled.
pub(crate) fn settle_behind(
state: Arc<AppState>,
acked: Vec<u64>,
) -> tokio::task::JoinHandle<usize> {
// A32-14, stays a task: its handle is the caller's, which awaits it.
tokio::task::spawn_blocking(move || crate::fleet::after_report(&state, &acked))
}
#[cfg(test)]
mod settle_tests {
use super::*;
/// ★ **The settle behind a report runs and answers** (score robustness.no-swallowed-save: a
/// raw thread whose start failed was `let _ =`, and nothing settled until the next tick).
#[tokio::test]
async fn the_settle_behind_a_report_runs_on_the_blocking_pool_and_answers() {
let state = Arc::new(AppState::in_memory(crate::registry::Registry::new()));
assert_eq!(
settle_behind(state, vec![]).await.expect("the settle ran"),
0,
"a report that acked nothing settles nothing"
);
}
}