crates/gunnar-server/src/repair.rs
//! **The estate's repair, on the box** (5.4 resilience, owner 2026-10-10): the box
//! repairs itself where it can, through the ONE implementation of the verbs
//! (`gunnar-repair`). Two callers live here, both handing the same [`ServerEstate`]
//! to the crate:
//!
//! * the **boot check** ([`boot`]): this box came up with its data set dead or empty
//! while a twin exists — it says `data lost; recharge from twin available`, and it
//! recharges by itself ONLY when the estate opted into auto-fix (`--repair-auto`,
//! OFF by default, like `--watch-primary`);
//! * the **pull channel** ([`pull_loop`]): the box REPORTS to monetize's `Fleet`,
//! PULLS the orders queued for it, verifies each under the monetize key and runs it
//! through `gunnar_repair::exec::execute`; the ack rides the next report. Monetize
//! never calls in: no order reaches a box that does not fetch it.
//!
//! **Whose key runs a verb.** Every call is signed with the box's repair key
//! (`--repair-key`, else the twin's own `twin.key`): reading the twin's log needs a
//! member of group `twin`; a fresh main's `Bootstrap.Restore` takes only the keys it
//! was fed (`--restore-pubkey`, the installer's admin / monetize / console keys). A
//! twin's `twin.key` therefore recharges a main only when the installer fed it to the
//! re-laid main as a restore key; otherwise the front's console runs the recharge.
use std::collections::BTreeMap;
use std::path::PathBuf;
use std::sync::Arc;
use std::time::Duration;
use gunnar_core::ssh_key::PrivateKey;
use gunnar_grpc::client_signer::ControlChannel;
use gunnar_repair::exec::{self, BootReport, BootStep, Estate};
#[cfg(feature = "fleet-pull")]
use gunnar_repair::fleet;
use gunnar_repair::proto::{ProtoFence, ProtoMain, ProtoTwin};
use gunnar_repair::{Fault, Refusal, RefusalCode};
use crate::stores::ServerStores;
/// The log target of everything here.
pub const TARGET: &str = "gunnar::repair";
/// The longest pause between reports on a door the feed named live (`--fleet-every-s`'s default).
pub const FED_EVERY: Duration = Duration::from_secs(60);
/// How the box repairs itself (`--repair-*`, `--fleet*`).
#[derive(Debug, Clone, Default)]
pub struct RepairConfig {
/// **Auto-fix**: the boot check recharges by itself. The estate's opt-in, OFF by default.
pub auto: bool,
/// The repair key (OpenSSH private key). `None`: this box's `twin.key`, when it has one.
pub key: Option<PathBuf>,
/// The twin to recharge from. `None`: the twin the installer's re-admit named
/// (`<data>/pairing/recharge_from`), if any.
pub twin: Option<String>,
/// The pull channel to monetize; `None` = off.
pub fleet: Option<FleetConfig>,
}
/// Where and as whom this box pulls.
#[derive(Debug, Clone)]
pub struct FleetConfig {
/// monetize's `Fleet` endpoint (`https://…`; `http://` only to a loopback address).
pub url: String,
/// PEM of the CA monetize's certificate is signed by; `None` for `https://` = the public roots.
pub ca: Option<PathBuf>,
/// The CA as fed ([`crate::feed::FEED_FLEET_CA`]), PEM text; used when [`Self::ca`] is `None`.
pub ca_pem: Option<String>,
/// Where to connect (`host:port`, [`crate::feed::FEED_FLEET_AT`]); `None` = the URL's own
/// host. TLS verifies the URL's host either way.
pub at: Option<String>,
pub instance: String,
/// `main` / `twin`; empty = what this box is ([`crate::twin::is_twin_root`]).
pub role: String,
/// The key orders are verified under; `None` = the monetize key in force
/// ([`crate::feed::monetize_in_force`]), which the installer may feed later.
pub monetize_pubkey: Option<[u8; 32]>,
/// The longest pause between reports when monetize names none.
pub every: Duration,
}
/// **How this box reaches the others**, every call signed with its repair key.
pub struct ServerEstate {
pub key: PrivateKey,
/// This box's own control plane, as another box dials it.
pub own: String,
/// What `restate_entitlements` reaches ([`pull_loop`] installs the box's own).
pub entitlements: Entitlements,
}
/// ★ UW-4: the `Admin.Health` alarm a `grow_room` order raises.
pub const ROOM_WANTED: &str = "grow.room_wanted";
/// **What `restate_entitlements` reaches on this box**: its stores (does it serve?)
/// and, in a metered build, the entitlement slot the facts go through — the same
/// `Entitlement.Set` path, newer-wins, nothing deleted. `Default`: none, refused by name.
#[derive(Clone, Default)]
pub struct Entitlements {
stores: Option<Arc<ServerStores>>,
/// The box's own control plane, read in-process for `Report.product`.
control: Option<Arc<dyn gunnar_grpc::ControlPlane>>,
#[cfg(feature = "metered")]
slot: Option<Arc<crate::entitlement::EntitlementSlot>>,
/// This box's own grow machine, for a pulled `grow_go_ahead` (G16-d).
grow: Option<crate::paid_growth::PaidGrowth>,
/// `--repair-twin`: the twin a main waits for before it grows.
twin: Option<String>,
/// As whom the pull channel pulls ([`Entitlements::pulled_as`]).
instance: String,
role: String,
monetize_key: Option<[u8; 32]>,
/// S10: how this main mails a resend sign-in link it minted ([`crate::postman`]).
#[cfg(feature = "fleet-pull")]
postman: Option<Arc<dyn crate::postman::Postman>>,
}
impl Entitlements {
pub fn of(stores: Arc<ServerStores>) -> Entitlements {
Entitlements {
stores: Some(stores),
..Entitlements::default()
}
}
/// The control plane `Report.product` is read from.
pub fn with_control(mut self, control: Arc<dyn gunnar_grpc::ControlPlane>) -> Entitlements {
self.control = Some(control);
self
}
/// **`Report.product`** (ONE monetize, G1 pull): JSON text of
/// `monetize_product::pull::ProductReport`, on a main that serves; `None` otherwise.
pub fn product(&self, now_unix_ms: u64) -> Option<String> {
let (stores, control) = (self.stores.as_ref()?, self.control.as_ref()?);
if health(stores, "").get("serving").map(String::as_str) != Some("yes") {
return None;
}
let (promised, metered) = self.promised_and_door();
let mut report =
crate::product_report::read(control.as_ref(), promised, metered, now_unix_ms);
report.accounts = crate::product_report::accounts_of(control.as_ref(), stores, &report);
report.activity =
crate::product_report::activity_of(control.as_ref(), &report, now_unix_ms);
serde_json::to_string(&report)
.map_err(|e| tracing::warn!(target: TARGET, "Report.product does not serialize: {e}"))
.ok()
}
/// Each tenant's enforced caps, and the paid door as this build knows it.
fn promised_and_door(
&self,
) -> (
BTreeMap<String, BTreeMap<String, u64>>,
crate::product_report::wire::MeteredDoor,
) {
use crate::product_report::wire::MeteredDoor;
#[cfg(feature = "metered")]
if let Some(slot) = &self.slot {
let promised = match slot.list() {
Ok(facts) => facts
.into_iter()
.map(|f| (f.namespace, f.caps.into_iter().collect()))
.collect(),
Err(e) => {
tracing::warn!(target: TARGET, "Report.product: the entitlements could not be listed: {e}");
BTreeMap::new()
}
};
let door = if slot.keyed() {
MeteredDoor::Enforcing(format!("gunnar {}: a metered build enforcing monetize's signed facts, said by the box itself", env!("CARGO_PKG_VERSION")))
} else {
MeteredDoor::Unproven(format!("gunnar {}: a metered build with no monetize key yet, so it serves every tenant Free", env!("CARGO_PKG_VERSION")))
};
return (promised, door);
}
#[cfg(feature = "metered")]
let door = MeteredDoor::Unproven(format!("gunnar {}: a metered build started without its entitlement slot (no --monetize-pubkey)", env!("CARGO_PKG_VERSION")));
#[cfg(not(feature = "metered"))]
let door = MeteredDoor::Absent(format!(
"gunnar {}: a community build, no metered services compiled in",
env!("CARGO_PKG_VERSION")
));
(BTreeMap::new(), door)
}
/// The grow machine a pulled `grow_go_ahead` drives.
pub fn with_grow(mut self, grow: crate::paid_growth::PaidGrowth) -> Entitlements {
self.grow = Some(grow);
self
}
/// The twin a main waits for (`--repair-twin`); `None` = the installer's re-admit, else the
/// twin the door named this main ([`crate::pairing::door_twin`]).
pub fn with_twin(mut self, twin: Option<String>) -> Entitlements {
self.twin = twin.filter(|t| !t.trim().is_empty());
self
}
/// How this main mails (`sign_in_again`); `None` = no relay, a resend refused by name.
#[cfg(feature = "fleet-pull")]
pub fn with_postman(
mut self,
postman: Option<Arc<dyn crate::postman::Postman>>,
) -> Entitlements {
self.postman = postman;
self
}
/// The estate (`--fleet-instance`), role and order key the pull channel runs as.
pub fn pulled_as(
mut self,
instance: &str,
role: &str,
monetize_key: Option<[u8; 32]>,
) -> Entitlements {
self.instance = instance.to_owned();
self.role = role.to_owned();
self.monetize_key = monetize_key;
self
}
fn paid(&self) -> Result<&crate::paid_growth::PaidGrowth, Refusal> {
self.grow
.as_ref()
.ok_or_else(|| crate::paid_growth::refuse(RefusalCode::BadOrder, "this box has no grow machine on its pull channel, so it grows no data set: grow_go_ahead refused"))
}
/// `grow_go_ahead` step 1: monetize's step verified for THIS estate and role.
pub fn grow_check(
&self,
go_ahead: &str,
data_gib: u64,
) -> Result<gunnar_repair::exec::GrowCheck, Refusal> {
let key = self.monetize_key.or_else(crate::feed::monetize_in_force);
self.paid()?.check(
go_ahead,
data_gib,
&self.instance,
key.as_ref(),
self.role == "twin",
crate::twin::now_ms(),
)
}
/// The twin first: its `Bootstrap.GrowStatus` must say it holds this main's set.
pub async fn twin_holds(
&self,
check: &gunnar_repair::exec::GrowCheck,
) -> Result<String, Refusal> {
let paid = self.paid()?;
let Some(addr) = self.twin.clone().or_else(|| {
self.stores
.as_ref()
.and_then(|s| awaited(s).or_else(|| door_twin(s)))
}) else {
return Err(crate::paid_growth::refuse(
RefusalCode::NoTwin,
"the main grows only after its twin holds its set, and this box knows no twin address (--repair-twin, the installer's recharge_from, or a twin in the door's peer list): nothing moved",
));
};
let unreachable = |why: String| {
crate::paid_growth::refuse(
RefusalCode::Unreachable,
format!("the twin at {addr} did not say what it holds ({why}): nothing moved"),
)
};
let channel = dial(&addr).map_err(|f| unreachable(f.to_string()))?;
let status = gunnar_proto::bootstrap_client::BootstrapClient::new(channel)
.grow_status(gunnar_proto::GrowStatusRequest::default())
.await
.map_err(|s| unreachable(format!("{}: {}", s.code(), s.message())))?
.into_inner();
crate::paid_growth::twin_holds(&status, paid.members(), check.data_gib)
.map(|said| format!("the twin at {addr}: {said}"))
}
pub fn grow_start(
&self,
check: &gunnar_repair::exec::GrowCheck,
nonce: &str,
) -> Result<gunnar_repair::exec::GrowState, Refusal> {
self.paid()?.start(check, nonce, crate::twin::now_ms())
}
pub async fn grow_settle(&self, past_waiting: bool) -> gunnar_repair::exec::GrowState {
match self.paid() {
Ok(g) => g.settle(past_waiting).await,
Err(r) => gunnar_repair::exec::GrowState::Failed(r.to_string()),
}
}
pub fn grow_look_again(&self, give_up: bool) -> Result<(), Refusal> {
self.paid()?.look_again(give_up)
}
#[cfg(feature = "metered")]
pub fn with_slot(
mut self,
slot: Option<Arc<crate::entitlement::EntitlementSlot>>,
) -> Entitlements {
self.slot = slot;
self
}
/// **The snapshot** (`monetize_embed::Snapshot` as JSON) from the ONE monetize:
/// only on a box that serves; verified whole (envelope and every fact) under the
/// box's monetize key, else refused whole; applied newer-wins; nothing deleted.
/// An empty snapshot is normal and asks nothing, so even a box without
/// entitlements answers it ok.
pub fn restate(&self, snapshot: &serde_json::Value) -> Result<String, Refusal> {
let bad = |message: String| Refusal {
code: RefusalCode::BadOrder,
message,
};
let Some(stores) = &self.stores else {
return Err(bad(
"this box holds no entitlements, so it cannot run restate_entitlements".into(),
));
};
let serving = health(stores, "")
.remove("serving")
.unwrap_or_else(|| "no (a twin)".into());
if serving != "yes" {
return Err(Refusal { code: RefusalCode::NotServing, message: format!("this box does not serve as the primary (serving={serving}): no entitlements restated") });
}
let Some(facts) = snapshot
.get("facts")
.and_then(|f| f.as_array())
.map(Vec::len)
else {
return Err(bad(
"the snapshot carries no list of facts: not a monetize snapshot".into(),
));
};
#[cfg(feature = "metered")]
if let Some(slot) = &self.slot {
let snap: monetize_embed::signing::Snapshot = serde_json::from_value(snapshot.clone())
.map_err(|e| bad(format!("the snapshot does not decode: {e}")))?;
let done = slot.restate(&snap).map_err(bad)?;
return Ok(restated(done.applied, done.held));
}
if facts == 0 {
return Ok(restated(0, 0));
}
Err(bad(format!("this box enforces no entitlements (not a metered build, or no monetize key): the snapshot's {facts} entitlement facts are not restated")))
}
}
impl Entitlements {
/// **Order `entitlement`**: one signed fact (`monetize_embed::EntitlementFact` as
/// JSON) from the ONE monetize — what `Entitlement.Set` took. Only on a box that
/// serves; verified under the box's monetize key (a forgery moves nothing); newer
/// wins, and a fact this box has moved past is acked as held, not refused, so
/// monetize stops asking.
pub fn entitle(&self, fact: &serde_json::Value) -> Result<String, Refusal> {
let bad = |message: String| Refusal {
code: RefusalCode::BadOrder,
message,
};
let Some(stores) = &self.stores else {
return Err(bad(
"this box holds no entitlements, so it cannot run an entitlement order".into(),
));
};
let serving = health(stores, "")
.remove("serving")
.unwrap_or_else(|| "no (a twin)".into());
if serving != "yes" {
return Err(Refusal { code: RefusalCode::NotServing, message: format!("this box does not serve as the primary (serving={serving}): the entitlement is not applied") });
}
#[cfg(feature = "metered")]
if let Some(slot) = &self.slot {
use crate::entitlement::SetRefused;
let fact: monetize_embed::EntitlementFact = serde_json::from_value(fact.clone())
.map_err(|e| bad(format!("the body is not a signed entitlement fact: {e}")))?;
let tenant = fact.tenant.0.clone();
return match slot.push(&fact) {
Ok(_) => Ok(format!("{tenant} is {:?}", fact.state)),
Err(SetRefused::Replayed { .. } | SetRefused::Behind { .. }) => Ok(format!(
"{tenant}: this box already holds a fact as new or newer; nothing moved"
)),
Err(e) => Err(bad(e.to_string())),
};
}
let tenant = fact.get("tenant").and_then(|t| t.as_str()).unwrap_or("?");
Err(bad(format!("this box enforces no entitlements (not a metered build, or no slot): the fact for {tenant} is not applied")))
}
}
impl Entitlements {
/// **Order `add_principal`** (ONE monetize, G1 pull): a paid signup's account — the
/// principal and its namespace, as `Access.AddPrincipal` makes them — on a main that
/// serves. The ack is monetize-product's contract: `Ok(namespace)`, also on a repeat for
/// the same `signup`; [`RefusalCode::Taken`] when the name is somebody else's (a
/// principal, a namespace, or another signup's); anything else a final refusal in words.
/// Only a `person` is made (G-D36). Which signup made which name is kept in the control
/// store ([`SIGNUPS_REF`]), so the twin follows it and a recharged main still knows.
pub fn add_principal(
&self,
username: &str,
kind: &str,
signup: &str,
) -> Result<String, Refusal> {
let bad = |message: String| Refusal {
code: RefusalCode::BadOrder,
message,
};
let (Some(stores), Some(control)) = (&self.stores, &self.control) else {
return Err(bad(format!("this box makes no accounts (no control plane in reach), so it cannot run add_principal for {username}")));
};
let serving = health(stores, "")
.remove("serving")
.unwrap_or_else(|| "no (a twin)".into());
if serving != "yes" {
return Err(Refusal { code: RefusalCode::NotServing, message: format!("this box does not serve as the primary (serving={serving}): no account made for {username}") });
}
if kind != "person" {
return Err(bad(format!("{username} is asked as {kind:?}: a signup makes a person's account only; an org is a clan its admin makes")));
}
if gunnar_grpc::service::access::is_reserved_principal_name(username) {
return Err(bad(format!("{username:?} is a reserved name (`console-*`, `kit-*`, `gunnar-ui`, `uitest-console`): no person may carry it")));
}
let store = stores.control_store().as_ref();
let exists = |name: &str| -> Result<(bool, bool), Refusal> {
let principal = stores
.auth()
.principal(name)
.map_err(|e| bad(format!("the principals cannot be read: {e}")))?
.is_some();
let namespace = stores
.catalog()
.namespace(name)
.map_err(|e| bad(format!("the namespaces cannot be read: {e}")))?
.is_some();
Ok((principal, namespace))
};
let made = signups::read(store).map_err(bad)?;
match made.get(signup) {
Some(name) if name == username => {
// A repeat: monetize lost the ack. Made, or the record outran the account.
if !exists(username)?.0 {
create(control.as_ref(), username, signup)
.map_err(|e| e.into_refusal(username))?;
}
return Ok(username.to_owned());
}
Some(name) if exists(name)?.0 => {
return Err(bad(format!("signup {signup} already made the account {name}; it makes no second one ({username})")));
}
_ => {}
}
let claimed = made.iter().any(|(s, name)| name == username && s != signup);
let (principal, namespace) = exists(username)?;
if claimed || principal || namespace {
return Err(taken(username));
}
if let Some(plane) = control.enterprise() {
let verdict = plane.name_check("account", username, "").map_err(|e| {
bad(format!(
"the name filter did not answer for {username}: {e}"
))
})?;
if !verdict.is_clean() {
return Err(bad(format!(
"the name filter refuses {username:?} ({}): {}",
verdict.verdict, verdict.sentence
)));
}
}
// The record first: a crash between the two leaves a record a repeat completes,
// never an account no signup owns.
signups::put(store, signup, Some(username)).map_err(bad)?;
if let Err(e) = create(control.as_ref(), username, signup) {
if let Err(why) = signups::put(store, signup, None) {
tracing::warn!(target: TARGET, "add_principal {username}: the signup record {signup} could not be taken back: {why}");
}
return Err(e.into_refusal(username));
}
Ok(username.to_owned())
}
}
/// The control document of which signup made which account: `{signup: username}`.
pub const SIGNUPS_REF: &str = "refs/gunnar/signups";
impl Entitlements {
/// **Order `sign_in_again`** (S10, owner 2026-10-11 05:53Z): a sold estate's buyer asked
/// the shop for a NEW sign-in link; the shop queued this signed order and holds no secret.
/// On a main that serves, with a mail relay: `account`'s principal, a person holding NO
/// passkey yet (a link enrols the first one only), gets a fresh invite minted HERE — the
/// `Access.CreateInvite` shape, its hash stored, living [`gunnar_repair::sign_in::ttl`] —
/// and the link (the first link's shape) is mailed to `to` by this box. The ack says who
/// was mailed, never the secret. A mail that does not go withdraws the invite.
/// ★ UW-5 **Order `add_users`** (ONE monetize, G1 pull): "Add users", granted on the
/// account `tenant` as the SELLER — the `Accounts.EnrollEmployee` rules
/// ([`gunnar_grpc::service::accounts::users_asked`]), under `order` (a retry answers the
/// first grant `unchanged`), no room check (monetize sold exactly the shortfall before the
/// money moved). A new named user's invite is minted here and its secret rides the answer
/// once (monetize mails it). The ack is monetize-product's `UsersAnswer` as JSON.
pub fn add_users(
&self,
tenant: &str,
ask: &serde_json::Value,
order: &str,
) -> Result<String, Refusal> {
let bad = |message: String| Refusal {
code: RefusalCode::BadOrder,
message,
};
let (Some(stores), Some(control)) = (&self.stores, &self.control) else {
return Err(bad(format!("this box adds no users (no control plane in reach), so it cannot run add_users for {tenant}")));
};
let serving = health(stores, "")
.remove("serving")
.unwrap_or_else(|| "no (a twin)".into());
if serving != "yes" {
return Err(Refusal { code: RefusalCode::NotServing, message: format!("this box does not serve as the primary (serving={serving}): no users added to {tenant}") });
}
let count = ask
.get("count")
.and_then(|v| v.as_u64())
.unwrap_or(1)
.clamp(1, u64::from(u32::MAX)) as u32;
let name = ask
.get("name")
.and_then(|v| v.as_str())
.unwrap_or_default()
.trim()
.to_owned();
let named = count == 1;
let request = gunnar_proto::EnrollEmployeeRequest {
account: tenant.to_owned(),
id: if named { name.clone() } else { String::new() },
name: if named { name.clone() } else { String::new() },
public_key: String::new(),
room_bytes: ask.get("room_bytes").and_then(|v| v.as_u64()).unwrap_or(0),
count,
from_seat: false,
dry_run: false,
order: order.to_owned(),
invite: named,
};
let employee = gunnar_grpc::service::accounts::users_asked(&request)
.map_err(|s| bad(format!("add_users for {tenant}: {}", s.message())))?;
for text in [employee.id.as_str(), employee.name.as_str()] {
if !text.is_empty() && gunnar_grpc::service::access::is_reserved_principal_name(text) {
return Err(bad(format!("{text:?} is a reserved name (`console-*`, `kit-*`, `gunnar-ui`, `uitest-console`): no user may carry it")));
}
}
let answer = control
.enroll_employee(tenant, &employee)
.map_err(|e| bad(format!("add_users for {tenant}: {e}")))?;
let mut invite_secret = String::new();
if let (Some(principal), true, false) = (&answer.principal, named, answer.unchanged) {
// The identity core's first-login link (IDENTITY-HOST): seven days.
let link = stores
.identity()
.ok_or_else(|| "this box's identity core is not up yet".to_owned())
.and_then(|identity| {
identity
.enrolment(
&principal.id,
"monetize:add_users",
Some(gunnar_grpc::service::access::INVITE_DEFAULT_MS / 1000),
)
.map_err(|refused| refused.to_string())
});
match link {
Ok(made) => invite_secret = made.secret,
Err(e) => {
tracing::warn!(target: TARGET, "add_users {tenant}: {} was added and its invite was refused: {e}", principal.id)
}
}
}
if !answer.unchanged {
tracing::info!(
target: gunnar_grpc::auth::AUDIT_TARGET,
event = if named { "account.employee_enrolled" } else { "account.seats_added" },
account = %tenant,
employee = %answer.principal.as_ref().map(|p| p.id.as_str()).unwrap_or(""),
seats = employee.seats,
room_bytes = answer.room_bytes,
order = %order,
"an \"Add users\" order was delivered (add_users)"
);
}
let out = serde_json::json!({
"principal": answer.principal.as_ref().map(|p| p.id.clone()).unwrap_or_default(),
"room_bytes": answer.room_bytes,
"need_bytes": answer.need_bytes,
"unused_bytes": answer.unused_bytes,
"shortfall_bytes": answer.shortfall_bytes,
"seats": answer.seats,
"seats_open": answer.seats_open,
"unchanged": answer.unchanged,
"invite_secret": invite_secret,
});
Ok(out.to_string())
}
/// ★ UW-4 **Order `grow_room`**: room was sold past this set (owner: the customer never
/// sees a cap — queued, resized later, the operator told). Kept as an `Admin.Health` alarm
/// (`grow.room_wanted`) naming what is wanted and why, for this estate's operator; the
/// set is resized by its growth. Acked with what was kept.
pub fn grow_room(&self, need_bytes: u64, reason: &str) -> Result<String, Refusal> {
let gib = need_bytes as f64 / (1u64 << 30) as f64;
let words = format!("{gib:.1} GiB of stored room was committed past this set ({reason}) — queued: grow the data set");
crate::feed::raise(ROOM_WANTED, &words);
tracing::warn!(target: TARGET, event = ROOM_WANTED, need_bytes, "{words}");
Ok(format!(
"kept: {gib:.1} GiB wanted past the set; the operator is told ({ROOM_WANTED})"
))
}
/// ★ UW-11 **Order `purge_tenant`** (ONE monetize, G1 pull): the account `tenant` lapsed
/// past retention and is deleted — `Accounts.Purge`'s own job on this main's control
/// plane (lock first, then unlink; a machine's account is never purged). No `job`: the
/// purge of the account already running is adopted, else one is started. A `job`: that
/// job's state, refused when this box no longer knows it (monetize then starts again,
/// never assumes the data went). The ack is monetize-product's `PurgeAnswer` as JSON.
pub fn purge_tenant(&self, tenant: &str, reason: &str, job: &str) -> Result<String, Refusal> {
use gunnar_grpc::seam::{PurgeAccountOptions, PurgeState};
let bad = |message: String| Refusal {
code: RefusalCode::BadOrder,
message,
};
let (Some(stores), Some(control)) = (&self.stores, &self.control) else {
return Err(bad(format!("this box purges no accounts (no control plane in reach), so it cannot run purge_tenant for {tenant}")));
};
let serving = health(stores, "")
.remove("serving")
.unwrap_or_else(|| "no (a twin)".into());
if serving != "yes" {
return Err(Refusal { code: RefusalCode::NotServing, message: format!("this box does not serve as the primary (serving={serving}): {tenant} is not purged here") });
}
let latest = control
.purge_account_status(tenant)
.map_err(|e| bad(format!("purge_tenant {tenant}: {e}")))?
.0;
let running = |s: PurgeState| matches!(s, PurgeState::Queued | PurgeState::Running);
let found = match (job.trim(), latest) {
("", Some(j)) if running(j.state) => j,
("", _) => {
let options = PurgeAccountOptions { principal_id: "monetize:nonpay".into(), reason: reason.to_owned(), remove_principal: false, expect_empty: false };
let (j, _) = control.purge_account(tenant, options).map_err(|e| bad(format!("purge_tenant {tenant}: {e}")))?;
tracing::info!(target: gunnar_grpc::auth::AUDIT_TARGET, event = "account.purge_ordered", account = %tenant, job = %j.job_id, reason = %reason, "a nonpay purge was ordered by monetize (purge_tenant)");
j
}
(asked, Some(j)) if j.job_id == asked => j,
(asked, _) => return Err(bad(format!("{tenant}: this box no longer reports purge job {asked}; nothing can say whether the data went"))),
};
let gib = found.bytes_reclaimed as f64 / (1u64 << 30) as f64;
let out = serde_json::json!({
"job": found.job_id,
"running": running(found.state),
"failed": matches!(found.state, PurgeState::Failed),
"error": found.error.clone().unwrap_or_default(),
"detail": format!("{} of {} stores, {gib:.1} GiB", found.stores_purged, found.stores_found),
"bytes_reclaimed": found.bytes_reclaimed,
"still_locked": !found.gate_released,
"failures": found.failures,
});
Ok(out.to_string())
}
/// **`mail_relay`** (MAIL-RELAY 2026-10-28): the relay the ONE box fed this main again,
/// kept sealed under the data root ([`crate::mail_relay::keep`]); the postman reads it over
/// the lay's feed from the next mail.
#[cfg(feature = "fleet-pull")]
pub fn keep_mail_relay(
&self,
relay: &gunnar_repair::order::MailRelay,
) -> Result<String, Refusal> {
let bad = |message: String| Refusal {
code: RefusalCode::BadOrder,
message,
};
let Some(stores) = &self.stores else {
return Err(bad(format!(
"this box has no data root in reach, so it keeps no mail relay (as {})",
relay.smtp_from
)));
};
crate::mail_relay::keep(stores.root(), relay).map_err(bad)
}
/// **A console setting the ONE box fed again** (a `refeed`, gunnar-repair's
/// `keep_console_setting`): `value` stored under the generic setting `name`, as an
/// administrator's `Admin.SetSetting` stores it. This server gives the name no meaning (it
/// must stay generic: `tests/server_settings.rs`); it takes only a `console.*` name, the
/// console's own namespace, so a refeed cannot write a setting of anything else. Only a
/// serving main stores it; a twin answers that its main keeps it (the settings are control
/// state, replicated to it). Audited on `gunnar::audit` (who: the ONE box's order).
pub fn keep_console_setting(&self, name: &str, value: &str) -> Result<String, Refusal> {
let bad = |message: String| Refusal {
code: RefusalCode::BadOrder,
message,
};
if !name.starts_with("console.") {
return Err(bad(format!(
"a refeed keeps only a console setting (console.*), not {name}"
)));
}
let Some(stores) = &self.stores else {
return Err(bad(format!("this box has no data root in reach, so it keeps no console setting ({name} = {value})")));
};
let serving = health(stores, "")
.remove("serving")
.unwrap_or_else(|| "no (a twin)".into());
if serving != "yes" {
return Ok(format!("this box does not serve (serving={serving}): its main keeps {name}, and the setting replicates here"));
}
let previous = stores
.settings()
.set(name, value.as_bytes())
.map_err(|e| bad(format!("the console setting {name} was not stored: {e}")))?;
let was = previous
.as_deref()
.map(|b| String::from_utf8_lossy(b).into_owned())
.unwrap_or_else(|| "(unset)".to_owned());
tracing::warn!(
target: gunnar_grpc::AUDIT_TARGET,
event = "admin.setting_set",
principal = "monetize (refeed order)",
name = name,
old = %was,
new = value,
"a server setting was written: the ONE box fed a console setting again"
);
Ok(format!(
"{name}: {was} -> {value}; the console follows it within its recheck"
))
}
#[cfg(feature = "fleet-pull")]
pub async fn sign_in_again(
&self,
account: &str,
to: &str,
host: &str,
ttl_secs: u64,
) -> Result<String, Refusal> {
let bad = |message: String| Refusal {
code: RefusalCode::BadOrder,
message,
};
let (Some(stores), Some(_)) = (&self.stores, &self.control) else {
return Err(bad(format!("this box mints no sign-in links (no control plane in reach), so it cannot run sign_in_again for {account}")));
};
let serving = health(stores, "")
.remove("serving")
.unwrap_or_else(|| "no (a twin)".into());
if serving != "yes" {
return Err(Refusal { code: RefusalCode::NotServing, message: format!("this box does not serve as the primary (serving={serving}): no sign-in link minted for {account}") });
}
let Some(postman) = &self.postman else {
return Err(bad(format!(
"this box has no mail relay ({} and a token), so it mints no sign-in link for {account} it could not mail",
crate::postman::FROM_ENV
)));
};
postman.ready().map_err(|why| bad(format!("this box cannot mail ({why}), so it mints no sign-in link for {account} it could not mail")))?;
gunnar_repair::sign_in::check(account, to, host)
.map_err(|why| bad(format!("sign_in_again: {why}")))?;
let principal = stores
.auth()
.principal_by_name(account)
.map_err(|e| bad(format!("the principals cannot be read: {e}")))?
.ok_or_else(|| {
bad(format!(
"{account} is no account on this box: no sign-in link minted"
))
})?;
if principal.kind != gunnar_core::PrincipalKind::User {
return Err(bad(format!(
"{account} is not a person: a sign-in link enrols a person's passkey"
)));
}
let passkeys = stores
.auth()
.web_credentials(Some(&principal.id))
.map_err(|e| bad(format!("the passkeys cannot be read: {e}")))?;
if !passkeys.is_empty() {
return Err(bad(format!(
"{account} already signs in with a passkey ({} held): a sign-in link only enrols the first one — sign in, or add a passkey from a signed-in device",
passkeys.len()
)));
}
let ttl = gunnar_repair::sign_in::ttl(ttl_secs);
// The identity core's first-login link (IDENTITY-HOST): minted here,
// kept only as its hash, withdrawn when the mail does not go.
let identity = stores.identity().ok_or_else(|| {
bad(format!(
"this box's identity core is not up yet: no sign-in link for {account}; ask again"
))
})?;
let invite = identity
.enrolment(
&principal.id,
"monetize:sign_in_again",
Some(i64::try_from(ttl).unwrap_or(i64::MAX)),
)
.map_err(|e| bad(format!("the invite for {account} could not be made: {e}")))?;
let secret = invite.secret.clone();
let mail = gunnar_repair::sign_in::mail(host, account, to, &secret, ttl);
drop(secret);
if let Err(why) = postman.post(&mail).await {
if let Err(e) = identity.withdraw_invite(&invite.invite_id, "monetize:sign_in_again") {
tracing::warn!(target: TARGET, "sign_in_again {account}: the unmailed invite {} could not be withdrawn: {e}", invite.invite_id);
}
return Err(bad(format!("the sign-in link for {account} could not be mailed to {to} ({why}); the invite is withdrawn — ask again")));
}
tracing::info!(target: TARGET, account, invite = %invite.invite_id, ttl_secs = ttl, "sign_in_again: a new sign-in link was minted here and mailed to the order's address");
Ok(format!(
"a new sign-in link for {account} (invite {}, {}) was mailed to {to}",
invite.invite_id,
gunnar_repair::sign_in::ttl_words(ttl)
))
}
}
fn taken(username: &str) -> Refusal {
Refusal { code: RefusalCode::Taken, message: format!("{username} is somebody else's (a principal, a namespace or another signup holds it): choose another name") }
}
enum NotMade {
Exists,
Other(String),
}
impl NotMade {
fn into_refusal(self, username: &str) -> Refusal {
match self {
NotMade::Exists => taken(username),
NotMade::Other(why) => Refusal {
code: RefusalCode::BadOrder,
message: format!("the account {username} was not made: {why}"),
},
}
}
}
/// The principal and its namespace, in one claim (`ControlPlane::add_principal`), and the
/// account's audit line.
fn create(
control: &dyn gunnar_grpc::ControlPlane,
username: &str,
signup: &str,
) -> Result<(), NotMade> {
use gunnar_grpc::ControlError;
match control.add_principal(username, username, gunnar_proto::PrincipalKind::User) {
Ok(_) => {}
Err(ControlError::AlreadyExists(_)) => return Err(NotMade::Exists),
Err(e) => return Err(NotMade::Other(e.to_string())),
}
tracing::info!(target: TARGET, username, signup, "add_principal: a signup's account was made");
if let Some(plane) = control.enterprise() {
if let Ok(Some(account)) = plane.account_of_principal(username) {
let event = gunnar_grpc::enterprise::AccountAuditEvent {
kind: "account.created".into(),
actor: "monetize".into(),
actor_machine: true,
target: username.to_owned(),
detail: format!("a signup at the shop ({signup}), order add_principal"),
refused: false,
};
if let Err(e) = plane.account_audit_record(&account, &event) {
tracing::warn!(target: TARGET, "add_principal {username}: the account's audit line was not written: {e}");
}
}
}
Ok(())
}
mod signups {
use std::collections::BTreeMap;
use gunnar_core::ControlStore;
use super::SIGNUPS_REF;
pub(super) fn read(store: &dyn ControlStore) -> Result<BTreeMap<String, String>, String> {
Ok(load(store)?.1)
}
fn load(
store: &dyn ControlStore,
) -> Result<(Option<gunnar_core::BlobId>, BTreeMap<String, String>), String> {
let at = store
.read_doc(SIGNUPS_REF)
.map_err(|e| format!("{SIGNUPS_REF} cannot be read: {e}"))?;
let Some(id) = at else {
return Ok((None, BTreeMap::new()));
};
let bytes = store
.get_blob(&id)
.map_err(|e| format!("{SIGNUPS_REF} cannot be read: {e}"))?
.ok_or_else(|| format!("{SIGNUPS_REF} points at {id}, which is not stored"))?;
let map = serde_json::from_slice(&bytes)
.map_err(|e| format!("{SIGNUPS_REF} does not decode: {e}"))?;
Ok((Some(id), map))
}
/// Record (`Some`) or take back (`None`) what `signup` made; compare-and-swap, re-read on a lost race.
pub(super) fn put(
store: &dyn ControlStore,
signup: &str,
username: Option<&str>,
) -> Result<(), String> {
for _ in 0..16 {
let (at, mut map) = load(store)?;
match username {
Some(u) => map.insert(signup.to_owned(), u.to_owned()),
None => map.remove(signup),
};
let bytes = serde_json::to_vec(&map).map_err(|e| e.to_string())?;
let id = store
.put_blob(&bytes)
.map_err(|e| format!("{SIGNUPS_REF} cannot be written: {e}"))?;
if store
.commit_doc(SIGNUPS_REF, at, id)
.map_err(|e| format!("{SIGNUPS_REF} cannot be written: {e}"))?
{
return Ok(());
}
}
Err(format!("{SIGNUPS_REF}: 16 writers in a row moved it first"))
}
}
fn restated(applied: usize, held: usize) -> String {
format!("restated {applied} entitlement facts ({held} already held as new or newer)")
}
impl ServerEstate {
/// The repair key named by `cfg`, else the FED one ([`crate::feed::FEED_REPAIR_KEY`]),
/// else `<data>/twin.key`.
pub fn new(
root: &std::path::Path,
cfg: &RepairConfig,
own: String,
) -> Result<ServerEstate, String> {
Self::with_fed(root, cfg, own, crate::feed::Fed::current().as_ref())
}
/// [`ServerEstate::new`] over the feed `fed` (the one in force, or a test's).
pub fn with_fed(
root: &std::path::Path,
cfg: &RepairConfig,
own: String,
fed: Option<&crate::feed::Fed>,
) -> Result<ServerEstate, String> {
let key = match (&cfg.key, fed.and_then(|f| f.repair_key.as_ref())) {
(Some(path), _) => {
let bytes = std::fs::read(path)
.map_err(|e| format!("the repair key {}: {e}", path.display()))?;
PrivateKey::from_openssh(&bytes).map_err(|e| {
format!(
"the repair key {} is not an OpenSSH private key: {e}",
path.display()
)
})?
}
(None, Some(k)) => PrivateKey::from_openssh(k.0.trim().as_bytes()).map_err(|e| {
format!(
"the fed {} is not an OpenSSH private key: {e}",
crate::feed::FEED_REPAIR_KEY
)
})?,
(None, None) => crate::twin::twin_private_key(root).map_err(|e| {
format!(
"no --repair-key, no fed {}, and no twin key on this box ({e})",
crate::feed::FEED_REPAIR_KEY
)
})?,
};
Ok(ServerEstate {
key,
own,
entitlements: Entitlements::default(),
})
}
}
/// **Is auto-fix on**: the medium's `--repair-auto`, or the estate's fed opt-in
/// ([`crate::feed::FEED_REPAIR_AUTO`]) in force now.
pub fn auto_on(cfg: &RepairConfig) -> bool {
auto_with(cfg, crate::feed::Fed::current().as_ref())
}
/// Pure: [`auto_on`] over the feed `fed`.
pub fn auto_with(cfg: &RepairConfig, fed: Option<&crate::feed::Fed>) -> bool {
cfg.auto || fed.is_some_and(|f| f.repair_auto)
}
/// What the serve armed the awaiting-main kick with ([`arm`]).
struct Armed {
cfg: RepairConfig,
own: String,
rt: tokio::runtime::Handle,
}
fn armed() -> &'static std::sync::OnceLock<Armed> {
static ARMED: std::sync::OnceLock<Armed> = std::sync::OnceLock::new();
&ARMED
}
/// When the last awaiting-main kick started, and whether one is running.
fn kicked() -> &'static std::sync::Mutex<(Option<std::time::Instant>, bool)> {
static KICKED: std::sync::Mutex<(Option<std::time::Instant>, bool)> =
std::sync::Mutex::new((None, false));
&KICKED
}
/// After a kick that did not finish the recharge, the next is no sooner than this.
pub const KICK_AGAIN_AFTER: Duration = Duration::from_secs(60);
/// **Arm the awaiting-main kick** (main rebuild, 2026-10-12): the serve hands the repair
/// configuration, this box's own address and its runtime. Once per process.
pub fn arm(cfg: RepairConfig, own: String, rt: tokio::runtime::Handle) {
let _ = armed().set(Armed { cfg, own, rt });
}
/// Pure: **may an awaiting main kick its own recharge now?** Auto-fix on, none running,
/// and the last one at least [`KICK_AGAIN_AFTER`] ago.
pub fn kick_due(auto: bool, running: bool, last: Option<Duration>) -> bool {
auto && !running && last.is_none_or(|ago| ago >= KICK_AGAIN_AFTER)
}
/// **A re-laid main awaits its twin** (pairing's `Hold::Awaiting`): with auto-fix on it
/// fills itself — the boot check's recharge, run now that the door has named the twin
/// (the boot check ran at start, before the feed did). `None`: nothing was started;
/// `Some(words)`: what was. The guard is the boot check's own: a fresh box with a twin
/// that can give it back (`gunnar_repair::rebuild::judge` → `Fill`); nothing is destroyed.
/// A kick that cannot start (no repair key) says why once and is due again at the next turn.
pub fn on_awaiting(stores: &Arc<ServerStores>, twin: &str) -> Option<String> {
let a = armed().get()?;
// A box-to-box call never leaves a box in the clear: with no client certificate for
// another box's control plane, this box cannot dial its twin — the FRONT fills it then
// (gunnar-front `autofill`, the console's key over the tunnel). Said once, not a WARN.
if !may_dial(twin) {
static SAID: std::sync::atomic::AtomicBool = std::sync::atomic::AtomicBool::new(false);
if auto_on(&a.cfg) && !SAID.swap(true, std::sync::atomic::Ordering::SeqCst) {
tracing::info!(target: TARGET, twin, "auto-fix: this box holds no client certificate to dial its twin; the front fills it from the twin");
}
return None;
}
{
let mut k = kicked().lock().unwrap_or_else(|p| p.into_inner());
if !kick_due(auto_on(&a.cfg), k.1, k.0.map(|t| t.elapsed())) {
return None;
}
*k = (Some(std::time::Instant::now()), true);
}
let estate = match ServerEstate::new(stores.root(), &a.cfg, a.own.clone()) {
Ok(e) => e,
Err(why) => {
// Nothing started: the next turn may try again at once (a feed may bring the
// key), and the same reason is said once, not on every turn.
*kicked().lock().unwrap_or_else(|p| p.into_inner()) = (None, false);
static SAID: std::sync::Mutex<String> = std::sync::Mutex::new(String::new());
let mut said = SAID.lock().unwrap_or_else(|p| p.into_inner());
if *said == why {
return None;
}
said.clone_from(&why);
tracing::warn!(target: TARGET, twin, "auto-fix: this re-laid main awaits its twin and cannot fill itself ({why}); an admin or a guardian order can");
return Some(format!("auto-fix cannot run: {why}"));
}
};
let (stores, twin_at) = (Arc::clone(stores), twin.to_owned());
// A32-14, stays a task: one boot check of the auto-fix, async on the server's runtime; its report is the record.
a.rt.spawn(async move {
let dead = data_dead(&stores);
let report = exec::boot_check(&estate, Some(&twin_at), dead, true, &|| false).await;
match &report.recharge {
Some(Ok(r)) => tracing::info!(target: TARGET, twin = %twin_at, head_seq = r.head_seq, "auto-fix: the re-laid main filled itself from its twin: {:?} — {}", r.phase, r.detail),
Some(Err(why)) => tracing::error!(target: TARGET, twin = %twin_at, "auto-fix: the recharge from the twin was refused — {why}"),
None => tracing::warn!(target: TARGET, twin = %twin_at, finding = ?report.finding, "auto-fix: the boot check found nothing it may recharge"),
}
kicked().lock().unwrap_or_else(|p| p.into_inner()).1 = false;
});
Some(format!(
"auto-fix: filling this re-laid main from its twin at {twin}"
))
}
/// Can this box dial `at` for a repair call: on this box, or with a client certificate.
pub fn may_dial(at: &str) -> bool {
gunnar_grpc::client_signer::is_loopback_target(at)
|| matches!(
gunnar_grpc::client_signer::ClientTrust::from_env(),
Ok(Some(_))
)
}
/// A control-plane channel to `at`: mutual TLS when this box holds a client
/// certificate, else plain (which `plain_channel` allows only on this machine).
pub fn dial(at: &str) -> Result<ControlChannel, Fault> {
use gunnar_grpc::client_signer as cs;
let trust = cs::ClientTrust::from_env().map_err(|e| Fault::new("unauthenticated", e))?;
match trust {
Some(trust) => cs::mutual_channel(at, &trust, &cs::tls_domain_for(at)),
None => cs::plain_channel(at),
}
.map_err(|e| Fault::new("unavailable", format!("{at}: {e}")))
}
impl Estate for ServerEstate {
type Twin = ProtoTwin;
type Main = ProtoMain;
type Fence = ProtoFence;
fn twin(&self, addr: &str) -> Result<ProtoTwin, Fault> {
Ok(ProtoTwin {
addr: addr.to_owned(),
channel: dial(addr)?,
key: self.key.clone(),
})
}
fn main(&self, addr: &str) -> Result<ProtoMain, Fault> {
Ok(ProtoMain {
addr: addr.to_owned(),
channel: dial(addr)?,
key: self.key.clone(),
})
}
fn fence(&self, addr: &str) -> Result<ProtoFence, Fault> {
Ok(ProtoFence {
addr: addr.to_owned(),
channel: dial(addr)?,
key: self.key.clone(),
})
}
fn restate(&self, snapshot: &serde_json::Value) -> Result<String, Refusal> {
self.entitlements.restate(snapshot)
}
fn entitle(&self, fact: &serde_json::Value) -> Result<String, Refusal> {
self.entitlements.entitle(fact)
}
fn add_principal(&self, username: &str, kind: &str, signup: &str) -> Result<String, Refusal> {
self.entitlements.add_principal(username, kind, signup)
}
fn add_users(
&self,
tenant: &str,
ask: &serde_json::Value,
order: &str,
) -> Result<String, Refusal> {
self.entitlements.add_users(tenant, ask, order)
}
fn grow_room(&self, need_bytes: u64, reason: &str) -> Result<String, Refusal> {
self.entitlements.grow_room(need_bytes, reason)
}
fn purge_tenant(&self, tenant: &str, reason: &str, job: &str) -> Result<String, Refusal> {
self.entitlements.purge_tenant(tenant, reason, job)
}
#[cfg(feature = "fleet-pull")]
async fn sign_in_again(
&self,
account: &str,
to: &str,
host: &str,
ttl_secs: u64,
) -> Result<String, Refusal> {
self.entitlements
.sign_in_again(account, to, host, ttl_secs)
.await
}
#[cfg(feature = "fleet-pull")]
fn keep_mail_relay(&self, relay: &gunnar_repair::order::MailRelay) -> Result<String, Refusal> {
self.entitlements.keep_mail_relay(relay)
}
fn keep_console_setting(&self, name: &str, value: &str) -> Result<String, Refusal> {
self.entitlements.keep_console_setting(name, value)
}
fn grow_check(
&self,
go_ahead: &str,
data_gib: u64,
) -> Result<gunnar_repair::exec::GrowCheck, Refusal> {
self.entitlements.grow_check(go_ahead, data_gib)
}
async fn twin_holds(&self, check: &gunnar_repair::exec::GrowCheck) -> Result<String, Refusal> {
self.entitlements.twin_holds(check).await
}
fn grow_start(
&self,
check: &gunnar_repair::exec::GrowCheck,
nonce: &str,
) -> Result<gunnar_repair::exec::GrowState, Refusal> {
self.entitlements.grow_start(check, nonce)
}
async fn grow_settle(&self, past_waiting: bool) -> gunnar_repair::exec::GrowState {
self.entitlements.grow_settle(past_waiting).await
}
fn grow_look_again(&self, give_up: bool) -> Result<(), Refusal> {
self.entitlements.grow_look_again(give_up)
}
fn own(&self) -> String {
self.own.clone()
}
fn now_unix_ms(&self) -> i64 {
crate::twin::now_ms()
}
}
/// This box's control plane as a peer dials it: an unspecified bind is this machine.
pub fn own_addr(bound: std::net::SocketAddr) -> String {
if bound.ip().is_unspecified() {
std::net::SocketAddr::from(([127, 0, 0, 1], bound.port())).to_string()
} else {
bound.to_string()
}
}
#[cfg(feature = "wg-client")]
fn fresh(stores: &ServerStores) -> bool {
crate::pairing::is_fresh(stores).unwrap_or(false)
}
#[cfg(not(feature = "wg-client"))]
fn fresh(stores: &ServerStores) -> bool {
stores.auth().is_empty().unwrap_or(false)
&& stores
.catalog()
.list(None)
.map(|l| l.is_empty())
.unwrap_or(false)
}
#[cfg(feature = "wg-client")]
fn awaited(stores: &ServerStores) -> Option<String> {
crate::pairing::recharge_awaited(stores)
}
#[cfg(not(feature = "wg-client"))]
fn awaited(_: &ServerStores) -> Option<String> {
None
}
/// The twin the door named this main ([`crate::pairing::door_twin`]).
#[cfg(feature = "wg-client")]
fn door_twin(stores: &ServerStores) -> Option<String> {
crate::pairing::door_twin(stores.root())
}
#[cfg(not(feature = "wg-client"))]
fn door_twin(_: &ServerStores) -> Option<String> {
None
}
/// **Is this box's data set dead**: a restore it never finished, or a box with no
/// identity and no repository (a re-laid main, an emptied disk). Never on a twin.
pub fn data_dead(stores: &ServerStores) -> bool {
!crate::twin::is_twin_root(stores.root())
&& (crate::twin::is_restoring(stores.root()) || fresh(stores))
}
/// `<data>/repair/hold`: the boot check found this box's data set lost with a twin
/// able to give it back (R-7b). It names that twin; while it stands and the box is
/// still dead, `Bootstrap.Enroll` takes no first admin and `Bootstrap.Status`
/// answers the twin in `recharge_from`.
pub const HOLD_REL: &str = "repair/hold";
/// **Hold this box** for a recharge from `twin` (R-7b). The boot check's act on
/// `Finding::DataLost`, whether or not auto-fix then recharges.
pub fn hold(stores: &ServerStores, twin: &str) -> anyhow::Result<()> {
if let Some(dir) = stores.root().join(HOLD_REL).parent() {
std::fs::create_dir_all(dir)?;
}
crate::durable::write_atomic_unframed(
stores.root(),
HOLD_REL,
format!("{}\n", twin.trim()).as_bytes(),
)
}
/// **The twin this box is held for**, if it is held. The hold lifts by itself once
/// the box holds a data set again (the recharge finished: the marker goes); the
/// admin's way out is removing the marker by hand.
pub fn held(stores: &ServerStores) -> Option<String> {
let path = stores.root().join(HOLD_REL);
let twin = std::fs::read_to_string(&path).ok()?.trim().to_owned();
if !data_dead(stores) {
if std::fs::remove_file(&path).is_ok() {
tracing::info!(target: TARGET, twin = %twin, "the hold is lifted: this box holds its data set again");
}
return None;
}
(!twin.is_empty()).then_some(twin)
}
/// What `Bootstrap.Enroll` answers on a held box.
pub fn held_refusal(twin: &str) -> String {
format!(
"{}: this box is HELD for a recharge from the twin at {twin}. A first admin now would start a new, \
empty estate and shut the window the recharge needs. Recharge it (the Twin panel, `gunnar control twin \
replay`, or a guardian order); to start empty on purpose, remove <data>/{HOLD_REL} and enrol again",
gunnar_repair::DATA_LOST_LINE
)
}
/// The twin the boot check reviews: the configured one, else the one the installer's
/// re-admit named.
pub fn twin_of(stores: &ServerStores, cfg: &RepairConfig) -> Option<String> {
cfg.twin
.clone()
.filter(|t| !t.trim().is_empty())
.or_else(|| awaited(stores))
}
/// **The boot check** — once, after the control plane listens. A healthy box, a twin,
/// or a box with no twin to name says nothing.
pub async fn boot(stores: Arc<ServerStores>, cfg: RepairConfig, own: String) -> Option<BootReport> {
if crate::twin::is_twin_root(stores.root()) {
return None;
}
let dead = data_dead(&stores);
let twin = twin_of(&stores, &cfg);
if !dead || twin.is_none() {
return None;
}
let estate = match ServerEstate::new(stores.root(), &cfg, own) {
Ok(e) => e,
Err(why) => {
tracing::warn!(target: TARGET, twin = twin.as_deref().unwrap_or(""), "{}: this box cannot run it ({why}); an admin or a guardian order can", gunnar_repair::DATA_LOST_LINE);
return None;
}
};
let auto = auto_on(&cfg);
let report = exec::boot_check(&estate, twin.as_deref(), dead, auto, &|| false).await;
if let gunnar_repair::Finding::DataLost { twin_addr, .. } = &report.finding {
match hold(&stores, twin_addr) {
Ok(()) => {
tracing::warn!(target: TARGET, twin = %twin_addr, "held: Bootstrap.Enroll takes no first admin until this box is recharged (or <data>/{HOLD_REL} is removed)")
}
Err(why) => {
tracing::error!(target: TARGET, twin = %twin_addr, "the hold could not be written ({why}); the enrolment window stays open")
}
}
}
match &report.step {
BootStep::Nothing => {}
BootStep::Say { line } => tracing::warn!(target: TARGET, auto_fix = auto, "{line}"),
BootStep::Recharge { line, .. } => match &report.recharge {
Some(Ok(r)) => {
tracing::info!(target: TARGET, head_seq = r.head_seq, "{line}: {:?} — {}", r.phase, r.detail)
}
Some(Err(why)) => tracing::error!(target: TARGET, "{line}: refused — {why}"),
None => {}
},
}
Some(report)
}
/// **The words the guardian reads** off this box's report.
pub fn health(stores: &ServerStores, own: &str) -> BTreeMap<String, String> {
let root = stores.root();
let mut h = BTreeMap::new();
h.insert("serves_at".to_owned(), own.to_owned());
if crate::twin::is_twin_root(root) {
let (busy, total) = crate::twin::own_load();
twin_words(&mut h, crate::twin::status(root).as_ref(), busy, total);
return h;
}
let restoring = crate::twin::is_restoring(root);
let dead = data_dead(stores);
let awaited = awaited(stores).or_else(|| held(stores));
let data = match (dead, restoring, awaited.is_some()) {
(false, _, _) => "ok",
(true, true, _) | (true, false, true) => "lost",
(true, false, false) => "fresh",
};
h.insert("data".to_owned(), data.to_owned());
// ★ main rebuild (R4): the main's own log head, so a guard can tell a main AHEAD of its
// twin (entries only its disk holds) from one the twin can give back whole.
if !dead {
h.insert(
"data_head".to_owned(),
stores.txlog().head_seq().to_string(),
);
}
// A primary fenced before a recharge (standby, no roster) answers and serves reads,
// and takes no client write: the guardian reads `fenced`, never `yes`.
let fenced = !dead
&& crate::mirror::ReplicationConfig::load(root)
.map(|c| c.role.refuses_client_writes())
.unwrap_or(false);
h.insert(
"serving".to_owned(),
if dead {
"no"
} else if fenced {
"fenced"
} else {
"yes"
}
.to_owned(),
);
if restoring {
h.insert("recharge".to_owned(), "replaying".to_owned());
}
h
}
/// **A twin's own review** (G12), JSON text of [`gunnar_repair::TwinReview`]: what its
/// `Mirror.LogHead` says, reviewed by the box itself on every heartbeat. `None` on a main.
pub fn twin_review(stores: &ServerStores, own: &str, now_unix_ms: i64) -> Option<String> {
if !crate::twin::is_twin_root(stores.root()) {
return None;
}
use gunnar_repair::AsTwinHead;
let head = crate::twin::own_head(stores).twin_head();
serde_json::to_string(&gunnar_repair::review_twin(
own,
Ok(&head),
None,
None,
now_unix_ms,
))
.ok()
}
/// **Where this box pulls, flag or feed** (ONE monetize, M-D35): `--fleet` /
/// `--fleet-instance` on the medium, else the installer's feed (`fleet`, `fleet_instance`,
/// which the front offers every box). No precedence: both and different is refused by name
/// ([`crate::feed::FEED_MISMATCH`]), as for the monetize key. A fed door is https with no CA
/// file: its certificate is checked against the public roots, or the fed `fleet_ca`; it is
/// dialled at its name, or at the fed `fleet_at`.
pub fn resolve_fleet(
flag: Option<FleetConfig>,
fed: Option<&crate::feed::Fed>,
every: Duration,
) -> Result<Option<FleetConfig>, String> {
let named = resolve_fleet_named(flag, fed, every)?;
if !PULLS {
if let Some(f) = &named {
return Err(format!(
"{NO_FLEET_PULL} (named: {} for {})",
f.url, f.instance
));
}
}
Ok(named)
}
/// **AIRGAP B5** (owner rule 2026-10-25): whether this build carries the pull channel to the
/// vendor's monetize. Only estate and metered builds do (`fleet-pull`, named by every
/// gunnar-server row); gunnar-solo and gunnar-service do not.
pub const PULLS: bool = cfg!(feature = "fleet-pull");
/// The refusal a named or fed pull channel gets in a build without `fleet-pull`.
pub const NO_FLEET_PULL: &str =
"this gunnar reports to no fleet: the pull channel to monetize needs gunnar-server's `fleet-pull` feature (estate and metered builds only)";
/// [`resolve_fleet`]'s reading of the flags and the feed, whatever the build.
pub fn resolve_fleet_named(
flag: Option<FleetConfig>,
fed: Option<&crate::feed::Fed>,
every: Duration,
) -> Result<Option<FleetConfig>, String> {
let (at, ca_pem) = (
fed.and_then(|f| f.fleet_at.clone()),
fed.and_then(|f| f.fleet_ca.clone()),
);
if let Some(why) = seam_refusal(at.as_deref(), ca_pem.is_some()) {
return Err(why);
}
let fed = match fed.and_then(|f| f.fleet.clone()) {
None => return Ok(flag),
Some(url) => {
let instance = fed.and_then(|f| f.fleet_instance.clone()).ok_or_else(|| {
format!("the installer's door fed `{}` = {url} but no `{}`: the estate this box reports for", crate::feed::FEED_FLEET, crate::feed::FEED_FLEET_INSTANCE)
})?;
(url, instance)
}
};
match flag {
None => Ok(Some(FleetConfig { url: fed.0, ca: None, ca_pem, at, instance: fed.1, role: String::new(), monetize_pubkey: None, every })),
Some(f) if f.url == fed.0 && f.instance == fed.1 => Ok(Some(f)),
Some(f) => Err(format!(
"REFUSED [{}] the medium names --fleet {} --fleet-instance {} and the installer's door fed `{}` = {} `{}` = {}; \
two answers to where this box reports, and neither wins. Remove the flags or make them agree.",
crate::feed::FEED_MISMATCH,
f.url,
f.instance,
crate::feed::FEED_FLEET,
fed.0,
crate::feed::FEED_FLEET_INSTANCE,
fed.1
)),
}
}
pub use crate::feed::seam_refusal;
/// **The pull channel the feed names LIVE** (a first boot: the front feeds `fleet` after
/// this process started). With a flag, a different fed door is an alarm
/// ([`crate::feed::FEED_MISMATCH`]) and the flag's channel stays; with none, the first fed
/// door is returned. `None` only if the feed never names one.
pub async fn fleet_when_fed(flag: Option<FleetConfig>, every: Duration) -> Option<FleetConfig> {
if let Some(f) = flag {
let mine = f.clone();
nordisk_pairing::feedbus::on_published("fleet", move |feed| {
if let Err(why) = resolve_fleet(
Some(mine.clone()),
Some(&crate::feed::Fed::from_feed(feed)),
every,
) {
crate::feed::raise(crate::feed::FEED_MISMATCH, &why);
}
});
return Some(f);
}
let (tx, rx) = tokio::sync::oneshot::channel();
let tx = std::sync::Mutex::new(Some(tx));
nordisk_pairing::feedbus::on_published("fleet", move |feed| {
match resolve_fleet(None, Some(&crate::feed::Fed::from_feed(feed)), every) {
Ok(Some(f)) => {
if let Some(tx) = tx.lock().unwrap_or_else(|p| p.into_inner()).take() {
tracing::info!(target: TARGET, fleet = %f.url, instance = %f.instance, "the installer fed the pull channel; pulling from now on");
let _ = tx.send(f);
}
}
Ok(None) => {}
Err(why) => tracing::warn!(target: TARGET, "{why}"),
}
});
rx.await.ok()
}
/// **What a twin reports** (F4 x G12, the twin-health PULL — 5.4 G1-pull #25/#26): its
/// primary's word, and its lag and load, which the ONE monetize's twin autoscale reads off
/// the report (monetize `estate_order::twin::TwinReading::of_reports`, its `twin_state` /
/// `lag_entries` / `cpu_millicores_*` words) instead of dialing the estate's door. A twin that
/// has not polled yet reports no lag (no reading).
pub fn twin_words(
h: &mut BTreeMap<String, String>,
status: Option<&crate::twin::TwinStatus>,
cpu_busy: Option<u64>,
cpu_total: Option<u64>,
) {
let primary = match status.map(|s| s.state) {
Some(crate::twin::State::Unreachable) => "silent",
Some(crate::twin::State::Diverged) => "diverged",
Some(_) => "live",
None => "unknown",
};
h.insert("primary".to_owned(), primary.to_owned());
if let Some(s) = status {
h.insert(TWIN_STATE.to_owned(), s.state.to_string());
h.insert(
TWIN_LAG_ENTRIES.to_owned(),
s.primary_head_seq.saturating_sub(s.applied_seq).to_string(),
);
}
if let (Some(busy), Some(total)) = (cpu_busy, cpu_total) {
h.insert(TWIN_CPU_BUSY.to_owned(), busy.to_string());
h.insert(TWIN_CPU_TOTAL.to_owned(), total.to_string());
}
}
/// The twin's health words ([`twin_words`]; monetize-server `estate_order::twin::TwinReading::of_reports`).
pub const TWIN_STATE: &str = "twin_state";
pub const TWIN_LAG_ENTRIES: &str = "lag_entries";
pub const TWIN_CPU_BUSY: &str = "cpu_millicores_busy";
pub const TWIN_CPU_TOTAL: &str = "cpu_millicores_total";
/// The role this box pulls as.
pub fn role_of(stores: &ServerStores, fleet: &FleetConfig) -> Option<String> {
nordisk_pairing::elect::pull_role(
&fleet.role,
crate::twin::is_twin_root(stores.root()),
door_role_of(stores.root()).as_ref().map(|r| r.as_deref()),
)
}
/// The role the installer's door elected for this box ([`nordisk_pairing::elect::pull_role`]'s `door`).
fn door_role_of(root: &std::path::Path) -> Option<Option<String>> {
#[cfg(feature = "wg-client")]
{
crate::pairing::door_role(root)
}
#[cfg(not(feature = "wg-client"))]
{
let _ = root;
None
}
}
/// The channel to monetize's `Fleet`: dialled at [`FleetConfig::at`] when one is named, the
/// URL's host otherwise; for `https://` the certificate is verified under the URL's host,
/// against the CA file, else the fed CA, else the public roots.
#[cfg(feature = "fleet-pull")]
pub fn fleet_channel(fleet: &FleetConfig) -> Result<tonic::transport::Channel, String> {
let origin: tonic::codegen::http::Uri = fleet
.url
.parse()
.map_err(|e| format!("--fleet {}: {e}", fleet.url))?;
let name = origin.host().unwrap_or("").to_owned();
let dial = match fleet.at.as_deref() {
Some(at) => format!("{}://{at}", origin.scheme_str().unwrap_or("https")),
None => fleet.url.clone(),
};
let endpoint = tonic::transport::Channel::from_shared(dial.clone())
.map_err(|e| format!("--fleet {} at {dial}: {e}", fleet.url))?
.origin(origin)
.connect_timeout(Duration::from_secs(10))
.timeout(Duration::from_millis(fleet::LONG_POLL_MAX_MS + 10_000));
let endpoint = if fleet.url.starts_with("https://") {
gunnar_grpc::ensure_crypto_provider();
let tls = match (fleet.ca.as_ref(), fleet.ca_pem.as_deref()) {
(Some(ca), _) => {
let pem =
std::fs::read(ca).map_err(|e| format!("--fleet-ca {}: {e}", ca.display()))?;
tonic::transport::ClientTlsConfig::new()
.ca_certificate(tonic::transport::Certificate::from_pem(pem))
}
(None, Some(pem)) => tonic::transport::ClientTlsConfig::new()
.ca_certificate(tonic::transport::Certificate::from_pem(pem)),
// A fed door names no CA: the public roots (monetize.ignalina.dk).
(None, None) => tonic::transport::ClientTlsConfig::new().with_webpki_roots(),
};
endpoint
.tls_config(tls.domain_name(name))
.map_err(|e| format!("--fleet {}: {e}", fleet.url))?
} else {
let host = dial
.trim_start_matches("http://")
.split(['/', ':'])
.next()
.unwrap_or("");
let local = host == "localhost"
|| host
.trim_matches(['[', ']'])
.parse::<std::net::IpAddr>()
.map(|ip| ip.is_loopback())
.unwrap_or(false);
if !local {
return Err(format!("--fleet {} is plain http to another machine: orders are signed, but this box's reports would not be private; use https:// with --fleet-ca", fleet.url));
}
endpoint
};
Ok(endpoint.connect_lazy())
}
/// **How long after this box starts pulling a refusal as `unauthenticated` means "not
/// known YET"** (i54a run16): a box laid for an ATTACHED estate is let in by monetize only
/// once the estate's owner has opened its pin window (`POST /attach/installed`), which may
/// land after the box's first report. Inside this grace such a refusal is said at INFO and
/// asked again soon ([`NOT_YET_KNOWN_RETRY`]); after it, it is the WARN it always was. It
/// covers a whole install (monetize's attached-install window is as long).
pub const NOT_YET_KNOWN_GRACE: Duration = Duration::from_secs(45 * 60);
/// How soon a box asks again while monetize does not know it yet ([`NOT_YET_KNOWN_GRACE`]).
pub const NOT_YET_KNOWN_RETRY: Duration = Duration::from_secs(10);
/// Pure: **after a refused report or pull**, the pause before asking again and whether it is
/// said quietly (INFO, "not known yet") — `unauthenticated` inside [`NOT_YET_KNOWN_GRACE`]
/// since the channel started: [`NOT_YET_KNOWN_RETRY`] at most, quiet. Anything else: `pause`,
/// a WARN.
pub fn after_refusal(code: &str, since_start: Duration, pause: Duration) -> (Duration, bool) {
if code == "unauthenticated" && since_start < NOT_YET_KNOWN_GRACE {
(pause.min(NOT_YET_KNOWN_RETRY), true)
} else {
(pause, false)
}
}
/// **The pull channel**: report, pull, verify, execute, ack — until `stop` turns true.
/// `entitlements` is what a `restate_entitlements` order reaches, with or without a repair key.
///
/// The pump is monetize-estate-side's (A28 M12): its [`fleet::Puller`] signs, checks and keeps
/// what it took, its [`fleet::TonicLink`] is the wire, and gunnar's executor is its
/// [`fleet::GunnarBox`]. The puller's calls block, so each runs on a blocking thread
/// ([`on_the_pump`]); this loop keeps the role wait, the pauses and the words.
#[cfg(feature = "fleet-pull")]
pub async fn pull_loop(
stores: Arc<ServerStores>,
fleet: FleetConfig,
estate: Option<ServerEstate>,
entitlements: Entitlements,
own: String,
mut stop: tokio::sync::watch::Receiver<bool>,
) {
// ★ i54a run27 (2026-10-11): the twin was fed the pull channel BEFORE the door elected its
// role, reported as `main` under its own key, and the ONE box — which pins one key per
// role on its first report — never knew it. A box whose role is not elected yet waits.
let mut said = false;
let role = loop {
if let Some(role) = role_of(&stores, &fleet) {
break role;
}
if !said {
tracing::info!(target: TARGET, "the pull channel waits for the installer's door to elect this box's role: it reports to monetize only as what it is");
said = true;
}
tokio::select! {
_ = tokio::time::sleep(ROLE_WAIT_EVERY) => {}
_ = stop.changed() => {}
}
if *stop.borrow() {
return;
}
};
let me = fleet::Addressee {
product: fleet::PRODUCT.to_owned(),
instance: fleet.instance.clone(),
role,
};
let entitlements = entitlements.pulled_as(&me.instance, &me.role, fleet.monetize_pubkey);
let rt = tokio::runtime::Handle::current();
let spent: Arc<dyn Fn() -> bool + Send + Sync> = {
let stop = stop.clone();
Arc::new(move || *stop.borrow())
};
match estate {
Some(e) => {
let boxed = fleet::GunnarBox::new(
ServerEstate {
entitlements: entitlements.clone(),
..e
},
rt,
)
.stopping(spent);
pull_rounds(stores, fleet, me, Arc::new(boxed), entitlements, own, stop).await
}
None => {
let boxed = fleet::GunnarBox::new(NoEstate(own.clone(), entitlements.clone()), rt)
.stopping(spent);
pull_rounds(stores, fleet, me, Arc::new(boxed), entitlements, own, stop).await
}
}
}
/// [`pull_loop`]'s rounds over one executor.
#[cfg(feature = "fleet-pull")]
async fn pull_rounds<E: Estate + Send + Sync + 'static>(
stores: Arc<ServerStores>,
fleet: FleetConfig,
me: fleet::Addressee,
boxed: Arc<fleet::GunnarBox<E>>,
entitlements: Entitlements,
own: String,
mut stop: tokio::sync::watch::Receiver<bool>,
) {
let puller = match fleet::Puller::at(stores.root(), me.clone(), env!("CARGO_PKG_VERSION")) {
Ok(p) => p,
Err(e) => {
return tracing::error!(target: TARGET, "the pull channel is off: this box's fleet key or state could not be read: {e}")
}
};
let channel = match fleet_channel(&fleet) {
Ok(c) => c,
Err(e) => return tracing::error!(target: TARGET, "the pull channel is off: {e}"),
};
let public: String = puller
.box_public()
.iter()
.map(|b| format!("{b:02x}"))
.collect();
tracing::info!(target: TARGET, url = %fleet.url, instance = %me.instance, role = %me.role, box_key = %public, "pulling repair orders from monetize; it never calls in");
let mut pump = Some(Pump {
puller,
link: fleet::TonicLink::new(channel, tokio::runtime::Handle::current()),
});
let started = std::time::Instant::now();
// RA3-6: panicked pump calls in a row; a call that answers resets it.
let mut panics = 0u32;
// After a panic: rebuild the puller from its kept state (the ALERT and the back-off are
// said where it panicked).
let rebuild = |pump: &mut Option<Pump>| {
if let Some(p) = pump.as_mut() {
match fleet::Puller::at(stores.root(), me.clone(), env!("CARGO_PKG_VERSION")) {
Ok(fresh) => p.puller = fresh,
Err(e) => {
tracing::error!(target: TARGET, "ALERT fleet.pump-rebuild the puller could not be rebuilt after the panic ({e}); the old one pulls again")
}
}
}
};
loop {
let now = || crate::twin::now_ms().max(0) as u64;
let mut pause = fleet.every;
let product = {
let e = entitlements.clone();
let at = now();
tokio::task::spawn_blocking(move || e.product(at))
.await
.unwrap_or(None)
};
let own_review = twin_review(&stores, &own, now() as i64);
let health = health(&stores, &own);
let at = now();
let reported = match on_the_pump(&mut pump, move |p: &mut Pump| {
p.puller
.report(&mut p.link, health, product, own_review, at)
})
.await
{
None => return,
Some(Pumped::Ran(r)) => r,
Some(Pumped::Panicked(why)) => {
panics += 1;
let pause = pump_backoff(panics);
tracing::error!(target: TARGET, "{}", pump_panicked_line("report", panics, &why, pause));
rebuild(&mut pump);
tokio::select! {
_ = tokio::time::sleep(pause) => {}
_ = stop.changed() => {}
}
if *stop.borrow() {
return;
}
continue;
}
};
match reported {
Ok(next) if next > 0 => {
pause = Duration::from_millis(next).clamp(
Duration::from_secs(1),
fleet.every.max(Duration::from_secs(1)),
)
}
Ok(_) => {}
Err(f) => {
let (p, quiet) = after_refusal(&f.code, started.elapsed(), pause);
pause = p;
if quiet {
tracing::info!(target: TARGET, "monetize does not know this box yet (the estate's owner opens its window as the install feeds it); asking again in {} s: {f}", pause.as_secs());
} else {
tracing::warn!(target: TARGET, "the report to monetize failed: {f}");
}
}
}
let key = fleet
.monetize_pubkey
.or_else(crate::feed::monetize_in_force);
let (b, at) = (Arc::clone(&boxed), now());
let pulled = match on_the_pump(&mut pump, move |p: &mut Pump| {
p.puller.pull(
&mut p.link,
b.as_ref(),
key.as_ref(),
fleet::LONG_POLL_MAX_MS,
at,
)
})
.await
{
None => return,
Some(Pumped::Ran(r)) => r,
Some(Pumped::Panicked(why)) => {
panics += 1;
let pause = pump_backoff(panics);
tracing::error!(target: TARGET, "{}", pump_panicked_line("pull", panics, &why, pause));
rebuild(&mut pump);
tokio::select! {
_ = tokio::time::sleep(pause) => {}
_ = stop.changed() => {}
}
if *stop.borrow() {
return;
}
continue;
}
};
let mut ran = false;
match pulled {
Ok(p) => {
panics = 0;
for a in &p.acks {
ran = true;
if a.ok {
tracing::info!(target: TARGET, seq = a.seq, "monetize's order ran: {}", a.detail);
} else {
tracing::warn!(target: TARGET, seq = a.seq, "monetize's order was refused: {}", a.detail);
}
}
for f in &p.forged {
tracing::error!(target: TARGET, "an order that is NOT monetize's was offered and ignored: {f}");
}
}
Err(f) if f.code == "no_key" => tracing::debug!(target: TARGET, "{}", f.message),
Err(f) => {
let (p, quiet) = after_refusal(&f.code, started.elapsed(), pause);
pause = pause.min(p);
if quiet {
tracing::info!(target: TARGET, "pulling orders: monetize does not know this box yet; asking again in {} s: {f}", pause.as_secs());
} else {
tracing::warn!(target: TARGET, "pulling orders failed: {f}");
}
}
}
if *stop.borrow() {
return;
}
if ran {
continue;
}
tokio::select! {
_ = tokio::time::sleep(pause) => {}
_ = stop.changed() => {}
}
if *stop.borrow() {
return;
}
}
}
/// The shared puller and its wire, handed to a blocking thread for each call.
#[cfg(feature = "fleet-pull")]
struct Pump {
puller: fleet::Puller,
link: fleet::TonicLink,
}
/// What one pump call came to ([`on_the_pump`]).
#[derive(Debug, PartialEq, Eq)]
pub enum Pumped<T> {
/// It answered.
Ran(T),
/// It panicked (RA3-6): the words of the panic. The pump is kept; the loop rebuilds its
/// puller from what it kept on disk, says an ALERT and pulls again after [`pump_backoff`].
Panicked(String),
}
/// **One blocking pump call** off the runtime's workers ([`fleet::TonicLink`] blocks on the
/// runtime, and so does a [`fleet::GunnarBox`] running an order). A panic in the call (one bad
/// order, or a panic inside the executor while it runs one) is caught on that thread
/// ([`Pumped::Panicked`]) and the pump handed back (RA3-6: it used to turn the pull channel
/// off until the next start, said only as an error). `None` only when the blocking thread
/// itself was lost (the runtime is stopping).
pub async fn on_the_pump<P: Send + 'static, T: Send + 'static>(
pump: &mut Option<P>,
call: impl FnOnce(&mut P) -> T + Send + 'static,
) -> Option<Pumped<T>> {
let mut p = pump.take()?;
match tokio::task::spawn_blocking(move || {
let out = std::panic::catch_unwind(std::panic::AssertUnwindSafe(|| call(&mut p)));
(p, out)
})
.await
{
Ok((p, out)) => {
*pump = Some(p);
Some(match out {
Ok(v) => Pumped::Ran(v),
Err(panic) => Pumped::Panicked(panic_words(panic.as_ref())),
})
}
Err(e) => {
tracing::error!(target: TARGET, "ALERT fleet.pump-lost the pull channel is off: its blocking thread was lost ({e}); the runtime is stopping");
None
}
}
}
/// The words of a caught panic.
fn panic_words(panic: &(dyn std::any::Any + Send)) -> String {
panic
.downcast_ref::<&str>()
.map(|s| s.to_string())
.or_else(|| panic.downcast_ref::<String>().cloned())
.unwrap_or_else(|| "a panic with no words".into())
}
/// The first pause after a pump call panicked ([`pump_backoff`]).
pub const PUMP_BACKOFF_FIRST: Duration = Duration::from_secs(2);
/// The longest pause after pump calls that keep panicking.
pub const PUMP_BACKOFF_MOST: Duration = Duration::from_secs(5 * 60);
/// Pure: **the pause after the `n`th panic in a row** (1-based): 2 s, doubling, at most
/// [`PUMP_BACKOFF_MOST`]. A call that answers resets the row.
pub fn pump_backoff(n: u32) -> Duration {
PUMP_BACKOFF_FIRST
.saturating_mul(1u32 << n.saturating_sub(1).min(16))
.min(PUMP_BACKOFF_MOST)
}
/// The ALERT line of a panicked pump call, by name (RA3-6).
pub fn pump_panicked_line(call: &str, n: u32, why: &str, pause: Duration) -> String {
format!("ALERT fleet.pump-panicked {call}: the pump call panicked ({why}); panic {n} in a row: the puller is rebuilt from its kept state and pulls again in {} s", pause.as_secs())
}
/// How often a box whose role is not elected yet looks again ([`nordisk_pairing::elect::pull_role`]).
pub const ROLE_WAIT_EVERY: Duration = Duration::from_secs(2);
/// A box with no repair key still pulls: every order it cannot run is acked as refused,
/// by name, so the guardian learns why instead of waiting.
#[cfg(feature = "fleet-pull")]
struct NoEstate(String, Entitlements);
#[cfg(feature = "fleet-pull")]
impl Estate for NoEstate {
type Twin = ProtoTwin;
type Main = ProtoMain;
type Fence = ProtoFence;
fn twin(&self, _: &str) -> Result<ProtoTwin, Fault> {
Err(Fault::new(
"unauthenticated",
"this box holds no repair key (--repair-key, or a twin key)",
))
}
fn main(&self, _: &str) -> Result<ProtoMain, Fault> {
Err(Fault::new(
"unauthenticated",
"this box holds no repair key (--repair-key, or a twin key)",
))
}
fn fence(&self, _: &str) -> Result<ProtoFence, Fault> {
Err(Fault::new(
"unauthenticated",
"this box holds no repair key (--repair-key, or a twin key)",
))
}
fn restate(&self, snapshot: &serde_json::Value) -> Result<String, Refusal> {
self.1.restate(snapshot)
}
fn entitle(&self, fact: &serde_json::Value) -> Result<String, Refusal> {
self.1.entitle(fact)
}
fn add_principal(&self, username: &str, kind: &str, signup: &str) -> Result<String, Refusal> {
self.1.add_principal(username, kind, signup)
}
fn add_users(
&self,
tenant: &str,
ask: &serde_json::Value,
order: &str,
) -> Result<String, Refusal> {
self.1.add_users(tenant, ask, order)
}
fn grow_room(&self, need_bytes: u64, reason: &str) -> Result<String, Refusal> {
self.1.grow_room(need_bytes, reason)
}
fn purge_tenant(&self, tenant: &str, reason: &str, job: &str) -> Result<String, Refusal> {
self.1.purge_tenant(tenant, reason, job)
}
async fn sign_in_again(
&self,
account: &str,
to: &str,
host: &str,
ttl_secs: u64,
) -> Result<String, Refusal> {
self.1.sign_in_again(account, to, host, ttl_secs).await
}
fn keep_mail_relay(&self, relay: &gunnar_repair::order::MailRelay) -> Result<String, Refusal> {
self.1.keep_mail_relay(relay)
}
fn keep_console_setting(&self, name: &str, value: &str) -> Result<String, Refusal> {
self.1.keep_console_setting(name, value)
}
fn grow_check(
&self,
go_ahead: &str,
data_gib: u64,
) -> Result<gunnar_repair::exec::GrowCheck, Refusal> {
self.1.grow_check(go_ahead, data_gib)
}
async fn twin_holds(&self, check: &gunnar_repair::exec::GrowCheck) -> Result<String, Refusal> {
self.1.twin_holds(check).await
}
fn grow_start(
&self,
check: &gunnar_repair::exec::GrowCheck,
nonce: &str,
) -> Result<gunnar_repair::exec::GrowState, Refusal> {
self.1.grow_start(check, nonce)
}
async fn grow_settle(&self, past_waiting: bool) -> gunnar_repair::exec::GrowState {
self.1.grow_settle(past_waiting).await
}
fn grow_look_again(&self, give_up: bool) -> Result<(), Refusal> {
self.1.grow_look_again(give_up)
}
fn own(&self) -> String {
self.0.clone()
}
fn now_unix_ms(&self) -> i64 {
crate::twin::now_ms()
}
}
#[cfg(test)]
mod pump_tests {
use super::*;
/// ★ RA3-6 RED: **a pump call that panics keeps the pump, and the next call runs** — after
/// M12 one panic in a pump call (a bad order, or the executor panicking while it runs one)
/// turned the pull channel off until the next start, said only as an error.
#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
async fn a_pump_that_panics_on_its_first_order_still_takes_the_second() {
let mut pump = Some(0u32);
let first = on_the_pump(&mut pump, |n: &mut u32| -> u32 {
*n += 1;
panic!("the first order is a bad one")
})
.await;
assert!(
first.is_some(),
"a panic is an answer, not the end of the channel"
);
assert!(pump.is_some(), "the pump is handed back");
let second = on_the_pump(&mut pump, |n: &mut u32| {
*n += 1;
*n
})
.await;
assert!(second.is_some(), "the second order is taken");
assert_eq!(
first,
Some(Pumped::Panicked("the first order is a bad one".into()))
);
assert_eq!(second, Some(Pumped::Ran(2)));
}
/// ★ RA3-6: **the back-off after panics in a row** doubles from 2 s and stops at 5 min, and
/// the ALERT names the call and the row.
#[test]
fn a_panicking_pump_backs_off_and_says_an_alert() {
assert_eq!(pump_backoff(1), Duration::from_secs(2));
assert_eq!(pump_backoff(2), Duration::from_secs(4));
assert_eq!(pump_backoff(4), Duration::from_secs(16));
assert_eq!(pump_backoff(40), PUMP_BACKOFF_MOST);
let line = pump_panicked_line("pull", 3, "boom", pump_backoff(3));
assert!(
line.starts_with("ALERT fleet.pump-panicked pull:")
&& line.contains("boom")
&& line.contains("in 8 s"),
"{line}"
);
}
}
#[cfg(test)]
mod tests {
use super::*;
use gunnar_grpc::{ControlError, FirstBoot};
/// ★ Main rebuild: auto-fix is the medium's flag OR the estate's fed opt-in; anything
/// but exactly `true` fed is off.
#[test]
fn auto_fix_is_the_flag_or_the_fed_opt_in() {
let off = RepairConfig::default();
assert!(!auto_with(&off, None));
assert!(auto_with(
&RepairConfig {
auto: true,
..RepairConfig::default()
},
None
));
let feed = |v: &str| {
crate::feed::Fed::from_feed(&std::collections::BTreeMap::from([(
crate::feed::FEED_REPAIR_AUTO.to_string(),
v.to_string(),
)]))
};
assert!(auto_with(&off, Some(&feed("true"))));
for v in ["yes", "on", "TRUE", "1", ""] {
assert!(!auto_with(&off, Some(&feed(v))), "{v:?} is not the opt-in");
}
}
/// The awaiting-main kick: only with auto-fix, one at a time, and not again within a minute.
#[test]
fn the_awaiting_kick_is_due_once_a_minute_and_one_at_a_time() {
assert!(kick_due(true, false, None));
assert!(!kick_due(false, false, None), "auto-fix off");
assert!(!kick_due(true, true, None), "one running");
assert!(!kick_due(true, false, Some(Duration::from_secs(5))));
assert!(kick_due(true, false, Some(KICK_AGAIN_AFTER)));
}
/// The repair key: the medium's file, else the FED key, else the twin's key — and a fed
/// value that is no key is refused by name.
#[test]
fn the_repair_key_is_the_flag_then_the_fed_key_then_the_twin_key() {
let d = tempfile::tempdir().unwrap();
let none = RepairConfig::default();
let e = ServerEstate::with_fed(d.path(), &none, "o".into(), None)
.err()
.expect("no key anywhere");
assert!(e.contains(crate::feed::FEED_REPAIR_KEY), "{e}");
let pem = String::from_utf8(gunnar_core::mint_ed25519().unwrap()).unwrap();
let fed = crate::feed::Fed::from_feed(&std::collections::BTreeMap::from([(
crate::feed::FEED_REPAIR_KEY.to_string(),
pem.clone(),
)]));
let k = ServerEstate::with_fed(d.path(), &none, "o".into(), Some(&fed))
.expect("the fed key")
.key;
assert_eq!(
k.public_key().to_openssh().unwrap(),
PrivateKey::from_openssh(pem.as_bytes())
.unwrap()
.public_key()
.to_openssh()
.unwrap()
);
let other = String::from_utf8(gunnar_core::mint_ed25519().unwrap()).unwrap();
let path = d.path().join("flag.key");
std::fs::write(&path, &other).unwrap();
let k = ServerEstate::with_fed(
d.path(),
&RepairConfig {
key: Some(path),
..RepairConfig::default()
},
"o".into(),
Some(&fed),
)
.expect("the flag")
.key;
assert_eq!(
k.public_key().to_openssh().unwrap(),
PrivateKey::from_openssh(other.as_bytes())
.unwrap()
.public_key()
.to_openssh()
.unwrap(),
"the medium's flag wins"
);
let bad = crate::feed::Fed::from_feed(&std::collections::BTreeMap::from([(
crate::feed::FEED_REPAIR_KEY.to_string(),
"not a key".to_string(),
)]));
let e = ServerEstate::with_fed(d.path(), &none, "o".into(), Some(&bad))
.err()
.expect("refused");
assert!(e.contains("not an OpenSSH private key"), "{e}");
}
fn flag(url: &str, instance: &str) -> FleetConfig {
FleetConfig {
url: url.into(),
ca: Some(PathBuf::from("/etc/one-ca.pem")),
ca_pem: None,
at: None,
instance: instance.into(),
role: "main".into(),
monetize_pubkey: None,
every: Duration::from_secs(30),
}
}
fn fed(fleet: Option<&str>, instance: Option<&str>) -> crate::feed::Fed {
crate::feed::Fed {
fleet: fleet.map(Into::into),
fleet_instance: instance.map(Into::into),
..crate::feed::Fed::default()
}
}
/// ★ **AIRGAP B5: a build without `fleet-pull` reports to no fleet.** A named channel (the
/// flags) and a fed one are refused by name; nothing named is still no channel.
#[cfg(not(feature = "fleet-pull"))]
#[test]
fn a_build_without_fleet_pull_refuses_a_named_or_fed_channel_by_name() {
let every = Duration::from_secs(60);
assert!(resolve_fleet(None, None, every).unwrap().is_none());
let err = resolve_fleet(
Some(flag("https://one.mock.invalid:8443", "acme")),
None,
every,
)
.unwrap_err();
assert!(
err.contains("fleet-pull") && err.contains("one.mock.invalid"),
"{err}"
);
let err = resolve_fleet(
None,
Some(&fed(Some("https://one.mock.invalid:8443"), Some("acme"))),
every,
)
.unwrap_err();
assert!(err.contains("fleet-pull"), "{err}");
}
/// With `fleet-pull`, the channel the flags or the feed name, as before.
#[cfg(feature = "fleet-pull")]
#[test]
fn a_fleet_pull_build_pulls_where_it_is_told_as_before() {
let every = Duration::from_secs(60);
let f = resolve_fleet(
Some(flag("https://one.mock.invalid:8443", "acme")),
None,
every,
)
.unwrap()
.unwrap();
assert_eq!(f.url, "https://one.mock.invalid:8443");
}
/// ONE monetize (BROKEN: "gunnar-front must hand fleet/fleet_instance/monetize_pubkey to
/// main and twin"): a box with no `--fleet` pulls where the installer's feed says; a flag
/// and an equal feed agree; a flag and a different feed are refused by name.
#[test]
fn the_pull_channel_is_the_flags_or_the_feeds_and_a_disagreement_is_refused() {
let every = Duration::from_secs(60);
assert!(
resolve_fleet_named(None, None, every).unwrap().is_none(),
"nothing named: no pull channel"
);
assert!(resolve_fleet_named(None, Some(&fed(None, None)), every)
.unwrap()
.is_none());
let f = resolve_fleet_named(
None,
Some(&fed(
Some("https://monetize.ignalina.dk:8443"),
Some("acme"),
)),
every,
)
.unwrap()
.expect("the feed names it");
assert_eq!(
(
f.url.as_str(),
f.instance.as_str(),
f.role.as_str(),
f.ca.is_none(),
f.monetize_pubkey.is_none(),
f.every
),
(
"https://monetize.ignalina.dk:8443",
"acme",
"",
true,
true,
every
),
"role = what this box is; key = the one in force"
);
let f = resolve_fleet_named(
Some(flag("https://one:8443", "acme")),
Some(&fed(Some("https://one:8443"), Some("acme"))),
every,
)
.unwrap()
.unwrap();
assert_eq!(
(f.ca.is_some(), f.role.as_str(), f.every),
(true, "main", Duration::from_secs(30)),
"equal: the medium's, with its CA and role"
);
let f = resolve_fleet_named(Some(flag("https://one:8443", "acme")), None, every)
.unwrap()
.unwrap();
assert_eq!(f.url, "https://one:8443");
for other in [
fed(Some("https://two:8443"), Some("acme")),
fed(Some("https://one:8443"), Some("globex")),
] {
let e =
resolve_fleet_named(Some(flag("https://one:8443", "acme")), Some(&other), every)
.expect_err("two answers to where this box reports");
assert!(e.contains(crate::feed::FEED_MISMATCH), "{e}");
}
let e = resolve_fleet_named(None, Some(&fed(Some("https://one:8443"), None)), every)
.expect_err("a door without the estate");
assert!(e.contains("fleet_instance"), "{e}");
}
/// The fed door names no CA file: monetize's certificate is checked against the
/// public roots, never not at all.
#[cfg(feature = "fleet-pull")]
#[tokio::test]
async fn a_fed_https_door_without_a_ca_file_is_dialled_against_the_public_roots() {
let f = FleetConfig {
ca: None,
..flag("https://monetize.ignalina.dk:8443", "acme")
};
fleet_channel(&f).expect("public roots");
let plain = FleetConfig {
ca: None,
..flag("http://monetize.ignalina.dk:8080", "acme")
};
assert!(
fleet_channel(&plain).is_err(),
"plain http to another machine stays refused"
);
}
/// One TLS server on loopback for `name` (its own self-signed certificate): the
/// handshake's verdict and the name the client asked for, once. Returns the address,
/// the certificate (PEM) and the verdict.
#[cfg(feature = "fleet-pull")]
fn tls_door(
name: &str,
) -> (
std::net::SocketAddr,
String,
std::sync::mpsc::Receiver<Result<String, String>>,
) {
// A self-signed leaf, as the mock's ONE box serves (private-monetize-ops `bench_certificate`).
let cert = rcgen::generate_simple_self_signed(vec![name.to_owned()]).expect("cert");
let pem = cert.cert.pem();
let der = cert.cert.der().clone();
let key = rustls::pki_types::PrivateKeyDer::try_from(cert.signing_key.serialize_der())
.expect("key");
let cfg = rustls::ServerConfig::builder_with_provider(Arc::new(
rustls::crypto::ring::default_provider(),
))
.with_safe_default_protocol_versions()
.unwrap()
.with_no_client_auth()
.with_single_cert(vec![der], key)
.unwrap();
let listener = std::net::TcpListener::bind("127.0.0.1:0").unwrap();
let at = listener.local_addr().unwrap();
let (tx, rx) = std::sync::mpsc::channel();
std::thread::spawn(move || {
let Ok((mut tcp, _)) = listener.accept() else {
return;
};
tcp.set_read_timeout(Some(Duration::from_secs(10))).ok();
let mut conn = rustls::ServerConnection::new(Arc::new(cfg)).unwrap();
let said = loop {
if !conn.is_handshaking() {
break Ok(conn.server_name().unwrap_or("").to_owned());
}
if let Err(e) = conn.complete_io(&mut tcp) {
break Err(e.to_string());
}
};
let _ = tx.send(said);
});
(at, pem, rx)
}
/// ★ **A fed connect address and CA for the pull channel** (i54a run12: every box's
/// `fleet` = `https://monetize-test.ignalina.dk` answered "dns error" seven times a
/// minute — a mock guest resolves nothing, and the mock's ONE box has a bench
/// certificate). The installer may feed `fleet_at` (the address to dial) and
/// `fleet_ca` (the PEM the door's certificate is checked against): the box dials
/// THAT address, and TLS still verifies the URL's own name.
#[cfg(feature = "fleet-pull")]
#[tokio::test]
async fn a_fed_connect_address_and_ca_are_dialled_while_tls_verifies_the_urls_name() {
let (at, pem, said) = tls_door("one.test");
let feed: BTreeMap<String, String> = [
("fleet", "https://one.test"),
("fleet_instance", "acme"),
("offer.front.fleet_at", &at.to_string()),
("fleet_ca", &pem),
]
.iter()
.map(|(k, v)| (k.to_string(), v.to_string()))
.collect();
let f = resolve_fleet_named(
None,
Some(&crate::feed::Fed::from_feed(&feed)),
Duration::from_secs(60),
)
.unwrap()
.expect("the feed names a door");
assert_eq!(f.url, "https://one.test", "the URL stays the door's name");
let mut client =
fleet::pb::fleet_client::FleetClient::new(fleet_channel(&f).expect("a channel"));
let _ = tokio::time::timeout(
Duration::from_secs(10),
client.report(fleet::pb::FleetSigned::default()),
)
.await;
let said = said
.recv_timeout(Duration::from_secs(10))
.expect("the box dialled the fed address");
assert_eq!(
said,
Ok("one.test".to_owned()),
"the handshake completed against the fed CA, under the URL's name"
);
}
/// ★ ST-19 (STINKY 2026-10-11): **the mock's seam is honoured only as the mock feeds it.**
/// `fleet_at` is a bench forward (a private, loopback or link-local address literal); one
/// that names a public address or a name is refused by name, and so is a fed `fleet_ca`
/// without such a `fleet_at` — a release box never re-roots its trust on a feed's word.
#[test]
fn the_fed_fleet_seam_is_refused_unless_it_is_a_bench_address() {
let every = Duration::from_secs(60);
let feed = |pairs: &[(&str, &str)]| {
crate::feed::Fed::from_feed(
&pairs
.iter()
.map(|(k, v)| (k.to_string(), v.to_string()))
.collect(),
)
};
let base = [
("fleet", "https://monetize.ignalina.dk"),
("fleet_instance", "acme"),
];
for at in [
"10.0.2.2:40443",
"127.0.0.1:443",
"192.168.1.9:8443",
"172.20.0.4:443",
"169.254.1.1:443",
"[::1]:443",
"[fd00::2]:443",
"100.64.0.9:443",
] {
let f = resolve_fleet_named(
None,
Some(&feed(&[
base[0],
base[1],
("fleet_at", at),
("fleet_ca", "PEM"),
])),
every,
);
assert!(
f.as_ref()
.is_ok_and(|f| f.as_ref().is_some_and(|f| f.at.as_deref() == Some(at))),
"a bench forward {at} is dialled: {f:?}"
);
}
for at in [
"203.0.113.7:443",
"8.8.8.8:443",
"monetize.example:443",
"[2001:db8::1]:443",
"10.0.2.2",
] {
let e = resolve_fleet_named(
None,
Some(&feed(&[base[0], base[1], ("fleet_at", at)])),
every,
)
.expect_err(at);
assert!(e.contains("[fleet-seam-bench-only]"), "{at}: {e}");
}
let e = resolve_fleet_named(
None,
Some(&feed(&[base[0], base[1], ("fleet_ca", "PEM")])),
every,
)
.expect_err("a CA with no bench address");
assert!(e.contains("[fleet-seam-bench-only]"), "{e}");
let plain = resolve_fleet_named(None, Some(&feed(&base)), every)
.unwrap()
.expect("the public door");
assert_eq!(
(plain.at, plain.ca_pem),
(None, None),
"no seam: the name, under the public roots"
);
}
/// ★ **A box monetize does not know YET asks again soon, quietly** (i54a run16: main and
/// twin of an attached estate reported before its owner opened their pin window, were
/// refused `unauthenticated`, backed off a full `fleet.every`, and missed the window's
/// first minutes — WARN spam and ENT0's room never landed). Inside the grace: retry within
/// 10 s, said at INFO; after it, or for any other refusal: the pause as before, a WARN.
#[test]
fn a_box_not_known_yet_asks_again_soon_and_quietly_inside_the_grace() {
let every = Duration::from_secs(60);
assert_eq!(
after_refusal("unauthenticated", Duration::from_secs(30), every),
(NOT_YET_KNOWN_RETRY, true)
);
assert_eq!(
after_refusal(
"unauthenticated",
NOT_YET_KNOWN_GRACE - Duration::from_secs(1),
every
),
(NOT_YET_KNOWN_RETRY, true)
);
assert_eq!(
after_refusal(
"unauthenticated",
Duration::from_secs(1),
Duration::from_secs(3)
),
(Duration::from_secs(3), true),
"never longer than it would have paused"
);
assert_eq!(
after_refusal(
"unauthenticated",
NOT_YET_KNOWN_GRACE + Duration::from_secs(1),
every
),
(every, false),
"past the grace it is a WARN again"
);
assert_eq!(
after_refusal("unavailable", Duration::from_secs(30), every),
(every, false),
"a door that is down is not 'not known yet'"
);
assert!(
NOT_YET_KNOWN_GRACE >= Duration::from_secs(45 * 60),
"covers a whole install"
);
}
fn key_line() -> String {
let pem = gunnar_core::mint_ed25519().expect("mint");
PrivateKey::from_openssh(std::str::from_utf8(&pem).unwrap())
.unwrap()
.public_key()
.to_openssh()
.unwrap()
}
/// ★ R-7b (report §4.1): a main whose data set is lost, with a twin able to give it
/// back, HOLDS — its open bootstrap window takes no first admin (that would make a
/// new, empty estate and shut the window `Bootstrap.Restore` needs) — and says what
/// it awaits. The way out by hand: remove the marker.
#[test]
fn a_held_box_takes_no_first_admin_and_says_its_twin_until_the_marker_goes() {
let d = tempfile::tempdir().unwrap();
let stores = Arc::new(ServerStores::open(d.path()).unwrap());
let plane = crate::CoreControlPlane::new(Arc::clone(&stores));
assert_eq!(held(&stores), None, "nothing holds a box nobody held");
hold(&stores, "10.13.8.3:50051").expect("hold");
assert_eq!(held(&stores).as_deref(), Some("10.13.8.3:50051"));
assert_eq!(
plane.bootstrap_state().unwrap().recharge_from,
"10.13.8.3:50051",
"Bootstrap.Status says what it awaits"
);
assert_eq!(
health(&stores, "x").get("data").map(String::as_str),
Some("lost")
);
let err = plane
.enroll_first_identity("alice", &key_line())
.expect_err("a held box takes no first admin");
assert!(
matches!(err, ControlError::FailedPrecondition(_)),
"{err:?}"
);
let said = err.to_string();
for words in [gunnar_repair::DATA_LOST_LINE, "10.13.8.3:50051", HOLD_REL] {
assert!(said.contains(words), "`{words}` missing from: {said}");
}
assert!(
stores.auth().bootstrap_window_open().unwrap(),
"the window stays open for the recharge"
);
std::fs::remove_file(d.path().join(HOLD_REL)).expect("the admin removes the marker");
assert_eq!(held(&stores), None);
assert_eq!(plane.bootstrap_state().unwrap().recharge_from, "");
plane
.enroll_first_identity("alice", &key_line())
.expect("the box enrols once the marker is gone");
}
/// G12 (o-g11-split): monetize's twin autoscale reads the twin's lag off the newest
/// report's `twin_review` and its load off the twin box's health words. A twin says both
/// itself, on every heartbeat; a main says neither of its own.
#[test]
fn a_twin_reviews_itself_with_its_lag_and_says_its_cores() {
let d = tempfile::tempdir().unwrap();
std::fs::write(d.path().join(crate::twin::TWIN_CONFIG), b"{}").unwrap();
let stores = ServerStores::open(d.path()).unwrap();
let idle: gunnar_repair::TwinReview = serde_json::from_str(
&twin_review(&stores, "10.0.0.2:50051", 1_000)
.expect("a twin reviews itself before its first poll"),
)
.unwrap();
assert_eq!(idle.state, "idle", "nothing polled yet: said, not guessed");
let status = crate::twin::TwinStatus {
applied_seq: 5,
primary_head_seq: 12,
state: crate::twin::State::Following,
why: String::new(),
last_answer_unix_ms: 1_500,
last_level_unix_ms: 0,
};
crate::twin::set_status(stores.root(), status);
let r: gunnar_repair::TwinReview =
serde_json::from_str(&twin_review(&stores, "10.0.0.2:50051", 2_000).expect("review"))
.unwrap();
assert_eq!(
(r.applied_seq, r.primary_head_seq, r.state.as_str()),
(5, 12, "following"),
"{r:?}"
);
assert_eq!(r.twin_addr, "10.0.0.2:50051");
assert_eq!(r.reviewed_unix_ms, 2_000);
assert_eq!(
r.would_lose.unreplicated, 7,
"the lag, as monetize reads it"
);
let h = health(&stores, "10.0.0.2:50051");
let word = |k: &str| {
h.get(k)
.unwrap_or_else(|| panic!("`{k}` missing from {h:?}"))
.parse::<u64>()
.unwrap_or_else(|e| panic!("`{k}` is not a u64: {e}"))
};
assert!(word("cpu_millicores_total") >= 1_000, "{h:?}");
let _ = word("cpu_millicores_busy");
let m = tempfile::tempdir().unwrap();
let main = ServerStores::open(m.path()).unwrap();
assert_eq!(twin_review(&main, "10.0.0.1:50051", 2_000), None);
assert!(!health(&main, "10.0.0.1:50051").contains_key("cpu_millicores_busy"));
}
/// The hold lifts by itself once the box holds a data set again (a recharge
/// finished): the marker goes, nothing is held.
#[test]
fn the_hold_lifts_once_the_box_holds_data_again() {
let d = tempfile::tempdir().unwrap();
let stores = ServerStores::open(d.path()).unwrap();
hold(&stores, "t:1").expect("hold");
assert!(held(&stores).is_some());
stores
.catalog()
.create_namespace("alice")
.expect("namespace");
stores
.create_repo(
gunnar_core::NewRepo::new(gunnar_core::RepoRef::new("alice", "one").unwrap()),
false,
)
.expect("repo");
assert!(!data_dead(&stores));
assert_eq!(held(&stores), None, "a box with data is held by nothing");
assert!(!d.path().join(HOLD_REL).exists(), "and the marker is gone");
}
/// G16-d glue: a box whose pull channel has no grow machine refuses a paid step by
/// name; a main with no twin address does not grow before a twin it cannot see.
#[tokio::test]
async fn a_paid_step_needs_this_boxs_grow_machine_and_a_main_needs_its_twins_address() {
let r = Entitlements::default().grow_check("{}", 120).unwrap_err();
assert!(
r.code == RefusalCode::BadOrder && r.message.contains("no grow machine"),
"{r}"
);
let paid = crate::paid_growth::PaidGrowth::new(crate::grow::GrowMachine::new(Arc::new(
crate::grow::world::NoEngine,
)));
let e = Entitlements::default()
.with_grow(paid)
.with_twin(Some(" ".into()));
let c = gunnar_repair::exec::GrowCheck {
twin: false,
data_gib: 120,
reference: "estate/acme/change-2".into(),
carried: false,
};
let r = e.twin_holds(&c).await.unwrap_err();
assert!(
r.code == RefusalCode::NoTwin
&& r.message.contains("--repair-twin")
&& r.message.contains("nothing moved"),
"{r}"
);
assert!(
r.message.contains("peer list"),
"the refusal names the door's peer list too: {r}"
);
}
/// ★ **ASK-twin** (resilience-gunnar 21:10): a sold main fed neither `--repair-twin` nor
/// `recharge_from` asks the twin the door named it — not `no_twin`.
#[cfg(feature = "wg-client")]
#[tokio::test]
async fn a_main_asks_the_twin_the_door_named_when_nothing_else_names_one() {
use nordisk_pairing::join::{Assigned, JoinReply, Peer};
use nordisk_pairing::Role;
let d = tempfile::tempdir().unwrap();
let stores = Arc::new(ServerStores::open(d.path()).unwrap());
let lo = std::net::Ipv4Addr::LOCALHOST;
let twin = Peer {
name: "twin".into(),
role: Role::Twin,
tunnel_pubkey: String::new(),
overlay: lo,
endpoint: None,
utility: Some(lo),
};
let reply = JoinReply {
estate: "gunnar.rs".into(),
me: Assigned {
name: "main".into(),
overlay: nordisk_pairing::OVERLAY_BOX_A,
},
cert_pem: None,
role: Some(Role::Primary),
primary_overlay: None,
primary_utility: None,
peers: vec![twin],
feed: BTreeMap::new(),
};
crate::fed_at_rest::write(
d.path(),
crate::pairing::JOINED_REL,
&reply.render().unwrap(),
)
.unwrap();
let paid = crate::paid_growth::PaidGrowth::new(crate::grow::GrowMachine::new(Arc::new(
crate::grow::world::NoEngine,
)));
let e = Entitlements::of(stores).with_grow(paid).with_twin(None);
let c = gunnar_repair::exec::GrowCheck {
twin: false,
data_gib: 120,
reference: "estate/acme/change-2".into(),
carried: false,
};
let said = match e.twin_holds(&c).await {
Ok(said) => said,
Err(r) => {
assert_ne!(r.code, RefusalCode::NoTwin, "the door named a twin: {r}");
r.message
}
};
assert!(
said.contains("127.0.0.1:50051"),
"the twin the door named is the one asked: {said}"
);
}
/// The pull channel's estate, role and key reach the step's verification: a twin
/// checks as the twin (its disk holds main's whole set).
#[cfg(feature = "metered")]
#[test]
fn the_pull_channels_estate_role_and_key_reach_the_paid_steps_check() {
use crate::paid_growth::tests::signed::{shop_pub, step, SHOP};
let paid = crate::paid_growth::PaidGrowth::new(crate::grow::GrowMachine::new(Arc::new(
crate::grow::world::NoEngine,
)));
let go = step(SHOP, "acme", 120, crate::twin::now_ms() + 60_000);
let e = Entitlements::default().with_grow(paid.clone()).pulled_as(
"acme",
"twin",
Some(shop_pub()),
);
let c = e.grow_check(&go, 480).expect("the twin's own size");
assert!(c.twin && !c.carried && c.data_gib == 480, "{c:?}");
let r = Entitlements::default()
.with_grow(paid)
.pulled_as("globex", "main", Some(shop_pub()))
.grow_check(&go, 120)
.unwrap_err();
assert!(
r.code == RefusalCode::BadOrder && r.message.contains("globex"),
"{r}"
);
}
}
#[cfg(test)]
mod twin_report_tests {
use super::*;
/// ★ F4 x G12: **a twin REPORTS its lag and load** — what the one monetize used to dial the
/// front's `/twin/health` door for (a dial-in, gone under the 5.4 pull rule).
#[test]
fn a_twin_reports_its_lag_and_load_for_the_pull() {
let s = crate::twin::TwinStatus {
applied_seq: 60,
primary_head_seq: 100,
state: crate::twin::State::Following,
why: String::new(),
last_answer_unix_ms: 1,
last_level_unix_ms: 0,
};
let mut h = BTreeMap::new();
twin_words(&mut h, Some(&s), Some(420), Some(2000));
assert_eq!(h["primary"], "live");
assert_eq!(h[TWIN_STATE], "following");
assert_eq!(h[TWIN_LAG_ENTRIES], "40");
assert_eq!(
(h[TWIN_CPU_BUSY].as_str(), h[TWIN_CPU_TOTAL].as_str()),
("420", "2000")
);
let mut none = BTreeMap::new();
twin_words(&mut none, None, None, Some(2000));
assert_eq!(none["primary"], "unknown");
assert!(
!none.contains_key(TWIN_LAG_ENTRIES) && !none.contains_key(TWIN_CPU_BUSY),
"no poll yet, no reading"
);
}
}