server/lib/src/late.rs
//! **Late keys** (A7 row 8, lane G1): a value the installer feeds AFTER this box started
//! takes effect at once, without a restart.
//!
//! A box of an estate is fed twice or more: the shop's own values on its first visit (the
//! console's bearer, which the doors wait for), and on a later visit what the FRONT offers
//! once it is up (`offer.front.monetize_pubkey`, `offer.front.fleet`, …). Holger read the
//! later ones once at its start, so the monetize key arrived after a 600 s budget or never,
//! and the fleet door was re-read every 5 s (D13). Now every consumer subscribes to the
//! box's feed bus (Skidbladnir `nordisk-pairing::feedbus`, the bus of the box's
//! `Installed`), and takes each NEW or CHANGED value live:
//!
//! | value | consumer |
//! |---|---|
//! | `fleet`, `fleet_instance`, `fleet_at`, `fleet_ca` | [`crate::fleet_box::start_on`]: the pull channel starts, or re-dials |
//! | `monetize_pubkey` (the front's offer) | the pull channel's order key, the entitlement store, the paid limit, the growth's go-ahead, the meter plugin's key slot |
//! | `refusal_link` (the pay page) | the paid limit's refusals |
//! | `offsite_*` | [`crate::offsite::start_on`] |
//! | `secret.repair_bearer` | the repair orders the pull channel runs |
//!
//! The bus lives in nordisk-pairing; this module is the seam server-lib sees ([`Feeds`]),
//! so server-lib links no pairing code and a test drives it with any bus.
use std::collections::BTreeMap;
use std::sync::{Arc, Mutex};
use crate::lock::LockOrRecover as _;
/// A feed: key → value.
pub type Feed = BTreeMap<String, String>;
/// A subscriber of the feed.
pub type Take = Box<dyn Fn(&Feed) + Send + Sync>;
/// **Where this box's feed is published.** On an estate box: the feed bus of the box's
/// `Installed` (holger-server `paired`). Off an estate: [`NoFeeds`].
pub trait Feeds: Send + Sync {
/// `take` gets the feed in force now (if any) and every change after.
fn subscribe(&self, name: &'static str, take: Take);
/// The feed in force now (empty: none yet).
fn now(&self) -> Feed;
/// The role the installer's door ELECTED for this box
/// (nordisk-pairing `elect::pull_role`'s `door`): `None` on a box no door paired, `Some(None)` while the
/// door has not elected one, else `Some(Some("twin"))` or `Some(Some("primary"))`.
fn door_role(&self) -> Option<Option<String>>;
/// True when this box is on an installer-built estate: its feed may still change.
fn on_estate(&self) -> bool {
true
}
}
/// **No feed will ever come**: a box off an estate (a plain `holger-server start`).
pub struct NoFeeds;
impl Feeds for NoFeeds {
fn subscribe(&self, _: &'static str, take: Take) {
take(&Feed::new());
}
fn now(&self) -> Feed {
Feed::new()
}
fn door_role(&self) -> Option<Option<String>> {
None
}
fn on_estate(&self) -> bool {
false
}
}
/// **Subscribe to one view of the feed**: `take` runs with the view in force now and then
/// only when the view CHANGES (a late `fleet` wakes the fleet's subscriber once; a change
/// of another key never does).
pub fn on_change<V>(
feeds: &dyn Feeds,
name: &'static str,
view: impl Fn(&Feed) -> V + Send + Sync + 'static,
take: impl Fn(&V) + Send + Sync + 'static,
) where
V: PartialEq + Send + 'static,
{
let last: Mutex<Option<V>> = Mutex::new(None);
feeds.subscribe(
name,
Box::new(move |f| {
let v = view(f);
let mut seen = last.lock_or_recover();
if seen.as_ref() == Some(&v) {
return;
}
take(&v);
*seen = Some(v);
}),
);
}
/// Pure: `key`'s value in `feed`, plain or as a VM's offer (`offer.<vm>.<key>`), trimmed;
/// the plain one wins. `None` when absent or blank.
pub fn fed(feed: &Feed, key: &str) -> Option<String> {
feed.get(key)
.map(|v| v.trim().to_string())
.filter(|v| !v.is_empty())
.or_else(|| {
let tail = format!(".{key}");
feed.iter()
.find(|(k, v)| {
k.starts_with("offer.") && k.ends_with(&tail) && !v.trim().is_empty()
})
.map(|(_, v)| v.trim().to_string())
})
}
/// The feed key of the monetize key (the front offers it: `offer.front.monetize_pubkey`).
pub const FEED_MONETIZE_PUBKEY: &str = "monetize_pubkey";
/// The feed key of the estate's pay page (monetize-shop-pairing `contract::FEED_BILLING_URL`).
pub const FEED_PAY_PAGE: &str = "refusal_link";
/// Pure: the monetize key a feed names, 64 hex or base64 (`None`: none, or not 32 bytes).
pub fn monetize_key(feed: &Feed) -> Option<[u8; 32]> {
use base64::Engine as _;
let text = fed(feed, FEED_MONETIZE_PUBKEY)?;
let raw = hex::decode(&text)
.ok()
.or_else(|| base64::engine::general_purpose::STANDARD.decode(&text).ok())?;
<[u8; 32]>::try_from(raw.as_slice()).ok()
}
/// **The parts of a metered box that take a late monetize key and pay page**
/// ([`crate::Holger::live_keys`]).
#[cfg(feature = "metered")]
#[derive(Clone, Default)]
pub struct LiveKeys {
pub store: Option<Arc<crate::metered::EntitlementStore>>,
pub quota: Option<Arc<crate::metered::Quota>>,
pub grow: Option<Arc<appliance_grow::MonetizePolicy>>,
/// The meter plane's store (`<data_root>/meters`): the plugin's key slot.
pub meters: Option<std::path::PathBuf>,
}
#[cfg(feature = "metered")]
impl LiveKeys {
/// **monetize's key, applied in place**: the paid-state store verifies and polices
/// from now on, the growth takes monetize's go-ahead, the meter plugin reads the key.
pub fn take_key(&self, key: &[u8; 32]) -> Result<Vec<&'static str>, String> {
let mut took = Vec::new();
if let Some(s) = &self.store {
s.set_key(key)
.map_err(|_| "the fed monetize key is not an Ed25519 public key".to_string())?;
took.push("the paid-state store and the paid limit");
}
if let Some(g) = &self.grow {
g.set_key(key)?;
took.push("the growth's go-ahead");
}
if let Some(dir) = &self.meters {
use holger_plugin_host::abi::BlobStore as _;
let slot = holger_plugin_host::DirStore::new(dir.clone());
slot.put(
crate::fleet_box::METER_KEY_SLOT,
hex::encode(key).as_bytes(),
)
.map_err(|e| format!("the meter plugin's key slot: {e}"))?;
took.push("the meter plugin's key slot");
}
Ok(took)
}
/// **The pay page, applied in place.**
pub fn take_pay_page(&self, page: Option<String>) {
if let Some(q) = &self.quota {
q.set_pay_page(page);
}
}
}
/// **Follow the feed's monetize key and pay page** (A7 row 8): a key the front offers after
/// the start is enforced at once; a pay page the shop feeds later is named by the next paid
/// refusal. `configured` is the key the box started with (the RON or the kept feed): a fed
/// key that differs from the key in force (configured, or the first one fed) is refused by
/// name and the key in force stays.
#[cfg(feature = "metered")]
pub fn follow_monetize(feeds: &dyn Feeds, live: LiveKeys, configured: Option<[u8; 32]>) {
let keys = live.clone();
let in_force: Mutex<Option<[u8; 32]>> = Mutex::new(configured);
on_change(
feeds,
"late: the monetize key",
monetize_key,
move |k: &Option<[u8; 32]>| {
let Some(k) = k else { return };
let mut now = in_force.lock_or_recover();
match *now {
Some(c) if &c == k => {}
Some(c) => log::error!("monetize: the feed names a monetize key ({}) that is not the one this box enforces with ({}): refused, the key in force stays", hex::encode(k), hex::encode(c)),
None => match keys.take_key(k) {
Ok(took) => {
*now = Some(*k);
log::warn!("monetize: the front's key arrived after the start and is enforced now, without a restart ({})", took.join(", "));
}
Err(why) => log::error!("monetize: the fed key is refused: {why}; no plan is enforced on this box"),
},
}
},
);
on_change(
feeds,
"late: the pay page",
|f: &Feed| fed(f, FEED_PAY_PAGE),
move |p: &Option<String>| {
if p.is_some() {
live.take_pay_page(p.clone());
}
},
);
}
/// **A test's or a product's bus**: a feed published by hand, every change to every
/// subscriber (in process; the box's real bus is nordisk-pairing's).
#[derive(Default)]
pub struct HandFed {
now: Mutex<Option<Feed>>,
takers: Mutex<Vec<Arc<Take>>>,
door: Mutex<Option<Option<String>>>,
}
impl HandFed {
pub fn new() -> Arc<HandFed> {
Arc::new(HandFed::default())
}
/// Publish `feed`: every subscriber gets it when it changed.
pub fn publish(&self, feed: Feed) {
{
let mut now = self.now.lock_or_recover();
if now.as_ref() == Some(&feed) {
return;
}
*now = Some(feed.clone());
}
let takers: Vec<Arc<Take>> = self.takers.lock_or_recover().clone();
for t in takers {
t(&feed);
}
}
/// What the door elected (see [`Feeds::door_role`]).
pub fn elect(&self, door: Option<Option<&str>>) {
*self.door.lock_or_recover() = door.map(|d| d.map(str::to_string));
}
}
impl Feeds for HandFed {
fn subscribe(&self, _: &'static str, take: Take) {
let take = Arc::new(take);
self.takers.lock_or_recover().push(Arc::clone(&take));
if let Some(now) = self.now.lock_or_recover().clone() {
take(&now);
}
}
fn now(&self) -> Feed {
self.now.lock_or_recover().clone().unwrap_or_default()
}
fn door_role(&self) -> Option<Option<String>> {
self.door.lock_or_recover().clone()
}
}
#[cfg(test)]
mod tests {
use super::*;
fn feed(pairs: &[(&str, &str)]) -> Feed {
pairs
.iter()
.map(|(k, v)| (k.to_string(), v.to_string()))
.collect()
}
#[test]
fn a_view_subscriber_wakes_only_for_its_own_key() {
let bus = HandFed::new();
let seen = Arc::new(Mutex::new(Vec::new()));
let s = Arc::clone(&seen);
on_change(
bus.as_ref(),
"key",
monetize_key,
move |k: &Option<[u8; 32]>| s.lock().unwrap().push(k.is_some()),
);
bus.publish(feed(&[("holger_console_token", "t")]));
bus.publish(feed(&[("holger_console_token", "u")]));
bus.publish(feed(&[("offer.front.monetize_pubkey", &"ab".repeat(32))]));
assert_eq!(*seen.lock().unwrap(), vec![false, true]);
}
#[test]
fn a_value_is_plain_or_offered_and_the_plain_one_wins() {
assert_eq!(
fed(&feed(&[("offer.front.fleet", " https://one ")]), "fleet").as_deref(),
Some("https://one")
);
assert_eq!(
fed(
&feed(&[
("fleet", "https://plain"),
("offer.front.fleet", "https://one")
]),
"fleet"
)
.as_deref(),
Some("https://plain")
);
assert_eq!(fed(&feed(&[("offer.front.fleet", " ")]), "fleet"), None);
assert_eq!(monetize_key(&feed(&[("monetize_pubkey", "zz")])), None);
}
}