server/lib/src/fleet_box.rs
//! **Holger's verbs as the ONE monetize's orders reach them** (`metered` only): the
//! [`holger_fleet::HolgerBox`] over this server's own stores and lanes. Monetize never
//! dials this box; the pull channel (holger-fleet's jera round) hands each verified order
//! here.
//!
//! | order | holger's code |
//! |---|---|
//! | `entitlement`, `restate_entitlements` | [`crate::metered::EntitlementStore`] (verify, then write), newer wins |
//! | `purge_tenant` | [`crate::metered::purge_walk`] — the walk `EntitlementService.Purge` runs |
//! | `grow_room` | kept at `fleet/room-wanted.json`, said on the health words and the log |
//! | `grow_go_ahead` (+ `Fleet.Ask{grow_set}`) | [`crate::grow::GrowPlane`] start / settle / resume |
//! | `recharge_from_twin` | [`crate::recharge_lane::run_grpc`] — THE recharge path: fill from the twin, the boot hold released, the twin re-pointed |
//! | `adopt_twin` | `nordisk_twin::certify::once` over [`crate::twin_tls::GrpcBoxes`] |
//! | `review_twin` | [`crate::twin::TwinPosture`] (a twin) / the twin roster (a main) |
//! | `fence_primary` | the old primary fenced: ITSELF ([`crate::twin::TwinPosture::fence`]), or a peer over `AdminService.FencePrimary` ([`crate::twin_tls::GrpcBoxes::fence_primary`]); a recharge's fence-first the same |
//! | `become_twin` | holger's `BecomeTwin` election on this fresh box |
//! | `add_principal` | a repository on a plan: the signup bound to a served repository (`fleet/signups.json`), or a new one made for it (`fleet/sold-repositories.json`, served at once through [`crate::live_routes::LiveRoutes`]; a gated or pull-through one from the next start) |
//! | `add_users` | refused by name (holger-fleet `order`) |
//! | `sign_in_again` | [`crate::sign_in_again::resend`] through the fed relay |
//! | `refeed` | [`crate::refeed::keep`] at `fleet/refeed.json` (0600), over the lay's feed: the relay from the next mail, the public repositories from the next start |
use crate::lock::{LockOrRecover as _, RwLockOrRecover as _};
use std::collections::BTreeMap;
use std::path::PathBuf;
use std::sync::Arc;
use std::time::Duration;
use holger_fleet::exec::{Applied, GrowState, RechargeAsk, SECTORS_PER_GIB};
use holger_fleet::{HolgerBox, OrderedBy};
use monetize_product::pull::{AddPrincipal, GrowRoom, ProductReport, PurgeAnswer, PurgeTenant};
use monetize_product::{
EntitlementFact, Growth, MeteredDoor, Servable, TenantId, TenantOwner, Usage,
};
use crate::audit::{AuditAction, AuditEvent};
use crate::grpc::HolgerGrpc;
/// Which signup each repository was made for (`add_principal`), under the data root.
pub const SIGNUPS_REL: &str = "fleet/signups.json";
pub use crate::sold::{SoldRepository, SOLD_REPOS_REL};
/// The health word naming the repositories made for a paid signup that this start does not
/// serve yet (they are served from the next one).
///
/// **Nothing acts on it** (lane W9, checked 2026-10-19): since M4b a made repository serves at
/// once and the word does not appear. It remains only for a repository shaped like a template
/// the live route table cannot serve (a read gate or upstreams, which are wired at a start). The
/// ONE box reads no such word (monetize has no reader of it) and no order follows from it; it is kept in
/// [`SOLD_REPOS_REL`], and the box's next start for any reason serves it and the word clears
/// (`tests/repositories_wanted_clears_at_the_next_start.rs`).
pub const REPOSITORIES_WANTED: &str = "repositories_wanted";
pub use crate::sold::repository_name_ok;
/// Where a `grow_room` want is kept under the data root.
pub const ROOM_WANTED_REL: &str = "fleet/room-wanted.json";
/// The health word (and log event) of a kept room want.
pub const ROOM_WANTED: &str = "room_wanted";
/// How long a pulled growth waits for the grow machine to stop moving.
pub const GROW_CEILING: Duration = Duration::from_secs(3600);
/// One Holger box, as the pull channel sees it.
pub struct FleetBox {
pub grpc: Arc<HolgerGrpc>,
/// This box's control plane (`host:port`): what it serves at, the primary a twin follows.
pub own: String,
/// The runtime the recharge lane and the certification block on.
pub rt: tokio::runtime::Handle,
/// The admin credential the lane's calls carry (both boxes accept it), followed live:
/// a `secret.repair_bearer` fed after the start reaches the next order (A7 row 8).
pub bearer: LiveBearer,
pub build: String,
/// The last pulled `Recharge` named failures (the `recharge=failed` word).
pub recharge_failed: std::sync::Mutex<bool>,
}
fn data_root(g: &HolgerGrpc) -> Result<PathBuf, String> {
g.data_root
.clone()
.ok_or_else(|| "this server has no data root".to_string())
}
/// **The repair orders' bearer, followed live** (A7 row 8).
#[derive(Clone, Default)]
pub struct LiveBearer(Arc<std::sync::RwLock<Option<String>>>);
impl LiveBearer {
pub fn now(&self) -> Option<String> {
self.0.read_or_recover().clone()
}
pub fn set(&self, bearer: Option<String>) {
*self.0.write_or_recover() = bearer;
}
}
impl From<Option<String>> for LiveBearer {
fn from(b: Option<String>) -> LiveBearer {
LiveBearer(Arc::new(std::sync::RwLock::new(b)))
}
}
impl FleetBox {
/// **A bundle sent by an operator order, through the chain's bundles door, as the order**
/// (monetize MONETIZE-FIX-2; rickard-e6's conditions, 2026-11-02). The writer is the verified
/// order's own principal, `fleet-order <seq> signed <monetize key>` ([`OrderedBy`]): it
/// exists only here, in-process, for an order the pump verified; no token, nothing anyone can
/// present over HTTP or gRPC. It writes the bundles DOOR (`promotion.intake.bundles`) and no
/// other repository, and the door runs its whole verify-bundle (the release signature, the
/// approvals, every member) against its trust policy before a byte is stored
/// ([`crate::import_door::ImportDoor::store_verified`], the HTTP PUT's own code). Every answer
/// is audited under that principal, never an admin's.
pub fn bundle_put(
&self,
by: &OrderedBy,
repository: &str,
path: &str,
data: &[u8],
) -> Result<String, String> {
let principal = by.principal();
let path = path.trim().trim_start_matches('/');
let audit = |repo: &str, status: u16, bytes: u64, detail: &str| {
let ev = AuditEvent::new(
&principal,
AuditAction::Upload,
repo,
path,
"fleet-order",
status,
bytes,
)
.with_detail(detail);
if let Err(e) = self.grpc.audit.record(ev) {
log::warn!("fleet: audit of {principal}'s bundle_put failed: {e}");
}
};
#[cfg(feature = "intake")]
{
let Some(door_repo) = self.grpc.intake.as_ref().and_then(|c| c.bundles_door()) else {
let why = "this box's chain names no bundles door (`promotion.intake.bundles`): an ordered bundle has nowhere to go; nothing was stored";
audit(repository, 503, 0, why);
return Err(why.into());
};
let repository = match repository.trim() {
"" => door_repo,
r => r,
};
if repository != door_repo {
let why = format!("{principal} writes only the bundles door '{door_repo}', never '{repository}': nothing was stored");
audit(repository, 403, 0, &why);
return Err(why);
}
if path.is_empty() || path.contains("..") || path.contains('?') {
let why = format!("bundle_put needs a plain `path` in the bundles door, not {path:?}: nothing was stored");
audit(repository, 400, 0, &why);
return Err(why);
}
let refused = |status: u16, why: String| {
audit(repository, status, 0, &why);
format!("{repository}/{path} was not taken: {why}")
};
let door = self.grpc.imports.clone().ok_or_else(|| {
refused(
503,
"this server has no import door to verify a bundle with".into(),
)
})?;
let store = self.grpc.routes.lookup(repository).ok_or_else(|| {
refused(
404,
format!("the bundles door '{repository}' is no repository this box serves"),
)
})?;
let mut staged = door.stage().map_err(|e| refused(e.status, e.tagged()))?;
staged
.write(data)
.map_err(|e| refused(e.status, e.tagged()))?;
staged.finish().map_err(|e| refused(e.status, e.tagged()))?;
let pq = format!("/{repository}/{path}");
match door.store_verified(
&staged.path,
store.as_ref(),
&pq,
&principal,
"a pulled operator order",
) {
Ok((what, n, (status, _, _))) if (200..300).contains(&status) => {
audit(repository, status, n, &format!("bundle published: {what}"));
Ok(serde_json::json!({ "repository": repository, "path": path, "bytes": n, "status": status, "published": what, "by": principal }).to_string())
}
Ok((_, _, (status, _, body))) => Err(refused(
status,
String::from_utf8_lossy(&body).trim().to_string(),
)),
Err(e) => Err(refused(e.status, e.tagged())),
}
}
#[cfg(not(feature = "intake"))]
{
let _ = data;
let why = "this build has no intake chain, so no bundles door: nothing was stored";
audit(repository, 503, 0, why);
Err(why.into())
}
}
/// The repositories paid signups made (`fleet/sold-repositories.json`).
fn sold(&self) -> BTreeMap<String, SoldRepository> {
data_root(&self.grpc)
.map(|r| crate::sold::read(&r))
.unwrap_or_default()
}
/// The made repositories this start does not serve yet (served from the next one).
fn sold_unserved(&self) -> Vec<String> {
self.sold()
.into_keys()
.filter(|n| self.grpc.get_repo(n).is_err())
.collect()
}
/// **A paid signup's first repository** (lane M4): declared shaped like the estate's own
/// sellable repository (the first writable, non-system one it serves), bound to the signup
/// at once, and served at once (lane M4b, `serve_now`). Idempotent for the same signup; another
/// signup's name is `taken`; a name that is no repository name is refused.
fn declare_sold(&self, repo: &str, signup: &str) -> Result<String, String> {
if !repository_name_ok(repo) {
return Err(format!("{repo:?} is no repository name (lower-case letters, digits, '-', '_' and '.', at most 64): nothing was made"));
}
let root = data_root(&self.grpc)?;
let owner = crate::sold::buyer_of(&root);
let made = crate::sold::update(&root, |sold| {
if let Some(s) = sold.get(repo) {
if s.signup != signup {
return Err(format!(
"{} {repo} is another signup's repository",
monetize_product::pull::order::TAKEN
));
}
// Asked again: a repository the last ask could not serve is tried once more.
return Ok(Err((s.template.clone(), s.owner.clone())));
}
let mut served: Vec<(String, Arc<dyn traits::RepositoryBackendTrait>)> =
self.grpc.routes.all_repos();
served.sort_by(|a, b| a.0.cmp(&b.0));
let Some(template) = served
.iter()
.find(|(n, b)| {
!self.grpc.read_gate.is_system(n) && b.is_writable() && !sold.contains_key(n)
})
.map(|(n, _)| n.clone())
else {
return Err(format!("this estate serves no repository a sold one is shaped like (a writable one that is not the estate's own): '{repo}' was not made"));
};
// ★ Lane P1: private, and whose it is. On an estate the shop sold to a buyer (the
// first-login invite was taken in), the buyer's: the estate's owners (their subject once
// they registered). Elsewhere nobody's yet: the estate's admins alone, until the person
// the signup names is bound to it.
sold.insert(
repo.to_string(),
SoldRepository {
template: template.clone(),
signup: signup.to_string(),
owner: owner.clone(),
private: true,
..Default::default()
},
);
Ok(Ok(template))
})?;
let template = match made {
Ok(template) => template,
Err((template, owner)) => {
self.serve_owned(repo, &template, &owner);
return Ok(repo.to_string());
}
};
self.grpc.record_audit(
AuditEvent::new("monetize:fleet", AuditAction::Entitlement, repo, "", "fleet", 200, 0)
.with_detail(format!("a paid signup ({signup}) made this repository, shaped like {template}, private, owned by {}", if owner.is_empty() { "nobody yet (the estate's admins)" } else { owner.as_str() })),
);
self.serve_owned(repo, &template, &owner);
Ok(repo.to_string())
}
/// **Serve a made repository at once** (lane M4b, [`crate::live_routes::LiveRoutes`]): its
/// route is swapped into its door's table, so an upload and a download reach it with no
/// restart and the `repositories_wanted` word clears. What only a start can serve (no live
/// routes, a gated or pull-through template) is said and served from the next start.
/// [`Self::serve_now`] for a repository a paid order made (lane P1): its owner is known to
/// the gate every door holds BEFORE it serves, so it is owner-only from the first request.
fn serve_owned(&self, repo: &str, template: &str, owner: &str) {
let gate = data_root(&self.grpc)
.map(|root| crate::sold::gate_owner(&root, owner))
.unwrap_or_else(|_| owner.to_owned());
self.grpc.read_gate.own(repo, &gate);
self.serve_now(repo, template, true);
}
fn serve_now(&self, repo: &str, template: &str, private: bool) {
let why = match &self.grpc.live_routes {
None => "this box serves no repository made at run time".to_string(),
Some(live) => match live.serve_as(repo, template, private) {
Ok(()) => return,
Err(why) => why,
},
};
log::warn!("{REPOSITORIES_WANTED}: '{repo}' made for a paid signup, shaped like '{template}': served from the next start ({why})");
}
fn store(&self) -> Result<&Arc<crate::metered::EntitlementStore>, String> {
self.grpc.entitlements.as_ref().ok_or_else(|| {
"this server has no monetize key, so it keeps no entitlements".to_string()
})
}
fn plane(&self) -> Result<&Arc<crate::grow::GrowPlane>, String> {
self.grpc
.grow
.as_ref()
.ok_or_else(|| "this box has no grow machine, so it grows no data set".to_string())
}
fn is_twin(&self) -> bool {
self.grpc.twin.is_twin()
}
fn state_of(s: &crate::grpc::holger_grow_proto::GrowStatusReply) -> GrowState {
let words = format!(
"{}: {}",
s.phase,
if s.detail.is_empty() {
&s.why
} else {
&s.detail
}
);
match s.phase.as_str() {
"done" => GrowState::Done(words),
"failed" => GrowState::Failed(words),
"waiting-members" => GrowState::WaitingMembers(words),
"idle" if s.serving => GrowState::Done(words),
_ => GrowState::Running(words),
}
}
/// `ok` (holding, recharged, or new and not held), `fresh` (held for its twin, holding nothing of its own), `restoring` (a
/// recharge is not whole) — `RechargeStatus`, read on the box.
fn data_word(&self) -> String {
let Some(f) = &self.grpc.follower else {
return "ok".into();
};
let root = f.root();
if root.join(crate::twin_follow::RECHARGED).exists() {
return "ok".into();
}
if root.join(crate::twin_follow::RECHARGING).exists() {
return "restoring".into();
}
// `fresh` is a DEAD main: held by its boot hold for a twin, holding nothing of its own.
// A first install that is not held (nobody to be filled from) is simply new: `ok`.
if !self.is_twin()
&& self.grpc.twin.awaiting_from().is_some()
&& f.recharge_blocker().is_none()
{
return "fresh".into();
}
"ok".into()
}
fn look(&self) -> Option<crate::follow_point::Judged> {
self.grpc.follower.as_ref().and_then(|f| f.look())
}
fn room_wanted(&self) -> Option<String> {
let root = data_root(&self.grpc).ok()?;
std::fs::read_to_string(root.join(ROOM_WANTED_REL))
.ok()
.and_then(|t| serde_json::from_str::<serde_json::Value>(&t).ok())
.and_then(|v| v.get("words").and_then(|w| w.as_str()).map(str::to_string))
}
}
impl HolgerBox for FleetBox {
fn serving(&self) -> bool {
!self.is_twin() && self.grpc.twin.serving()
}
/// **Holger's own order kinds** (monetize MONETIZE-FIX: the operator queues them on his
/// site and this main PULLS them; the ONE box never calls into the estate). Each runs
/// through the same hub the `/-/sync` doors use and answers what that door answers; the
/// bundle goes through the chain's bundles door as the order ([`FleetBox::bundle_put`]).
fn product_order(
&self,
by: &OrderedBy,
kind: &str,
params: &BTreeMap<String, String>,
file: Option<&[u8]>,
) -> Result<String, String> {
use holger_fleet::order::holger_kind as k;
let arg = |n: &str| params.get(n).cloned().unwrap_or_default();
let (actor, source) = ("monetize-operator", "a pulled operator order");
if kind == k::BUNDLE_PUT {
return self.bundle_put(
by,
&arg("repository"),
&arg("path"),
file.unwrap_or_default(),
);
}
if kind == k::ESTATE_BOARD {
let facts = self.grpc.topology_sources().gather();
return Ok(crate::topology::json(&crate::topology::board(&facts)).to_string());
}
#[cfg(feature = "sync")]
{
let hub = self
.grpc
.intake
.as_ref()
.and_then(|c| c.sync())
.ok_or_else(|| {
"this box runs no sync hub (no `sync:` in its config)".to_string()
})?;
let failed = |e: crate::sync::SyncFailure| e.tagged();
let v = match kind {
k::SYNC_KEY => {
#[allow(unused_mut)]
let mut v = serde_json::json!({ "name": hub.name(), "key": hub.public_key() });
#[cfg(feature = "downstream-wg")]
if let Some(wg) = hub.wireguard_key() {
v["wireguard_key"] = serde_json::json!(wg);
}
v
}
k::SYNC_BOARD => {
serde_json::to_value(hub.board(holger_sync::now_secs()).map_err(failed)?)
.map_err(|e| e.to_string())?
}
k::DOWNSTREAM_LIST => {
let live = hub.downstreams().map_err(failed)?;
serde_json::json!({
"upstream": hub.name(),
"key": hub.public_key(),
"downstreams": live.iter().map(|p| serde_json::json!({ "name": p.name, "key": p.key, "via": p.via.word(),
"wireguard_key": match &p.via { holger_downstream_rules::via::Via::Wireguard { key } => key.as_str(), _ => "" } })).collect::<Vec<_>>(),
"revoked": hub.revoked().map_err(failed)?,
"wireguard": hub.wg_peers().map_err(failed)?,
})
}
k::DOWNSTREAM_ENROL => {
let peer = crate::sync::downstreams::peer_from_form(
&arg("name"),
&arg("key"),
&arg("via"),
&arg("wireguard_key"),
)
.map_err(failed)?;
let p = hub.add_downstream(peer, actor, source).map_err(failed)?;
serde_json::json!({ "name": p.name, "via": p.via.word() })
}
k::DOWNSTREAM_REVOKE => {
serde_json::json!({ "revoked": hub.revoke_downstream(&arg("name"), actor, source).map_err(failed)? })
}
k::DOWNSTREAM_INVITE => {
let i = hub
.invite_for(
&arg("name"),
&arg("door"),
&arg("hub_key"),
&arg("endpoint"),
actor,
source,
)
.map_err(failed)?;
serde_json::json!({ "invite": i.signed, "address": i.address, "schema": i.schema, "door": i.door, "upstream": i.upstream, "key": i.key })
}
other => return Err(format!("Holger runs no {other:?} order")),
};
Ok(v.to_string())
}
#[cfg(not(feature = "sync"))]
{
let _ = (actor, source, arg);
Err(format!(
"this build runs no sync hub: {kind} is refused (build with `sync`)"
))
}
}
/// **A sold estate's resend sign-in link** (A16, Gunnar's S10): minted for the shop's
/// buyer in this box's passkey store and mailed through the fed relay.
#[cfg(feature = "auth-passkey")]
fn sign_in_again(&self, s: &holger_fleet::order::SignInAgain) -> Result<String, String> {
let auth = &self.grpc.auth_config;
let store = crate::passkey::door_for(auth).ok_or_else(|| {
format!(
"this box holds no passkey door, so it mints no sign-in link for {}",
s.account
)
})?;
let postman = crate::sign_in_again::postman();
crate::sign_in_again::resend(&store, auth, s, postman.as_deref())
}
/// **The estate keys the ONE box fed this box again** (MAIL-RELAY + NIGHT-PUBREPO): kept
/// under the data root, over the lay's feed. The public repositories are checked against what
/// this box serves now: a name it does not serve, and a repository that gates itself, are
/// refused by name and nothing is kept.
///
/// RA5-19: the check is the boot's. The boot makes public only a repository the declaration
/// names (`Holger::resolve_public_repositories_at_boot`), so a repository made at run time
/// (a paid sign-up's, owner-only) is no repository a refeed may name, and one that gates
/// itself (its declaration, a late gate, an owner) is never public.
fn keep_refeed(&self, r: &holger_fleet::order::Refeed) -> Result<String, String> {
let gate = &self.grpc.read_gate;
let declared = |n: &str| {
self.grpc.live_routes.as_ref().is_none_or(|l| l.declares(n))
&& gate.sold_owner(n).is_none()
};
let served: Vec<String> = self
.grpc
.routes
.all_repos()
.into_iter()
.map(|(n, _)| n)
.filter(|n| declared(n))
.collect();
let gated: Vec<String> = served
.iter()
.filter(|n| gate.repo_requires_auth(n))
.cloned()
.chain(gate.gated_repos().map(str::to_owned))
.collect();
crate::refeed::keep(&data_root(&self.grpc)?, r, &|names: &[String]| {
crate::refeed::public_check(names, &served, &gated)
})
}
/// **One generic setting the ONE box fed this box** (holger-fleet `Refeed::settings`), kept
/// where an administrator's `/-/settings/<name>` write keeps it ([`crate::settings`]) and
/// audited BEFORE it is applied, the same way (who: the ONE box's order). The server gives
/// the name no meaning: holger-fleet names it. Main and twin keep it alike, so a twin that
/// serves later holds it too.
fn keep_setting(&self, name: &str, value: &[u8]) -> Result<String, String> {
let root = data_root(&self.grpc)?;
let audit = self.grpc.audit.clone();
let path = format!("{}{name}", crate::settings::DOOR);
let previous = crate::settings::change(&root, name, value, |line| {
let ev = AuditEvent::new(
"monetize:fleet",
AuditAction::Other,
"",
&path,
"fleet",
200,
value.len() as u64,
)
.with_detail(format!("{line} (the ONE box's refeed order)"));
audit
.record(ev)
.and_then(|()| audit.flush())
.map_err(|e| format!("{e:?}"))
})?;
Ok(format!(
"the setting {name} is {} (was {}); a console that follows it reads it within its recheck",
crate::settings::words(Some(value)),
crate::settings::words(previous.as_deref())
))
}
/// The words monetize's holger guardian reads (monetize `products/holger/src/guardian.rs`).
fn health(&self) -> BTreeMap<String, String> {
let mut h = BTreeMap::new();
let p = &self.grpc.twin;
let twin = self.is_twin();
let serving = if p.fenced().is_some() {
"fenced"
} else if self.serving() {
"yes"
} else {
"no"
};
h.insert("serving".into(), serving.into());
h.insert("data".into(), self.data_word());
if *self.recharge_failed.lock_or_recover() {
h.insert("recharge".into(), "failed".into());
}
h.insert("serves_at".into(), self.own.clone());
if twin {
let (whole, why, last_whole) = p.completeness();
// A twin whose last whole cycle is older than three polls no longer hears its primary.
let fresh = last_whole > 0
&& (holger_fleet::now_unix_ms() / 1000) as i64 - last_whole
<= 3 * crate::twin::DEFAULT_POLL_SECONDS as i64 + 60;
h.insert(
"primary".into(),
if fresh { "live" } else { "silent" }.into(),
);
h.insert("complete".into(), if whole { "yes" } else { "no" }.into());
if !whole {
h.insert("why".into(), why);
}
// The pattern's twin words, every report: what the ONE box sizes the twin by.
crate::follow_point::twin_words(
&mut h,
self.look().as_ref(),
crate::follow_point::own_cpu(),
);
} else if let Some(plane) = &self.grpc.grow {
let rows = plane.twins().map(|t| t.twins).unwrap_or_default();
h.insert(
"twin".into(),
match rows.iter().map(|r| r.state.as_str()).next() {
None => "none".into(),
Some("ok" | "behind") => "paired".into(),
Some(_) => "silent".to_string(),
},
);
h.insert("epoch".into(), plane.status().epoch.to_string());
}
if let Some(w) = self.room_wanted() {
h.insert(ROOM_WANTED.into(), w);
}
let wanted = self.sold_unserved();
if !wanted.is_empty() {
h.insert(
REPOSITORIES_WANTED.into(),
format!(
"{}: made for paid signups, served from the next start",
wanted.join(", ")
),
);
}
// Each paired certificate's grade (lane E2): `expiring` and `expired` ride
// `cert_alert`, so a renewal that keeps failing reaches the guardian before the end.
h.extend(crate::cert_health::words(
data_root(&self.grpc).ok().as_deref(),
crate::cert_health::now_unix(),
));
h
}
fn product_report(&self) -> Option<String> {
if !self.serving() {
return None;
}
let g = &self.grpc;
let root = g.data_root.clone();
let mut r = ProductReport {
measured_unix_ms: holger_fleet::now_unix_ms(),
..Default::default()
};
let mut repos: Vec<String> = g
.routes
.all_repos()
.into_iter()
.map(|(name, _)| name)
.collect();
repos.sort();
for repo in &repos {
// A system repository (the sealed slot, Admin-floored) is the estate's own: its
// name is taken, it is never a tenant a buyer pays for (Gunnar leaves its
// Machine-owned namespaces out of `tenants` and keeps them in `names`). A
// tenant's name is not in `names`: a signup is bound to it by name.
if g.read_gate.is_system(repo) {
r.names.push(repo.clone());
continue;
}
// A repository a paid signup made is a person's (lane P1); the estate's own are not.
r.tenants.push(TenantOwner {
tenant: TenantId(repo.clone()),
person: g.read_gate.sold_owner(repo).is_some(),
seller: false,
});
if let Ok(counted) =
crate::metering::count(&g.routes, repo, root.as_deref(), Some(&g.storage_places))
{
let meters = counted
.into_iter()
.filter_map(|c| c.value.ok().map(|v| (c.name.to_string(), v)))
.collect();
r.usage.insert(
repo.clone(),
Usage {
meters,
measured_at_unix_ms: r.measured_unix_ms,
},
);
}
}
if let Some(root) = &root {
r.servable = Some(match crate::metering::statvfs(root) {
Ok(fs) => {
let (_, reason, _, hotplug) = crate::grow::capacity_growth(root);
Servable::Measured {
servable_bytes: fs.servable_bytes,
total_bytes: fs.total_bytes,
growth: if hotplug {
Growth::AtRuntime(reason)
} else {
Growth::Unknown(reason)
},
ceilings: BTreeMap::new(),
}
}
Err(why) => Servable::Unmeasured(why),
});
if let Ok(fs) = crate::metering::statvfs(root) {
r.room.insert(
crate::metering::METER_DISK_BYTES.to_string(),
fs.servable_bytes,
);
}
}
if let Some(store) = &g.entitlements {
for f in store.list() {
r.promised.insert(f.tenant.0.clone(), f.caps.clone());
}
}
r.metered = Some(match &g.entitlements {
Some(_) => MeteredDoor::Enforcing(format!("holger {}: a metered build enforcing monetize's signed facts, said by the box itself", self.build)),
None => MeteredDoor::Unproven(format!("holger {}: a metered build with no monetize key yet, so it serves every repository Free", self.build)),
});
serde_json::to_string(&r).ok()
}
fn own_review(&self) -> Option<String> {
if self.is_twin() {
self.review().ok().map(|v| v.to_string())
} else {
None
}
}
fn entitle(&self, fact: &EntitlementFact) -> Result<Applied, String> {
let store = self.store()?;
let repo = fact.tenant.0.clone();
// A repository a paid signup made (lane M4) takes its plan at once: it is served, with
// its fact, from the next start.
if !store.serves(&repo) && !self.sold().contains_key(&repo) {
return Err(format!("this server serves no repository '{repo}'"));
}
let unpoliced = crate::metered::caps_holger_cannot_honour(fact.caps.keys());
if !unpoliced.is_empty() {
return Err(format!(
"the fact for {repo} caps {} — meters holger measures only server-wide: refused",
unpoliced.join(", ")
));
}
if let Ok(Some(held)) = store.get(&repo) {
if held == *fact || held.issued_unix_ms > fact.issued_unix_ms {
return Ok(Applied::Held(format!(
"{repo}: this box already holds a fact as new or newer; nothing moved"
)));
}
}
store.put(fact).map_err(|e| match e {
crate::metered::PutRefused::Signature(w) => {
format!("the fact for {repo} does not verify: {w}")
}
crate::metered::PutRefused::Io(w) => {
format!("the fact for {repo} could not be stored: {w}")
}
})?;
self.grpc.record_audit(
AuditEvent::new(
"monetize:fleet",
AuditAction::Entitlement,
&repo,
"",
"fleet",
200,
0,
)
.with_detail(format!(
"state={:?} plan={} source={}",
fact.state, fact.plan, fact.source
)),
);
Ok(Applied::New(format!("{repo} is {:?}", fact.state)))
}
fn purge(&self, p: &PurgeTenant) -> Result<PurgeAnswer, String> {
let store = self.store()?;
let repo = p.tenant.as_str();
if !store.serves(repo) {
return Err(format!("this server serves no repository '{repo}'"));
}
let job = format!("holger-purge:{repo}");
let fact = store
.get(repo)
.map_err(|e| format!("read the fact of {repo}: {e}"))?;
// The ONE box decided the day (its grace); holger only refuses a repository still paid for.
crate::metered::purgeable(repo, fact.as_ref(), crate::clock::now_unix_ms())?;
let backend = self
.grpc
.get_repo(repo)
.map_err(|s| s.message().to_string())?;
let answer = match crate::metered::purge_walk(store, backend.as_ref(), repo, &p.reason) {
crate::metered::PurgeRun::Busy => PurgeAnswer {
job,
running: true,
still_locked: true,
detail: format!("a purge of `{repo}` is walking"),
..Default::default()
},
crate::metered::PurgeRun::Done {
walked,
failures,
record,
} => {
self.grpc.record_audit(
AuditEvent::new(
"monetize:fleet",
AuditAction::Purge,
repo,
"",
"fleet",
if failures.is_empty() { 200 } else { 500 },
walked.bytes,
)
.with_detail(format!(
"{} artifact(s) deleted; reason: {}; record {record}",
walked.deleted, p.reason
)),
);
PurgeAnswer {
job,
running: false,
failed: !failures.is_empty(),
error: failures.first().cloned().unwrap_or_default(),
detail: format!(
"{} artifact(s), {} byte(s) deleted from `{repo}`",
walked.deleted, walked.bytes
),
bytes_reclaimed: Some(walked.bytes),
still_locked: false,
failures,
}
}
};
Ok(answer)
}
/// **A paid signup is a repository on a plan.** The repository is one this server serves
/// (the operator declares repositories; a sold one is in the seed's config); the signup is
/// bound to it here, once, so monetize asking again answers the same; the plan arrives as
/// the repository's entitlement fact. A repository bound to another signup is `taken`.
fn add_principal(&self, a: &AddPrincipal) -> Result<String, String> {
let repo = a.username.trim();
if self.grpc.get_repo(repo).is_err() {
// ★ Lane M4: a paid signup MAKES its first repository (Gunnar's AddPrincipal makes
// the account and its namespace).
return self.declare_sold(repo, a.signup.trim());
}
if self.grpc.read_gate.is_system(repo) {
return Err(format!(
"{} {repo} is a system repository of this estate, never sold",
monetize_product::pull::order::TAKEN
));
}
let path = data_root(&self.grpc)?.join(SIGNUPS_REL);
let mut bound: BTreeMap<String, String> = std::fs::read(&path)
.ok()
.and_then(|b| serde_json::from_slice(&b).ok())
.unwrap_or_default();
match bound.get(repo) {
Some(s) if s == a.signup.trim() => return Ok(repo.to_string()),
Some(_) => {
return Err(format!(
"{} {repo} is another signup's repository",
monetize_product::pull::order::TAKEN
))
}
None => {}
}
bound.insert(repo.to_string(), a.signup.trim().to_string());
let bytes = serde_json::to_vec_pretty(&bound).map_err(|e| e.to_string())?;
std::fs::create_dir_all(path.parent().unwrap_or(&path))
.and_then(|_| std::fs::write(&path, bytes))
.map_err(|e| format!("{}: {e}", path.display()))?;
self.grpc.record_audit(
AuditEvent::new(
"monetize:fleet",
AuditAction::Entitlement,
repo,
"",
"fleet",
200,
0,
)
.with_detail(format!(
"a paid signup ({}) is this repository on a plan",
a.signup
)),
);
Ok(repo.to_string())
}
fn want_room(&self, g: &GrowRoom) -> Result<String, String> {
let root = data_root(&self.grpc)?;
let gib = g.need_bytes as f64 / (1u64 << 30) as f64;
let words = format!("{gib:.1} GiB of stored room was committed past this set ({}) — queued: grow the data set", g.reason);
let body = serde_json::json!({ "need_bytes": g.need_bytes, "reason": g.reason, "words": words, "at_unix_ms": holger_fleet::now_unix_ms() });
let path = root.join(ROOM_WANTED_REL);
std::fs::create_dir_all(path.parent().unwrap_or(&root))
.and_then(|_| std::fs::write(&path, body.to_string()))
.map_err(|e| format!("{}: {e}", path.display()))?;
log::warn!("{ROOM_WANTED}: {words}");
Ok(format!(
"kept: {gib:.1} GiB wanted past the set; the operator is told ({ROOM_WANTED})"
))
}
fn grow_carried(&self, data_gib: u64) -> Result<bool, String> {
let s = self.plane()?.status();
let target = data_gib.saturating_mul(SECTORS_PER_GIB);
Ok(s.members_verdict == "complete"
&& !s.members.is_empty()
&& s.members
.iter()
.all(|m| appliance_grow::target_reached(m.set_bytes / 512, target)))
}
fn twin_holds(&self, data_gib: u64) -> Result<String, String> {
let plane = self.plane()?;
let members = plane.status().members.len().max(1) as u64;
let want = members.saturating_mul(data_gib).saturating_mul(1 << 30);
let rows = plane.twins()?.twins;
if rows.is_empty() {
return Err("the main grows only after its twin holds its set, and no twin reports to this primary: nothing moved".into());
}
match rows.iter().find(|r| r.state == "ok" && r.disk_total_bytes >= want) {
Some(r) => Ok(format!("the twin {} holds {} GiB (main's set: {} GiB)", r.address, r.disk_total_bytes >> 30, want >> 30)),
None => Err(format!("no twin of this primary holds main's whole set of {} GiB yet (it grows first): nothing moved", want >> 30)),
}
}
fn grow_start(&self, data_gib: u64, nonce: &str) -> Result<GrowState, String> {
// The step was verified under the monetize key by the executor: the go-ahead the
// machine takes is minted here, as the admin door's operator one.
let target = data_gib.saturating_mul(SECTORS_PER_GIB);
let go = serde_json::json!({ "v": 1, "signer": "operator", "target_sectors": target, "nonce": nonce, "issued_unix_ms": crate::clock::now_unix_ms(), "signature": "" });
let req = crate::grpc::holger_grow_proto::GrowStartRequest {
target_sectors_per_member: target,
go_ahead: go.to_string().into_bytes(),
drain_timeout_secs: 0,
wait_members_timeout_secs: 0,
};
self.plane()?
.start(&req)
.map(|s| Self::state_of(&s))
.map_err(|why| format!("the grow machine refused monetize's signed grow step: {why}"))
}
fn grow_settle(&self, past_waiting: bool) -> GrowState {
let Ok(plane) = self.plane() else {
return GrowState::Failed("this box has no grow machine".into());
};
let until = std::time::Instant::now() + GROW_CEILING;
loop {
let state = Self::state_of(&plane.status());
match &state {
GrowState::Done(_) | GrowState::Failed(_) => return state,
GrowState::WaitingMembers(_) if !past_waiting => return state,
_ if std::time::Instant::now() >= until => return state,
_ => std::thread::sleep(Duration::from_secs(2)),
}
}
}
fn grow_look_again(&self, give_up: bool) -> Result<(), String> {
self.plane()?.resume(give_up).map(|_| ())
}
/// **The one recharge path** (Gunnar's pattern: a single pulled order): holger's recharge
/// lane, all of it — the four refusals, the main certified against the twin, `Recharge`
/// (which releases the main's boot hold once it is whole), and the twin re-pointed at the
/// new main. Every box is dialled at its overlay address (`tunnel_addr`, lane E1). A
/// partial pull stays held and resumes on the next order.
fn recharge(&self, r: &RechargeAsk) -> Result<String, String> {
use crate::recharge_lane::Outcome;
// Fence before recharge (A7 row 4): holger-fleet's executor ran the order's own fence
// first and reaches here only when it held; the guardian's cloud proof rides as
// `fenced`. Either is the lane's proof; with neither, the lane refuses not_fenced.
let fenced_at = r
.fence
.as_deref()
.map(str::trim)
.filter(|f| !f.is_empty())
.map(str::to_string);
let cloud = r.fenced.as_deref().is_some_and(|p| !p.trim().is_empty());
let proof = crate::recharge_lane::Proof {
primary_fenced: fenced_at.is_some() || cloud,
fenced_at: if cloud { None } else { fenced_at },
..Default::default()
};
let done = crate::recharge_lane::run_grpc(
self.rt.clone(),
self.bearer.now(),
&r.main,
Some(&r.twin),
r.accept_lost,
&proof,
);
let failed = matches!(done, Ok(Outcome::Resumable { .. }));
{
let mut f = self.recharge_failed.lock_or_recover();
*f = failed;
}
match done? {
Outcome::Recharged { artifacts, twin_follows, .. } => Ok(format!("recharged {} from {} ({} repositories); the twin follows it at {twin_follows}", r.main, r.twin, artifacts.len())),
Outcome::Resumable { failed } => Err(format!("the recharge of {} moved bytes but is not whole ({} failed); it stays held and resumes on the next order", r.main, failed.len())),
Outcome::AlreadyRecharged => Ok(format!("{} was recharged before: nothing moved", r.main)),
}
}
/// `fence_primary` (and a recharge's fence-first): the old primary refuses every write
/// from now on — this box ITSELF when the order names it (or no box), else the peer it
/// names, over that box's admin door, read back from its answer.
fn fence(&self, primary: &str, epoch: u64) -> Result<String, String> {
let primary = primary.trim();
if !primary.is_empty() && primary != self.own {
return crate::twin_tls::GrpcBoxes::new(self.rt.clone(), self.bearer.now())
.fence_primary(primary, epoch);
}
if self.is_twin() {
return Err("this box is a twin: a twin takes no writes and is never fenced".into());
}
self.grpc.twin.fence(epoch);
log::warn!("fleet: this primary is FENCED at epoch {epoch}: every door refuses writes");
Ok(format!(
"{} is fenced at epoch {epoch}: every door refuses writes",
self.own
))
}
/// `become_twin`: holger's own `BecomeTwin` election on this (fresh) box.
fn become_twin(&self, primary: &str) -> Result<String, String> {
use nordisk_twin::certify::Boxes as _;
let boxes = crate::twin_tls::GrpcBoxes::new(self.rt.clone(), self.bearer.now());
boxes.become_twin(&self.own, primary).map(|w| {
format!(
"this box is the twin of {primary}{}",
if w.is_empty() {
String::new()
} else {
format!(" ({w})")
}
)
})
}
fn adopt(&self, twin: &str, principal: &str) -> Result<String, String> {
let boxes = crate::twin_tls::GrpcBoxes::new(self.rt.clone(), self.bearer.now());
let principal = if principal.trim().is_empty() {
crate::twin_tls::TWIN_PRINCIPAL.to_string()
} else {
principal.to_string()
};
let t = nordisk_twin::certify::Target {
reader_ctl: twin.to_string(),
source_ctl: self.own.clone(),
become_twin_of: Some(self.own.clone()),
principal,
purpose: "twin".into(),
};
nordisk_twin::certify::once(&boxes, &t).map(|p| {
format!(
"the twin at {twin} follows {} ({} days of certificate)",
p.replication, p.days_left
)
})
}
fn review(&self) -> Result<serde_json::Value, String> {
let p = &self.grpc.twin;
if self.is_twin() {
let (complete, why, last_whole) = p.completeness();
let verdict = if complete {
"the twin is whole".to_string()
} else {
format!("the twin is not whole: {why}")
};
let mut v = serde_json::json!({ "role": "twin", "complete": complete, "why": why, "last_whole_unix": last_whole, "follow": p.follow(), "primary": p.role().primary().unwrap_or_default(), "verdict": verdict, "reviewed_unix_ms": holger_fleet::now_unix_ms() });
if let (Some(j), Some(o)) = (self.look(), v.as_object_mut()) {
o.extend(crate::follow_point::review_point(
j.state,
j.applied.saturating_add(j.lag),
j.applied,
));
}
return Ok(v);
}
let rows = self.plane()?.twins()?.twins;
let verdict = match rows.first() {
None => "no twin reports to this primary".to_string(),
Some(r) => format!(
"the twin {} is {} ({} entries behind)",
r.address, r.state, r.lag_entries
),
};
let twins: Vec<serde_json::Value> = rows.iter().map(|r| serde_json::json!({ "address": r.address, "state": r.state, "lag_entries": r.lag_entries, "disk_total_bytes": r.disk_total_bytes, "seen_ago_ms": r.seen_ago_ms })).collect();
let mut v = serde_json::json!({ "role": "main", "twins": twins, "verdict": verdict, "reviewed_unix_ms": holger_fleet::now_unix_ms() });
// The review's seqs: the twin furthest behind, under the worst state of the roster.
if let (Some(r), Some(o)) = (rows.iter().max_by_key(|r| r.lag_entries), v.as_object_mut()) {
let state = rows.iter().fold("ok", |w, r| {
crate::follow_point::worse(
w,
match r.state.as_str() {
"ok" => "ok",
"behind" => "behind",
"stalled" => "stalled",
_ => "silent",
},
)
});
o.extend(crate::follow_point::review_point(
state,
r.head_seq,
r.applied_seq,
));
}
Ok(v)
}
}
// ── the meter plugin's key ──────────────────────────────────────────────────
/// **Where the commercial meter plugin reads monetize's verifying key** in the meter plane's
/// store (monetize-impl `monetize-plugin-holger::KEY_SLOT`): the shop's product key for holger,
/// fed at load from what the box's seed / feed handed it (`monetize_pubkey`), never compiled in.
pub const METER_KEY_SLOT: &str = "monetize/key.hex";
/// Pure: the key slot's bytes for `monetize_pubkey` (64 hex or base64), lower-case hex;
/// `None` when no usable key was handed to this box (the plugin then refuses every fact by name).
pub fn meter_key_bytes(monetize_pubkey: Option<&str>) -> Option<Vec<u8>> {
holger_fleet::key32(monetize_pubkey?).map(|k| holger_fleet::hex(&k).into_bytes())
}
// ── the pull channel's start ────────────────────────────────────────────────
static SERVED: std::sync::OnceLock<(Arc<HolgerGrpc>, String)> = std::sync::OnceLock::new();
/// The served gRPC state and its address, published by the first endpoint that starts.
pub fn served(grpc: &Arc<HolgerGrpc>, addr: &str) {
let _ = SERVED.set((Arc::clone(grpc), addr.to_string()));
{
let mut all = SERVED_AT.lock_or_recover();
all.insert(addr.to_string(), Arc::clone(grpc));
}
}
/// Every endpoint's served state by its address: several boxes in one process (tests).
static SERVED_AT: std::sync::Mutex<BTreeMap<String, Arc<HolgerGrpc>>> =
std::sync::Mutex::new(BTreeMap::new());
/// The served state of the endpoint at `addr` (tests that run more than one box).
pub fn served_at(addr: &str) -> Option<Arc<HolgerGrpc>> {
SERVED_AT.lock_or_recover().get(addr).cloned()
}
/// The served state, once an endpoint started (tests, the CLI's enrolment).
pub fn served_state() -> Option<(Arc<HolgerGrpc>, String)> {
SERVED.get().cloned()
}
#[cfg(test)]
mod meter_key_tests {
use super::*;
#[test]
fn the_seeds_monetize_key_is_fed_to_the_meter_plugin_as_lower_hex() {
let hex = "AB".repeat(32);
assert_eq!(
meter_key_bytes(Some(&format!(" {hex} "))),
Some("ab".repeat(32).into_bytes())
);
let b64 = monetize_embed::signing::base64_encode(&[0xab; 32]);
assert_eq!(
meter_key_bytes(Some(&b64)),
Some("ab".repeat(32).into_bytes()),
"the base64 spelling too"
);
assert_eq!(meter_key_bytes(Some("nope")), None);
assert_eq!(meter_key_bytes(None), None);
}
}
/// Where this box's own fleet jobs are recorded (a jera ledger of its own).
pub const JOBS_REL: &str = "fleet/jobs";
/// The door a feed names: what a change of re-dials (the key is followed on its own).
#[derive(Debug, Clone, PartialEq, Eq)]
struct Door {
url: String,
at: Option<String>,
ca_pem: Option<String>,
instance: String,
}
impl Door {
fn of(c: &holger_fleet::FleetConfig) -> Door {
Door {
url: c.url.clone(),
at: c.at.clone(),
ca_pem: c.ca_pem.clone(),
instance: c.instance.clone(),
}
}
}
/// What the running pull channel holds that a later feed changes.
#[derive(Default)]
struct Live {
/// The round's duty, once the served state is up.
duty: Option<Arc<holger_fleet::duty::Duty>>,
/// The door the channel was started on (or re-dialled to).
door: Option<Door>,
/// The key orders verify under: the feed's, else the configured one.
key: Option<[u8; 32]>,
/// Said once: the box waits for a door.
said_waiting: bool,
}
/// The bearer a repair order runs under: the environment's, else the feed's `secret.repair_bearer`.
pub fn repair_bearer(feed: &BTreeMap<String, String>) -> Option<String> {
std::env::var("HOLGER_REPAIR_BEARER")
.ok()
.or_else(|| feed.get("secret.repair_bearer").cloned())
.filter(|b| !b.trim().is_empty())
}
/// **The role this box pulls as, asked at every round** (A7 row 1, Gunnar b009cbcc):
/// nordisk-pairing `elect::pull_role` (money-free, shared with Gunnar) over the configured role (`flag`), this box's twin
/// record (a box configured as a twin, or one ELECTED twin after its start) and the role
/// the installer's door elected. `None` while the door has not elected one: the round waits.
pub fn role_now(
flag: String,
twin: bool,
data_root: PathBuf,
feeds: Arc<dyn crate::late::Feeds>,
) -> holger_fleet::duty::RoleNow {
Arc::new(move || {
let recorded = crate::twin::recorded_role(&data_root)
.ok()
.flatten()
.is_some_and(|r| r.role.is_twin());
let door = feeds.door_role();
nordisk_pairing::elect::pull_role(
&flag,
twin || recorded,
door.as_ref().map(|d| d.as_deref()),
)
})
}
/// **Start the pull channel on the box's feed bus** (A7 rows 1 and 8): the way Gunnar's main
/// does (gunnar-server `repair::fleet_when_fed`, `feed::on_published`). The shop's pairing
/// feeds the ONE monetize's door to the FRONT alone, the front relays it to every box as its
/// offer (`offer.front.fleet`, …), and the installer feeds a box that offer on its NEXT visit,
/// after the box already serves (test3-holger-rs 2026-10-20: read once at the start, the main
/// never reported). So the channel subscribes:
///
/// * the door (`fleet`, `fleet_instance`, `fleet_at`, `fleet_ca`, plain or offered): the
/// channel starts the moment a feed names one, and re-dials when a later feed names
/// another; with none fed, what an enrolment kept (`fleet/fleet.json`);
/// * the monetize key: orders verify under it from the next round;
/// * `secret.repair_bearer`: the next repair order carries it;
/// * the role is asked at every round ([`role_now`]): never reported under a guess.
///
/// `twin` is the configured role (the RON). `configured_key` is the RON's `monetize_pubkey`.
/// `say` tells the boot log what the channel does. `Ok(true)` when it reports now,
/// `Ok(false)` when it is off or waits for the feed.
pub fn start_on(
feeds: Arc<dyn crate::late::Feeds>,
data_root: PathBuf,
twin: bool,
configured_key: Option<String>,
say: fn(&str),
) -> Result<bool, String> {
let kept = holger_fleet::FleetConfig::load(&data_root).map_err(|e| format!("fleet: {e}"))?;
let rt = tokio::runtime::Handle::try_current()
.map_err(|_| "fleet: no runtime to run the pull channel on".to_string())?;
let configured = configured_key.as_deref().and_then(holger_fleet::key32);
let live: Arc<std::sync::Mutex<Live>> = Arc::new(std::sync::Mutex::new(Live {
key: configured.or_else(|| kept.as_ref().and_then(|c| c.key())),
..Default::default()
}));
let bearer = LiveBearer::default();
// The repair bearer, live.
let b = bearer.clone();
crate::late::on_change(
feeds.as_ref(),
"fleet: the repair bearer",
repair_bearer,
move |v: &Option<String>| b.set(v.clone()),
);
// The key orders verify under, live: the feed's, else the configured one.
let l = Arc::clone(&live);
crate::late::on_change(
feeds.as_ref(),
"fleet: the monetize key",
crate::late::monetize_key,
move |k: &Option<[u8; 32]>| {
let mut st = l.lock_or_recover();
let next = k.or(configured);
if next == st.key {
return;
}
st.key = next;
if let Some(d) = &st.duty {
d.set_key(next);
log::warn!("fleet: the monetize key was fed after the start: orders verify under it from the next round");
}
},
);
// The door, live.
let (l, f, root, h) = (
Arc::clone(&live),
Arc::clone(&feeds),
data_root.clone(),
rt.clone(),
);
let on_estate = feeds.on_estate();
crate::late::on_change(
feeds.as_ref(),
"fleet: the ONE monetize's door",
|feed: &BTreeMap<String, String>| holger_fleet::FleetConfig::from_feed(feed),
move |cfg: &Option<holger_fleet::FleetConfig>| {
let cfg = match (cfg, &kept) {
(Some(c), _) => c.clone(),
(None, Some(k)) => k.clone(),
(None, None) => {
let mut st = l.lock_or_recover();
if st.door.is_none() && !st.said_waiting {
st.said_waiting = true;
if on_estate {
say("no ONE monetize named yet: the box reports once its feed names one (the front's offer)");
log::log!(holger_fleet::duty::SAID_AT, "fleet: the feed names no ONE monetize yet (`fleet`/`fleet_instance`, plain or the front's offer): the channel starts when one is fed");
} else {
say("no ONE monetize named: nothing reports");
log::log!(holger_fleet::duty::SAID_AT, "fleet: no ONE monetize is named (feed `fleet`/`fleet_instance`, or an enrolment): the pull channel is off");
}
}
return;
}
};
let door = Door::of(&cfg);
let mut st = l.lock_or_recover();
match st.door.clone() {
Some(d) if d == door => {}
Some(was) => match holger_fleet::grpc::channel(&cfg) {
Ok(channel) => {
st.door = Some(door);
match &st.duty {
Some(duty) => {
duty.relink(Box::new(holger_fleet::grpc::TonicLink::new(channel, h.clone())), &cfg.instance);
log::warn!("fleet: the feed names another ONE monetize door ({} for {}, was {} for {}): the next round dials it", cfg.url, cfg.instance, was.url, was.instance);
}
None => log::warn!("fleet: the feed names another ONE monetize door before the first round ({}): the channel starts on the first one", cfg.url),
}
}
Err(why) => log::error!(
"fleet: the feed's new door {} is refused ({why}); the channel keeps {}",
cfg.url,
was.url
),
},
None => {
st.key = st.key.or_else(|| cfg.key());
match launch(
&cfg,
root.clone(),
twin,
bearer.clone(),
Arc::clone(&f),
Arc::clone(&l),
&h,
) {
Ok(()) => {
st.door = Some(door);
say(if st.said_waiting {
"the feed named the ONE monetize: the pull channel reports to it"
} else {
"the pull channel reports to the ONE monetize"
});
}
Err(why) => {
log::error!("fleet: the pull channel is off: {why}");
say(&format!("the pull channel is off: {why}"));
}
}
}
}
},
);
let on = live.lock_or_recover().door.is_some();
Ok(on)
}
/// **Start the pull channel** on `cfg`: waits (as a task) for the gRPC state to be served,
/// then runs the rounds as jera jobs; the duty is kept in `live` for later feeds.
fn launch(
cfg: &holger_fleet::FleetConfig,
data_root: PathBuf,
twin: bool,
bearer: LiveBearer,
feeds: Arc<dyn crate::late::Feeds>,
live: Arc<std::sync::Mutex<Live>>,
rt: &tokio::runtime::Handle,
) -> Result<(), String> {
// A sold estate's resend sign-in link is mailed through the relay the shop fed this box
// (A16, Gunnar's S10), read from the feed in force at each mail.
#[cfg(feature = "auth-passkey")]
{
let fed = Arc::clone(&feeds);
// MAIL-RELAY: the relay the ONE box fed again (kept under the data root) goes over it.
let root = data_root.clone();
crate::sign_in_again::set_postman(Arc::new(crate::sign_in_again::FedRelay {
fed: move || crate::refeed::over(fed.now(), &root),
}));
}
// The lean estate: a box behind its front leaves through the egress door it is fed.
{
let fed = Arc::clone(&feeds);
crate::egress::read_feed_from(move || fed.now());
}
let role = role_now(cfg.role.clone(), twin, data_root.clone(), feeds);
let first = role().unwrap_or_else(|| "main".into());
let me = holger_fleet::Addressee {
product: holger_fleet::PRODUCT.into(),
instance: cfg.instance.clone(),
role: first,
};
let build = format!(
"{} ({})",
env!("CARGO_PKG_VERSION"),
env!("HOLGER_GIT_HASH")
);
let puller = holger_fleet::Puller::at(&data_root, me.clone(), build.clone())
.map_err(|e| format!("fleet: this box's key or state could not be read: {e}"))?;
crate::egress::check("the ONE monetize pull channel", "the ONE box")
.map_err(|e| e.to_string())?;
let channel = holger_fleet::grpc::channel(cfg)?;
let jobs = jera::jobs::JobStore::open(&data_root.join(JOBS_REL))
.map_err(|e| format!("fleet: the job ledger: {e}"))?;
let link = Box::new(holger_fleet::grpc::TonicLink::new(channel, rt.clone()));
log::log!(holger_fleet::duty::SAID_AT, "fleet: reporting to {} as {}/{} (the role is read at every round; box key {}); monetize never calls in", cfg.url, me.product, me.instance, holger_fleet::hex(&puller.box_public()));
let handle = rt.clone();
// A32-14, stays a task: the fleet link's round loop for the box's life; what it runs is kept in jera's job ledger.
rt.spawn(async move {
let (grpc, own) = loop {
if let Some((g, a)) = SERVED.get() {
break (Arc::clone(g), a.clone());
}
tokio::time::sleep(Duration::from_millis(500)).await;
};
let boxed: Arc<dyn HolgerBox + Send + Sync> = Arc::new(FleetBox {
grpc,
own,
rt: handle.clone(),
bearer,
build,
recharge_failed: Default::default(),
});
let duty = {
let mut st = live.lock_or_recover();
let duty = holger_fleet::duty::Duty::with_role(
puller,
link,
Arc::new(holger_fleet::Pumped(boxed)),
st.key,
Some(role),
);
st.duty = Some(Arc::clone(&duty));
duty
};
let _queue = holger_fleet::duty::start(&handle, jobs.sink(), duty);
std::future::pending::<()>().await;
});
Ok(())
}