crates/lake-product/src/lake_box.rs
//! **A Lake box**: one machine of a Lake estate, in one role (`front`, `ingestor`, `be`, `fe`,
//! `graph`, or a numbered member such as `be-2`). The same code runs every role, as one binary
//! would. It is PULLED: nothing calls into it for the product; it reports its tenants and usage
//! over `Fleet.Report` and pulls its signed orders over `Fleet.Orders`. Each call is printed;
//! each answer is the smallest correct one.
//!
//! | door | method path | who calls it | the answer |
//! |---|---|---|---|
//! | pairing | `POST /pair` | the installer, after the lay | the box's fleet key and ticket key |
//! | front | `GET /login`, `GET /first-login` | the sale check | `ok`; the first-admin door's state |
//! | live update | `GET /live/running`, `POST /live/stage` | the update (custody step) | the running app; `staged` |
//! | keep update | `POST /keep` | the update's keep boot | `kept` |
//! | fleet (outbound) | `Fleet.Report`, `Fleet.Orders` | the box itself, every minute | — |
use std::collections::BTreeMap;
use std::net::{SocketAddr, TcpListener};
use std::sync::{Arc, Mutex};
use ed25519_dalek::{Signer, SigningKey};
use monetize_embed::fleet::{fleet_call_message, FleetCall, OrderAck, OrdersAfter, Report};
use monetize_embed::signing::base64_encode;
use monetize_product::pull::ProductReport;
use crate::{http, say};
pub use monetize_guest_lake::{role, PRODUCT};
/// **The ack of one pulled order**: `ok` for a kind the Lake's facts name, else the refusal BY
/// NAME from those facts (A28 M23: data, no Lake code decides it).
pub fn ack(role: &str, seq: u64, kind: &str) -> OrderAck {
match monetize_guest_lake::facts().runs(kind) {
Ok(()) => OrderAck {
seq,
ok: true,
detail: format!("the Lake {role} ran {kind}"),
receipt_json: String::new(),
},
Err(words) => OrderAck {
seq,
ok: false,
detail: words,
receipt_json: String::new(),
},
}
}
/// The meter the Lake's boxes report (the bytes ingested into the input store).
pub const METER: &str = "ingested_bytes";
/// What the box holds.
#[derive(Default)]
pub struct Held {
/// tenant → bytes stored.
pub tenants: BTreeMap<String, u64>,
/// tenant → the caps of the last entitlement.
pub entitled: BTreeMap<String, serde_json::Value>,
/// The estate it was paired into (`fleet_instance`), empty before the pairing.
pub instance: String,
/// The running app's pin and slot (the live door).
pub app: String,
pub slot: String,
/// Keep boots taken.
pub keeps: u32,
/// The first-admin door: `open` until the first admin signs in.
pub first_door: String,
/// The orders run, to ack in the next report.
pub acks: Vec<OrderAck>,
pub last_seq: u64,
}
/// **One Lake box.**
pub struct LakeBox {
/// Its role in the estate (`front`, `be`, `be-2`, …).
pub role: String,
pub addr: SocketAddr,
pub key: SigningKey,
/// The ticket key (an actor ticket's signer on a real box; named only here).
pub ticket_key: SigningKey,
pub held: Arc<Mutex<Held>>,
}
fn seed(word: &str) -> [u8; 32] {
use std::hash::{BuildHasher, Hasher};
let mut out = [0u8; 32];
for (i, chunk) in out.chunks_mut(8).enumerate() {
let mut h = std::collections::hash_map::RandomState::new().build_hasher();
h.write(word.as_bytes());
h.write_usize(i);
chunk.copy_from_slice(&h.finish().to_le_bytes());
}
out
}
impl LakeBox {
/// Start the box of `role` on `127.0.0.1:0` with one tenant, `acme`, holding 42 bytes.
pub fn start(role: &str) -> std::io::Result<Arc<LakeBox>> {
let listener = TcpListener::bind("127.0.0.1:0")?;
let addr = listener.local_addr()?;
let held = Arc::new(Mutex::new(Held {
app: "app-1".into(),
slot: "a".into(),
first_door: "open".into(),
..Default::default()
}));
held.lock().unwrap().tenants.insert("acme".into(), 42);
let b = Arc::new(LakeBox {
role: role.to_string(),
addr,
key: SigningKey::from_bytes(&seed(&format!("fleet-{role}"))),
ticket_key: SigningKey::from_bytes(&seed(&format!("ticket-{role}"))),
held,
});
let me = b.clone();
http::serve(listener, move |req| me.handle(req));
say(
"box",
"start",
"bench",
&format!("{role} listening at {addr}"),
);
Ok(b)
}
/// The box's fleet key, base64 (what the installer's verdict names for the role).
pub fn fleet_pubkey(&self) -> String {
base64_encode(self.key.verifying_key().as_bytes())
}
fn handle(&self, req: http::Request) -> http::Response {
let from = req.header("x-lake-from").unwrap_or("monetize").to_string();
let mut h = self.held.lock().unwrap();
let path = req.path.split('?').next().unwrap_or_default().to_string();
match (req.method.as_str(), path.as_str()) {
("POST", "/pair") => {
let j = req.json();
h.instance = j["fleet_instance"].as_str().unwrap_or_default().to_string();
say(
"pairing",
"feed",
&from,
&format!(
"company={} instance={} role={}",
j["company"].as_str().unwrap_or_default(),
h.instance,
self.role
),
);
let offer = serde_json::json!({ "role": self.role, "fleet_pubkey": self.fleet_pubkey(), "ticket_key": base64_encode(self.ticket_key.verifying_key().as_bytes()), "status": "PAIRED" });
(200, offer.to_string())
}
("GET", "/login") => {
say("front", "login", &from, "");
(200, "\"Lake login\"".into())
}
("GET", "/first-login") => {
say(
"front",
"first-login",
&from,
&format!("door={}", h.first_door),
);
(200, serde_json::json!({ "door": h.first_door }).to_string())
}
("GET", "/live/running") => {
say(
"live",
"running",
&from,
&format!("app={} slot={}", h.app, h.slot),
);
(200, serde_json::json!({ "base": "lake-base", "app": h.app, "slot": h.slot, "live_update": true }).to_string())
}
("POST", "/live/stage") => {
let j = req.json();
let pin = j["pin_hex"].as_str().unwrap_or_default().to_string();
say("live", "stage", &from, &format!("pin={pin}"));
h.app = pin;
h.slot = "b".into();
(200, "\"staged\"".into())
}
("POST", "/keep") => {
let j = req.json();
h.keeps += 1;
say(
"keep",
"boot",
&from,
&format!(
"release={} keeps={}",
j["release"].as_str().unwrap_or_default(),
h.keeps
),
);
(200, "\"kept\"".into())
}
(m, p) => {
say("box", "unknown", &from, &format!("{} {m} {p}", self.role));
(404, format!("\"the Lake box has no {m} {p}\""))
}
}
}
fn signed(
&self,
instance: &str,
call: &str,
body: String,
now_ms: u64,
) -> monetize_proto::FleetSigned {
let c = FleetCall {
product: PRODUCT.into(),
instance: instance.into(),
role: self.role.clone(),
call: call.into(),
body,
at_unix_ms: now_ms,
};
let signature = self.key.sign(&fleet_call_message(&c)).to_bytes().to_vec();
monetize_proto::FleetSigned {
product: c.product,
instance: c.instance,
role: c.role,
call: c.call,
body: c.body,
at_unix_ms: c.at_unix_ms,
signature,
box_key: self.key.verifying_key().as_bytes().to_vec(),
}
}
/// **`Fleet.Report`**: the heartbeat, the health words the sale check reads (`serving`,
/// `data`), the acks of the orders run, and the product's own state (tenants and usage:
/// what monetize meters). Answers the acked seqs.
pub async fn report(&self, endpoint: &str, now_ms: u64) -> Result<Vec<u64>, String> {
let (instance, body) = {
let mut h = self.held.lock().unwrap();
let usage = h
.tenants
.iter()
.map(|(t, b)| {
(
t.clone(),
monetize_product::Usage {
meters: [(METER.to_string(), *b)].into_iter().collect(),
measured_at_unix_ms: now_ms,
},
)
})
.collect();
let tenants = h
.tenants
.keys()
.map(|t| monetize_product::TenantOwner {
tenant: monetize_product::TenantId(t.clone()),
person: true,
seller: false,
})
.collect();
let product = ProductReport {
tenants,
usage,
measured_unix_ms: now_ms,
..Default::default()
};
let r = Report {
build: h.app.clone(),
health: [
("serving", "yes"),
("data", "ok"),
("serves_at", "127.0.0.1"),
]
.into_iter()
.map(|(k, v)| (k.to_string(), v.to_string()))
.collect(),
acked: std::mem::take(&mut h.acks),
product: Some(serde_json::to_string(&product).unwrap()),
measured_unix_ms: now_ms,
..Default::default()
};
(h.instance.clone(), serde_json::to_string(&r).unwrap())
};
let mut c = monetize_proto::fleet_client::FleetClient::connect(endpoint.to_string())
.await
.map_err(|e| format!("the ONE box's Fleet door at {endpoint}: {e}"))?;
let mut req = tonic::Request::new(self.signed(&instance, "report", body, now_ms));
req.set_timeout(http::TIMEOUT);
let a = c
.report(req)
.await
.map_err(|s| format!("Fleet.Report refused: {:?} {}", s.code(), s.message()))?
.into_inner();
say(
"fleet",
"report",
&self.role,
&format!(
"instance={instance} acked={:?} next_ms={}",
a.acked, a.next_report_ms
),
);
Ok(a.acked)
}
/// **`Fleet.Orders`**: the orders queued for this box (a short long-poll). Each is acked
/// in the next report ([`ack`]: a kind its facts do not name is refused by name); a real box
/// verifies the order's signature first and runs it.
pub async fn orders(
&self,
endpoint: &str,
now_ms: u64,
wait_ms: u64,
) -> Result<Vec<String>, String> {
let (instance, after) = {
let h = self.held.lock().unwrap();
(h.instance.clone(), h.last_seq)
};
let body = serde_json::to_string(&OrdersAfter { after, wait_ms }).unwrap();
let mut c = monetize_proto::fleet_client::FleetClient::connect(endpoint.to_string())
.await
.map_err(|e| format!("the ONE box's Fleet door at {endpoint}: {e}"))?;
let mut req = tonic::Request::new(self.signed(&instance, "orders", body, now_ms));
req.set_timeout(http::TIMEOUT + std::time::Duration::from_millis(wait_ms));
let a = c
.orders(req)
.await
.map_err(|s| format!("Fleet.Orders refused: {:?} {}", s.code(), s.message()))?
.into_inner();
let mut kinds = Vec::new();
let mut h = self.held.lock().unwrap();
for o in a.orders {
say(
"fleet",
"order",
"monetize",
&format!("role={} seq={} kind={}", self.role, o.seq, o.kind),
);
h.last_seq = h.last_seq.max(o.seq);
h.acks.push(ack(&self.role, o.seq, &o.kind));
kinds.push(o.kind);
}
Ok(kinds)
}
/// **A front's `Fleet.Ask`** (P3-FLEET, plan §8: the old `SoldEstate` door folded into
/// Fleet): signed by this box's own fleet key as its role, the old door's answer decoded.
async fn front_ask<M: prost::Message + Default>(
&self,
endpoint: &str,
kind: &str,
body: monetize_embed::fleet::FrontAsk,
now_ms: u64,
) -> Result<M, String> {
let instance = self.held.lock().unwrap().instance.clone();
let body = serde_json::to_string(&monetize_embed::fleet::Ask {
kind: kind.into(),
body: serde_json::to_value(body).unwrap(),
})
.unwrap();
let mut c = monetize_proto::fleet_client::FleetClient::connect(endpoint.to_string())
.await
.map_err(|e| format!("the ONE box's Fleet door at {endpoint}: {e}"))?;
let mut req = tonic::Request::new(self.signed(&instance, "ask", body, now_ms));
req.set_timeout(http::TIMEOUT);
let a = c
.ask(req)
.await
.map_err(|s| {
format!(
"Fleet.Ask{{{kind}}} refused: {:?} {}",
s.code(),
s.message()
)
})?
.into_inner();
M::decode(a.answer.as_slice()).map_err(|e| format!("Fleet.Ask{{{kind}}}: {e}"))
}
/// **`Fleet.Ask{update_take}`** (the front's poll): the open update, whose turn it is, and
/// the live app when live update is on.
pub async fn update_take(
&self,
endpoint: &str,
now_ms: u64,
) -> Result<monetize_proto::EstateUpdateTakeResponse, String> {
let a: monetize_proto::EstateUpdateTakeResponse = self
.front_ask(
endpoint,
monetize_embed::fleet::ask::UPDATE_TAKE,
Default::default(),
now_ms,
)
.await?;
say(
"fleet",
"update_take",
"box",
&format!(
"open={} release={} live={} role={} words={:?}",
a.open, a.release, a.live, a.role, a.words
),
);
Ok(a)
}
/// **`Fleet.Ask{update_report}`**: `role`'s step, `<outcome>\n<words>`, signed.
pub async fn update_report(
&self,
endpoint: &str,
role: &str,
outcome: &str,
words: &str,
now_ms: u64,
) -> Result<String, String> {
let body = monetize_embed::fleet::FrontAsk {
role: role.into(),
carries: format!("{outcome}\n{words}"),
..Default::default()
};
let a: monetize_proto::EstateUpdateReportResponse = self
.front_ask(
endpoint,
monetize_embed::fleet::ask::UPDATE_REPORT,
body,
now_ms,
)
.await?;
say(
"fleet",
"update_report",
"box",
&format!("role={role} outcome={outcome} answer={:?}", a.words),
);
Ok(a.words)
}
}